diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
new file mode 100644
index 0000000..57388f5
--- /dev/null
+++ b/.github/workflows/ci.yml
@@ -0,0 +1,34 @@
+name: CI
+
+on:
+ push:
+ branches: [main, develop]
+ pull_request:
+ branches: [main, develop]
+
+jobs:
+ test:
+ name: Test (Python ${{ matrix.python-version }})
+ runs-on: macos-latest
+ strategy:
+ fail-fast: false
+ matrix:
+ python-version: ["3.10", "3.11", "3.12"]
+
+ steps:
+ - uses: actions/checkout@v4
+
+ - name: Set up Python ${{ matrix.python-version }}
+ uses: actions/setup-python@v5
+ with:
+ python-version: ${{ matrix.python-version }}
+ cache: pip
+
+ - name: Install dependencies
+ run: pip install -e ".[dev]"
+
+ - name: Lint with ruff
+ run: ruff check .
+
+ - name: Run tests
+ run: pytest --cov=jigai --cov-report=term-missing -q
diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml
new file mode 100644
index 0000000..651a701
--- /dev/null
+++ b/.github/workflows/publish.yml
@@ -0,0 +1,51 @@
+name: Publish to PyPI
+
+on:
+ release:
+ types: [published]
+
+permissions:
+ contents: read
+
+jobs:
+ build:
+ name: Build distribution
+ runs-on: ubuntu-latest
+
+ steps:
+ - uses: actions/checkout@v4
+
+ - name: Set up Python
+ uses: actions/setup-python@v5
+ with:
+ python-version: "3.12"
+
+ - name: Install build tools
+ run: pip install build
+
+ - name: Build wheel and source distribution
+ run: python -m build
+
+ - name: Upload artifacts
+ uses: actions/upload-artifact@v4
+ with:
+ name: dist
+ path: dist/
+
+ publish:
+ name: Publish to PyPI
+ needs: build
+ runs-on: ubuntu-latest
+ environment: release
+ permissions:
+ id-token: write # Required for Trusted Publishing
+
+ steps:
+ - name: Download artifacts
+ uses: actions/download-artifact@v4
+ with:
+ name: dist
+ path: dist/
+
+ - name: Publish to PyPI
+ uses: pypa/gh-action-pypi-publish@release/v1
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..f316e47
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,34 @@
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.egg-info/
+dist/
+build/
+*.egg
+.eggs/
+
+# Virtual environments
+.venv/
+venv/
+env/
+
+# IDE
+.idea/
+.vscode/
+*.swp
+*.swo
+*~
+
+# Testing
+.pytest_cache/
+.coverage
+htmlcov/
+.mypy_cache/
+
+# OS
+.DS_Store
+Thumbs.db
+
+# JigAi runtime
+.jigai/
diff --git a/README.md b/README.md
new file mode 100644
index 0000000..a7b6293
--- /dev/null
+++ b/README.md
@@ -0,0 +1,477 @@
+# JigAi (জিগাই)
+
+
+ জিগাই — Bangla for "asking" — Know when your AI agent is waiting for you.
+
+
+
+ 
+ 
+ 
+ 
+ 
+
+
+---
+
+A **tool-agnostic** terminal notification system that watches AI coding agents — Claude Code, Codex, Gemini CLI, Aider, or any custom tool — and notifies you via **macOS notifications** and optionally your **phone over LAN** the moment they go idle and need your input.
+
+**No hooks. No per-tool config. Just wrap your command and go.**
+
+```bash
+pip install jigai
+jigai watch claude
+```
+
+---
+
+## The Problem
+
+You kick off Claude Code on a big refactor, switch to your browser to research something, and come back 20 minutes later to find it asked a clarifying question in the first 30 seconds. You just wasted 20 minutes.
+
+Every AI coding tool has this problem. None of them have a universal solution. Hook-based approaches require per-tool configuration that breaks across versions. Cloud notification services require accounts, subscriptions, or trusting a third party with your terminal output.
+
+**JigAi** fixes this with a single transparent PTY proxy that watches any terminal output, detects idle patterns, and notifies you — locally, privately, instantly.
+
+---
+
+## How It Works
+
+JigAi wraps your AI tool in a **PTY (pseudo-terminal) proxy**. Your tool runs exactly as normal — same colors, same interactivity, same behavior. JigAi intercepts the output stream silently:
+
+```
+You type: jigai watch claude
+ │
+ ▼
+ ┌──────────────────┐
+ │ JigAi Watcher │ ← sits here transparently
+ │ (PTY proxy) │
+ └────────┬─────────┘
+ │ passes all I/O through unchanged
+ ▼
+ ┌──────────────────┐
+ │ Claude Code │ ← behaves exactly as if launched directly
+ └──────────────────┘
+```
+
+When idle is detected (via pattern match or timeout), JigAi fires:
+1. A **macOS notification** with the last meaningful output line
+2. A **WebSocket push** to the JigAi server (if running)
+3. Your phone receives the notification via the LAN server *(mobile app coming in v0.2)*
+
+---
+
+## Quick Start
+
+### 1. Install
+
+```bash
+pip install jigai
+```
+
+For richer notifications (banner popups instead of silent NC delivery):
+
+```bash
+brew install terminal-notifier
+```
+
+> See [Notification Setup](#notification-setup) for macOS configuration steps.
+
+### 2. Watch a tool
+
+```bash
+# Claude Code
+jigai watch claude
+
+# OpenAI Codex CLI
+jigai watch codex
+
+# Gemini CLI
+jigai watch gemini
+
+# Aider
+jigai watch aider
+
+# Any arbitrary command
+jigai watch -- python my_agent.py
+
+# Override tool detection (for custom prompts)
+jigai watch --tool my_agent -- python agent.py
+```
+
+### 3. (Optional) Start the server for LAN mobile push
+
+```bash
+# Terminal 1 — keep the server running
+jigai server start
+
+# Terminal 2 — watch your tool as normal
+jigai watch claude
+```
+
+That's it. When Claude Code goes idle, you get notified.
+
+---
+
+## Supported Tools
+
+| Tool | Detection Method | Status |
+|------|-----------------|--------|
+| Claude Code | Pattern + timeout | Built-in |
+| OpenAI Codex CLI | Pattern + timeout | Built-in |
+| Gemini CLI | Pattern + timeout | Built-in |
+| Aider | Pattern + timeout | Built-in |
+| OpenCode | Pattern + timeout | Built-in |
+| Any custom tool | User-defined regex | Via `~/.jigai/patterns.yaml` |
+
+JigAi also has a **timeout fallback**: if no output is received for `timeout_seconds` (default: 30s), it fires regardless of pattern matching. This means it works with any tool, even ones not in the list above.
+
+---
+
+## Notification Setup
+
+JigAi uses two delivery mechanisms. Both are attempted in order:
+
+### 1. `terminal-notifier` (Recommended — banner popups)
+
+Install via Homebrew:
+
+```bash
+brew install terminal-notifier
+```
+
+Then configure macOS to show banners:
+
+1. Open **System Settings → Notifications → terminal-notifier**
+2. Enable **Allow Notifications**
+3. Check **Desktop** (required for banner popups)
+4. Set **Alert Style** to **Persistent** (stays until dismissed) or **Temporary** (auto-dismisses)
+5. Enable **Play sound for notification**
+
+### 2. `osascript` (Fallback — no installation required)
+
+If `terminal-notifier` is not installed, JigAi falls back to macOS's built-in `osascript`. Notifications will appear in **Notification Center** but may not show as banner popups, depending on your macOS version and Script Editor's notification settings.
+
+> **Recommendation:** Install `terminal-notifier` for the best experience.
+
+---
+
+## Known Issues and Caveats
+
+Read this section before filing a bug report — most common issues are documented here.
+
+### Focus Mode blocks banner notifications
+
+**Symptom:** Notifications appear in Notification Center but the banner popup never shows, even with correct settings.
+
+**Cause:** macOS Focus Mode (Do Not Disturb, Work, Personal, etc.) silences notification banners from apps not explicitly allowed.
+
+**Fix:**
+1. Open **System Settings → Focus**
+2. Select your active Focus profile
+3. Under **Allowed Notifications → Apps**, add **terminal-notifier**
+
+This is a one-time setup per Focus profile. Once allowed, banners will appear even when Focus is active.
+
+---
+
+### `terminal-notifier` on macOS Sequoia (15.x)
+
+**Symptom:** Notifications appear in Notification Center but never pop as banners, or `terminal-notifier` produces no output at all.
+
+**Cause:** `terminal-notifier` has known compatibility issues on macOS Sequoia 15, particularly on M-series chips ([issue #312](https://github.com/julienXX/terminal-notifier/issues/312)). The binary is not updated for the newer UserNotifications framework in Sequoia.
+
+**Workarounds:**
+- Try resetting the notification registration:
+ ```bash
+ /System/Library/Frameworks/CoreServices.framework/Frameworks/LaunchServices.framework/Support/lsregister -kill -r -domain local -domain system -domain user
+ ```
+ Then re-open System Settings → Notifications → terminal-notifier and re-enable.
+- If still broken, notifications will still land in Notification Center via the `osascript` fallback — you just won't get the popup banner.
+- A native Swift notification backend is planned for v0.2 that will resolve this permanently.
+
+---
+
+### Display mirroring / screen sharing
+
+**Symptom:** Notifications don't show on the desktop while screen sharing or mirroring.
+
+**Fix:** System Settings → Notifications → enable **"Allow notifications when mirroring or sharing the display"**
+
+---
+
+### Notifications fire while you're at the terminal
+
+By default, JigAi will fire a notification even if you're actively looking at the terminal. This is intentional — the sound alone can be useful as a cue.
+
+To suppress notifications when a terminal window is focused:
+
+```yaml
+# ~/.jigai/config.yaml
+notifications:
+ only_when_away: true
+```
+
+JigAi checks the frontmost macOS application. If it's Terminal, iTerm2, Warp, Ghostty, Alacritty, Kitty, or similar, the notification is skipped. Supported terminal app names are: `terminal`, `iterm2`, `warp`, `hyper`, `alacritty`, `kitty`, `ghostty`, `tabby`, `rio`.
+
+---
+
+### Idle fires too early or too often
+
+**Cause:** Pattern matching may trigger on output lines that superficially resemble idle prompts, especially for tools with rich TUI output.
+
+**Fix:** Adjust the cooldown (minimum gap between notifications) and timeout:
+
+```yaml
+# ~/.jigai/config.yaml
+detection:
+ timeout_seconds: 45 # how long to wait before timeout-based idle
+ cooldown_seconds: 10 # minimum seconds between notifications
+```
+
+Or test which lines trigger detection:
+
+```bash
+jigai config test "some terminal output line"
+```
+
+---
+
+### Pattern detection for Claude Code
+
+Claude Code renders a rich TUI with box-drawing characters and animated spinners. JigAi strips all ANSI codes and decorative Unicode before matching, and the idle prompt (`>`) is the primary pattern matched.
+
+If you find detection is unreliable, set a longer timeout:
+
+```bash
+jigai watch --timeout 60 claude
+```
+
+---
+
+### iOS / Android mobile notifications (LAN)
+
+The mobile app is planned for **v0.2**. Currently, `jigai server start` runs a WebSocket server that a future React Native app will connect to.
+
+iOS note: iOS aggressively terminates background WebSocket connections. When the mobile app is built, foreground/background behavior will be documented. Users on iOS may want to keep the app in the foreground or use a notification relay for reliable background delivery.
+
+---
+
+## Configuration
+
+### Config file: `~/.jigai/config.yaml`
+
+Initialize with defaults:
+
+```bash
+jigai config init
+```
+
+Full reference:
+
+```yaml
+server:
+ port: 9384 # LAN server port
+ bind: "0.0.0.0" # Bind address
+
+notifications:
+ macos: true # Enable macOS notifications
+ only_when_away: false # Skip if a terminal is the frontmost app
+ sound: "Ping" # macOS sound name (Ping, Basso, Funk, etc.)
+ group_by_session: true # Group notifications per session
+ show_last_output: true # Include last output in notification body
+ output_lines: 3 # Lines of output to include
+ redact_patterns: # Auto-redact sensitive info from notifications
+ - '(?i)(token|password|secret|key|api_key)=\S+'
+
+detection:
+ timeout_seconds: 30 # Timeout-based idle threshold
+ cooldown_seconds: 5 # Minimum gap between notifications
+```
+
+### Custom patterns: `~/.jigai/patterns.yaml`
+
+Add your own tools or override built-in patterns:
+
+```yaml
+custom_tools:
+ my_agent:
+ name: "My Custom Agent"
+ idle_patterns:
+ - 'READY>'
+ - 'awaiting instruction'
+ - '(?i)what would you like'
+
+overrides:
+ timeout_seconds: 45 # Override global timeout
+```
+
+---
+
+## Commands
+
+```bash
+# Watch commands
+jigai watch # Wrap a command, notify on idle
+jigai watch --tool # Override tool auto-detection
+jigai watch --timeout 60 # Override idle timeout
+jigai watch --no-notify # Disable macOS notifications
+jigai watch --no-server # Don't push events to the server
+
+# Server (for mobile / LAN notifications)
+jigai server start # Start server on default port 9384
+jigai server start --port 8080 # Custom port
+jigai server status # Check if server is running
+
+# Configuration
+jigai config init # Create default ~/.jigai/config.yaml
+jigai config show # Dump current configuration as JSON
+jigai config test "" # Test if a line matches any pattern
+
+# Info
+jigai patterns # Show all loaded patterns and timeouts
+jigai sessions # List active sessions (requires server)
+jigai --version # Print version
+```
+
+---
+
+## Architecture
+
+```
+┌─────────────────────────────────────────────────────────┐
+│ jigai watch claude │
+├─────────────────────────────────────────────────────────┤
+│ │
+│ stdin ──▶ PTY Proxy ──▶ Claude Code │
+│ (pty_proxy) (child proc) │
+│ │ │
+│ stdout ◀─── │ ────────────────────────────────────── │
+│ │ │
+│ ▼ │
+│ Idle Detector │
+│ (detector.py) │
+│ • ANSI strip │
+│ • Pattern match ──▶ match found ──┐ │
+│ • Timeout check ──▶ N seconds ────┤ │
+│ ▼ │
+│ Notification │
+│ ┌──────────────────┐ │
+│ │ macOS banner │ │
+│ │ (terminal-notif) │ │
+│ │ Server push │ │
+│ │ (WebSocket/HTTP) │ │
+│ └──────────────────┘ │
+└─────────────────────────────────────────────────────────┘
+
+┌──────────────────────────────────┐
+│ jigai server start │
+│ FastAPI + WebSocket │
+│ mDNS/Bonjour broadcast │──▶ Mobile App (v0.2)
+│ REST API for session tracking │ (React Native, LAN)
+└──────────────────────────────────┘
+```
+
+---
+
+## Development
+
+```bash
+git clone https://github.com/nafistiham/jigai.git
+cd jigai
+pip install -e ".[dev]"
+
+# Run tests
+pytest
+
+# Lint
+ruff check .
+
+# Type check
+mypy jigai/
+```
+
+### Project Structure
+
+```
+jigai/
+├── jigai/
+│ ├── cli.py # Typer CLI entry point
+│ ├── config.py # Config management (Pydantic + YAML)
+│ ├── models.py # Session and IdleEvent data models
+│ ├── notifier/
+│ │ └── macos.py # macOS notifications (osascript + terminal-notifier)
+│ ├── server/
+│ │ ├── app.py # FastAPI REST + WebSocket server
+│ │ ├── client.py # HTTP client (watcher → server)
+│ │ ├── discovery.py # mDNS/Bonjour service broadcasting
+│ │ └── ws_manager.py # WebSocket connection manager
+│ └── watcher/
+│ ├── detector.py # Idle detection engine
+│ ├── patterns.py # Pattern registry and loader
+│ ├── pty_proxy.py # Transparent PTY proxy
+│ └── watcher.py # Session orchestrator
+├── patterns/
+│ └── defaults.yaml # Built-in tool patterns
+└── tests/
+ ├── test_config.py
+ ├── test_detector.py
+ ├── test_models.py
+ └── test_patterns.py
+```
+
+---
+
+## Publishing to PyPI
+
+> For maintainers.
+
+1. Bump `version` in `pyproject.toml`
+2. Create a GitHub Release with tag `vX.Y.Z`
+3. The [CI workflow](.github/workflows/publish.yml) auto-publishes to PyPI via Trusted Publishing
+
+For the first publish, set up Trusted Publishing on PyPI:
+1. Go to pypi.org → your project → **Publishing**
+2. Add a trusted publisher: owner `nafistiham`, repo `jigai`, workflow `publish.yml`, environment `release`
+
+---
+
+## Roadmap
+
+- [x] v0.1 — CLI + PTY proxy + idle detection + macOS notifications + LAN server
+- [ ] v0.2 — React Native mobile app (iOS + Android), daemon mode
+- [ ] v0.3 — Homebrew formula, Linux support (libnotify)
+- [ ] v0.4 — Native Swift notification backend (Sequoia fix), web dashboard
+- [ ] Future — Slack/Discord webhooks, Windows toast notifications
+
+---
+
+## Contributing
+
+Contributions are welcome. Please open an issue before submitting a PR for non-trivial changes so we can discuss the approach.
+
+```bash
+# Fork, clone, create a branch
+git checkout -b feat/your-feature
+
+# Make changes, add tests
+pytest
+
+# Lint
+ruff check .
+
+# Open a PR against develop
+```
+
+---
+
+## License
+
+MIT — see [LICENSE](LICENSE).
+
+---
+
+
+ Built with frustration at missed Claude Code prompts.
+ github.com/nafistiham/jigai
+
diff --git a/jigai/__init__.py b/jigai/__init__.py
new file mode 100644
index 0000000..d463ca0
--- /dev/null
+++ b/jigai/__init__.py
@@ -0,0 +1,3 @@
+"""JigAi (জিগাই) — Tool-agnostic terminal notification system for AI coding agents."""
+
+__version__ = "0.1.0"
diff --git a/jigai/cli.py b/jigai/cli.py
new file mode 100644
index 0000000..cf2a0bb
--- /dev/null
+++ b/jigai/cli.py
@@ -0,0 +1,350 @@
+"""JigAi CLI — command-line interface."""
+
+from __future__ import annotations
+
+import os
+import sys
+from typing import Optional
+
+import typer
+from rich.console import Console
+from rich.table import Table
+
+from jigai import __version__
+
+app = typer.Typer(
+ name="jigai",
+ help="জিগাই — Tool-agnostic terminal notification system for AI coding agents.",
+ no_args_is_help=True,
+ add_completion=False,
+)
+
+console = Console()
+
+
+def version_callback(value: bool):
+ if value:
+ console.print(f"[bold cyan]JigAi[/bold cyan] (জিগাই) v{__version__}")
+ raise typer.Exit()
+
+
+@app.callback()
+def main(
+ version: bool = typer.Option(
+ False,
+ "--version",
+ "-v",
+ help="Show version and exit.",
+ callback=version_callback,
+ is_eager=True,
+ ),
+):
+ """জিগাই — Know when your AI agent is waiting for you."""
+ pass
+
+
+# ── Watch Command ───────────────────────────────────────────
+
+
+@app.command()
+def watch(
+ command: list[str] = typer.Argument(
+ ...,
+ help="Command to watch (e.g., 'claude', 'codex', 'python agent.py').",
+ ),
+ tool: Optional[str] = typer.Option(
+ None,
+ "--tool",
+ "-t",
+ help="Override tool detection (e.g., 'claude_code', 'codex', 'my_agent').",
+ ),
+ no_notify: bool = typer.Option(
+ False,
+ "--no-notify",
+ help="Disable macOS notifications.",
+ ),
+ no_server: bool = typer.Option(
+ False,
+ "--no-server",
+ help="Don't push events to JigAi server.",
+ ),
+ timeout: Optional[int] = typer.Option(
+ None,
+ "--timeout",
+ help="Override idle timeout in seconds.",
+ ),
+):
+ """
+ Watch a command and notify when it goes idle.
+
+ Usage:
+ jigai watch claude
+ jigai watch -- codex
+ jigai watch --tool my_agent -- python agent.py
+ """
+ from jigai.config import load_config
+ from jigai.models import IdleEvent
+ from jigai.server.client import ServerClient
+ from jigai.watcher.patterns import load_patterns
+ from jigai.watcher.watcher import Watcher
+
+ config = load_config()
+ registry = load_patterns()
+
+ if no_notify:
+ config.notifications.macos = False
+
+ if timeout is not None:
+ registry.timeout_seconds = timeout
+
+ # Set up server push if available
+ server_client = None
+ if not no_server:
+ client = ServerClient()
+ if client.is_server_running():
+ server_client = client
+ console.print(
+ "[dim] Server: Connected to JigAi server[/dim]", style="green"
+ )
+ else:
+ console.print(
+ "[dim] Server: Not running (use 'jigai server start' for mobile notifications)[/dim]"
+ )
+
+ def on_idle_event(event: IdleEvent) -> None:
+ """Push idle events to the server."""
+ if server_client:
+ server_client.push_event(event)
+
+ watcher = Watcher(
+ command=command,
+ tool_override=tool,
+ config=config,
+ registry=registry,
+ on_idle_event=on_idle_event if server_client else None,
+ )
+
+ # Register session with server
+ if server_client:
+ server_client.register_session(
+ session_id=watcher.session.session_id,
+ tool_name=watcher.session.tool_name,
+ command=command,
+ working_dir=watcher.session.working_dir,
+ )
+
+ # Run (blocks until command exits)
+ try:
+ exit_code = watcher.run()
+ finally:
+ # Unregister session
+ if server_client:
+ server_client.unregister_session(watcher.session.session_id)
+
+ raise typer.Exit(exit_code)
+
+
+# ── Server Commands ─────────────────────────────────────────
+
+
+server_app = typer.Typer(help="Manage the JigAi notification server.")
+app.add_typer(server_app, name="server")
+
+
+@server_app.command("start")
+def server_start(
+ port: int = typer.Option(9384, "--port", "-p", help="Server port."),
+ host: str = typer.Option("0.0.0.0", "--host", help="Bind address."),
+):
+ """Start the JigAi notification server."""
+ import uvicorn
+
+ from jigai.server.app import create_app
+ from jigai.server.discovery import get_local_ip
+
+ local_ip = get_local_ip()
+ console.print(f"\n[bold cyan]⚡ JigAi Server[/bold cyan]")
+ console.print(f" [dim]Local: http://localhost:{port}[/dim]")
+ console.print(f" [dim]Network: http://{local_ip}:{port}[/dim]")
+ console.print(f" [dim]WS: ws://{local_ip}:{port}/ws[/dim]")
+ console.print()
+
+ create_app(port=port)
+ uvicorn.run(
+ "jigai.server.app:app",
+ host=host,
+ port=port,
+ log_level="warning",
+ )
+
+
+@server_app.command("status")
+def server_status(
+ port: int = typer.Option(9384, "--port", "-p", help="Server port."),
+):
+ """Check if the JigAi server is running."""
+ from jigai.server.client import ServerClient
+
+ client = ServerClient(f"http://localhost:{port}")
+ if client.is_server_running():
+ console.print("[green]✓[/green] JigAi server is running")
+ else:
+ console.print("[red]✗[/red] JigAi server is not running")
+ console.print(" Start it with: [cyan]jigai server start[/cyan]")
+
+
+# ── Config Commands ─────────────────────────────────────────
+
+
+config_app = typer.Typer(help="Manage JigAi configuration.")
+app.add_typer(config_app, name="config")
+
+
+@config_app.command("init")
+def config_init():
+ """Create default configuration files."""
+ from jigai.config import (
+ CONFIG_FILE,
+ USER_PATTERNS_FILE,
+ ensure_dirs,
+ save_default_config,
+ )
+
+ ensure_dirs()
+
+ if CONFIG_FILE.exists():
+ console.print(f"[yellow]Config already exists:[/yellow] {CONFIG_FILE}")
+ else:
+ path = save_default_config()
+ console.print(f"[green]✓[/green] Created config: {path}")
+
+ if not USER_PATTERNS_FILE.exists():
+ # Create example user patterns file
+ example = (
+ "# JigAi — Custom tool patterns\n"
+ "# Add your own tools here.\n"
+ "#\n"
+ "# custom_tools:\n"
+ "# my_agent:\n"
+ '# name: "My Custom Agent"\n'
+ "# idle_patterns:\n"
+ "# - 'READY>'\n"
+ "# - 'awaiting instruction'\n"
+ "#\n"
+ "# overrides:\n"
+ "# timeout_seconds: 45\n"
+ )
+ USER_PATTERNS_FILE.write_text(example)
+ console.print(f"[green]✓[/green] Created patterns: {USER_PATTERNS_FILE}")
+ else:
+ console.print(
+ f"[yellow]Patterns already exists:[/yellow] {USER_PATTERNS_FILE}"
+ )
+
+
+@config_app.command("show")
+def config_show():
+ """Show current configuration."""
+ from jigai.config import load_config
+
+ config = load_config()
+ console.print_json(data=config.model_dump())
+
+
+@config_app.command("test")
+def config_test(
+ line: str = typer.Argument(..., help="A line of terminal output to test."),
+):
+ """Test if a line of output matches any idle pattern."""
+ from jigai.watcher.detector import strip_ansi
+ from jigai.watcher.patterns import load_patterns
+
+ registry = load_patterns()
+ clean = strip_ansi(line).strip()
+
+ console.print(f"Testing: [cyan]{clean}[/cyan]\n")
+
+ matched = False
+ for key, tool in registry.tools.items():
+ if tool.matches(clean):
+ console.print(f" [green]✓ MATCH[/green] → {tool.name} ({key})")
+ matched = True
+
+ if not matched:
+ console.print(" [yellow]No pattern matched.[/yellow]")
+ console.print(
+ f" [dim]Timeout fallback would trigger after "
+ f"{registry.timeout_seconds}s of silence.[/dim]"
+ )
+
+
+# ── Info Commands ───────────────────────────────────────────
+
+
+@app.command()
+def patterns():
+ """Show all loaded idle detection patterns."""
+ from jigai.watcher.patterns import load_patterns
+
+ registry = load_patterns()
+
+ table = Table(title="JigAi — Loaded Patterns")
+ table.add_column("Tool", style="cyan")
+ table.add_column("Key", style="dim")
+ table.add_column("Patterns", style="green")
+
+ for key, tool in registry.tools.items():
+ pat_list = "\n".join(p.pattern for p in tool.patterns)
+ table.add_row(tool.name, key, pat_list)
+
+ console.print(table)
+ console.print(
+ f"\n[dim]Timeout: {registry.timeout_seconds}s | "
+ f"Cooldown: {registry.cooldown_seconds}s[/dim]"
+ )
+
+
+@app.command()
+def sessions(
+ port: int = typer.Option(9384, "--port", "-p", help="Server port."),
+):
+ """List active watched sessions (requires server running)."""
+ import json
+ import urllib.request
+
+ try:
+ with urllib.request.urlopen(
+ f"http://localhost:{port}/api/sessions", timeout=2
+ ) as resp:
+ data = json.loads(resp.read())
+ except Exception:
+ console.print("[red]✗[/red] Cannot connect to server.")
+ console.print(" Start it with: [cyan]jigai server start[/cyan]")
+ raise typer.Exit(1)
+
+ sess_list = data.get("sessions", [])
+
+ if not sess_list:
+ console.print("[dim]No active sessions.[/dim]")
+ return
+
+ table = Table(title="Active Sessions")
+ table.add_column("Session ID", style="yellow")
+ table.add_column("Tool", style="cyan")
+ table.add_column("Status", style="green")
+ table.add_column("Working Dir", style="dim")
+
+ for s in sess_list:
+ status_style = "green" if s.get("status") == "active" else "yellow"
+ table.add_row(
+ s.get("session_id", "?"),
+ s.get("tool_name", "?"),
+ f"[{status_style}]{s.get('status', '?')}[/{status_style}]",
+ s.get("working_dir", ""),
+ )
+
+ console.print(table)
+
+
+if __name__ == "__main__":
+ app()
diff --git a/jigai/config.py b/jigai/config.py
new file mode 100644
index 0000000..ec9baa5
--- /dev/null
+++ b/jigai/config.py
@@ -0,0 +1,96 @@
+"""Configuration management for JigAi."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import Any
+
+import yaml
+from pydantic import BaseModel, Field
+
+
+JIGAI_DIR = Path.home() / ".jigai"
+CONFIG_FILE = JIGAI_DIR / "config.yaml"
+USER_PATTERNS_FILE = JIGAI_DIR / "patterns.yaml"
+DAEMON_PID_FILE = JIGAI_DIR / "daemon.pid"
+LOG_DIR = JIGAI_DIR / "logs"
+
+# Bundled defaults shipped with the package
+BUILTIN_PATTERNS_FILE = Path(__file__).parent.parent / "patterns" / "defaults.yaml"
+
+
+class NotificationConfig(BaseModel):
+ """Notification settings."""
+
+ macos: bool = True
+ only_when_away: bool = False # Skip notification if a terminal is the focused window
+ sound: str = "Ping"
+ group_by_session: bool = True
+ show_last_output: bool = True
+ output_lines: int = 3
+ redact_patterns: list[str] = Field(
+ default_factory=lambda: [r"(?i)(token|password|secret|key|api_key)=\S+"]
+ )
+
+
+class DetectionConfig(BaseModel):
+ """Detection engine settings."""
+
+ timeout_seconds: int = 30
+ cooldown_seconds: int = 5
+
+
+class ServerConfig(BaseModel):
+ """Server settings."""
+
+ port: int = 9384
+ bind: str = "0.0.0.0"
+
+
+class SessionConfig(BaseModel):
+ """Session display settings."""
+
+ show_working_dir: bool = True
+ show_last_output: bool = True
+
+
+class JigAiConfig(BaseModel):
+ """Root configuration model."""
+
+ server: ServerConfig = Field(default_factory=ServerConfig)
+ notifications: NotificationConfig = Field(default_factory=NotificationConfig)
+ detection: DetectionConfig = Field(default_factory=DetectionConfig)
+ sessions: SessionConfig = Field(default_factory=SessionConfig)
+
+
+def ensure_dirs() -> None:
+ """Create JigAi directories if they don't exist."""
+ JIGAI_DIR.mkdir(parents=True, exist_ok=True)
+ LOG_DIR.mkdir(parents=True, exist_ok=True)
+
+
+def load_config() -> JigAiConfig:
+ """Load configuration from ~/.jigai/config.yaml, falling back to defaults."""
+ if CONFIG_FILE.exists():
+ with open(CONFIG_FILE) as f:
+ raw: dict[str, Any] = yaml.safe_load(f) or {}
+ return JigAiConfig(**raw)
+ return JigAiConfig()
+
+
+def save_default_config() -> Path:
+ """Write default config to ~/.jigai/config.yaml."""
+ ensure_dirs()
+ config = JigAiConfig()
+ with open(CONFIG_FILE, "w") as f:
+ yaml.dump(config.model_dump(), f, default_flow_style=False, sort_keys=False)
+ return CONFIG_FILE
+
+
+def load_yaml(path: Path) -> dict[str, Any]:
+ """Safely load a YAML file, returning empty dict on failure."""
+ if not path.exists():
+ return {}
+ with open(path) as f:
+ return yaml.safe_load(f) or {}
diff --git a/jigai/models.py b/jigai/models.py
new file mode 100644
index 0000000..6e27885
--- /dev/null
+++ b/jigai/models.py
@@ -0,0 +1,49 @@
+"""Shared data models for JigAi."""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime, timezone
+from enum import Enum
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+
+class SessionStatus(str, Enum):
+ """Status of a watched session."""
+
+ ACTIVE = "active"
+ IDLE = "idle"
+ STOPPED = "stopped"
+
+
+class IdleEvent(BaseModel):
+ """Emitted when a watched session goes idle."""
+
+ session_id: str
+ tool_name: str
+ working_dir: str
+ timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+ last_output: str = ""
+ idle_seconds: float = 0.0
+ detection_method: str = "pattern" # "pattern" | "timeout" | "combined"
+
+
+class Session(BaseModel):
+ """Represents a watched terminal session."""
+
+ session_id: str = Field(default_factory=lambda: uuid.uuid4().hex[:8])
+ tool_name: str = "unknown"
+ command: list[str] = Field(default_factory=list)
+ working_dir: str = ""
+ started_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+ status: SessionStatus = SessionStatus.ACTIVE
+ last_output: str = ""
+ last_idle_event: Optional[IdleEvent] = None
+ pid: Optional[int] = None
+
+ def to_display_name(self) -> str:
+ """Short display name for the session."""
+ tool = self.tool_name or "session"
+ return f"{tool}-{self.session_id}"
diff --git a/jigai/notifier/__init__.py b/jigai/notifier/__init__.py
new file mode 100644
index 0000000..abbf8e9
--- /dev/null
+++ b/jigai/notifier/__init__.py
@@ -0,0 +1 @@
+"""JigAi notifier — macOS and cross-platform notification support."""
diff --git a/jigai/notifier/macos.py b/jigai/notifier/macos.py
new file mode 100644
index 0000000..70c1d6d
--- /dev/null
+++ b/jigai/notifier/macos.py
@@ -0,0 +1,121 @@
+"""macOS notification support via osascript and terminal-notifier."""
+
+from __future__ import annotations
+
+import shutil
+import subprocess
+from typing import Optional
+
+
+_TERMINAL_APPS = {
+ "terminal", "iterm2", "warp", "hyper", "alacritty",
+ "kitty", "ghostty", "tabby", "rio",
+}
+
+
+def _has_terminal_notifier() -> bool:
+ """Check if terminal-notifier is installed."""
+ return shutil.which("terminal-notifier") is not None
+
+
+def is_terminal_focused() -> bool:
+ """Return True if a terminal app is currently the frontmost window."""
+ try:
+ result = subprocess.run(
+ [
+ "osascript", "-e",
+ 'tell application "System Events" to get name of first '
+ 'application process whose frontmost is true',
+ ],
+ capture_output=True,
+ text=True,
+ timeout=2,
+ )
+ frontmost = result.stdout.strip().lower()
+ return any(term in frontmost for term in _TERMINAL_APPS)
+ except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
+ return False # Can't tell — assume not focused, allow notification
+
+
+def notify_macos(
+ title: str,
+ message: str,
+ subtitle: Optional[str] = None,
+ sound: str = "Ping",
+ group: Optional[str] = None,
+) -> None:
+ """
+ Send a macOS notification.
+
+ Uses terminal-notifier if available (richer features, click actions),
+ falls back to osascript (zero dependencies).
+ """
+ # Sanitize inputs for shell safety
+ title = _sanitize(title)
+ message = _sanitize(message)
+ if subtitle:
+ subtitle = _sanitize(subtitle)
+
+ if _has_terminal_notifier():
+ _notify_terminal_notifier(title, message, subtitle, sound, group)
+ else:
+ _notify_osascript(title, message, subtitle, sound)
+
+
+def _notify_osascript(
+ title: str,
+ message: str,
+ subtitle: Optional[str] = None,
+ sound: str = "Ping",
+) -> None:
+ """Send notification via osascript (built into macOS)."""
+ parts = [f'display notification "{message}"']
+ parts.append(f'with title "{title}"')
+ if subtitle:
+ parts.append(f'subtitle "{subtitle}"')
+ parts.append(f'sound name "{sound}"')
+
+ script = " ".join(parts)
+
+ try:
+ subprocess.run(
+ ["osascript", "-e", script],
+ capture_output=True,
+ timeout=5,
+ )
+ except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
+ pass # Silently fail — we're a notification, not critical
+
+
+def _notify_terminal_notifier(
+ title: str,
+ message: str,
+ subtitle: Optional[str] = None,
+ sound: str = "Ping",
+ group: Optional[str] = None,
+) -> None:
+ """Send notification via terminal-notifier (richer features)."""
+ cmd = [
+ "terminal-notifier",
+ "-title", title,
+ "-message", message,
+ "-sound", sound,
+ ]
+
+ if subtitle:
+ cmd += ["-subtitle", subtitle]
+
+ if group:
+ cmd += ["-group", f"jigai-{group}"]
+
+ try:
+ subprocess.run(cmd, capture_output=True, timeout=5)
+ except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
+ # Fall back to osascript
+ _notify_osascript(title, message, subtitle, sound)
+
+
+def _sanitize(text: str) -> str:
+ """Sanitize text for use in osascript/shell commands."""
+ # Replace characters that could break AppleScript strings
+ return text.replace('"', '\\"').replace("\\", "\\\\").replace("\n", " ⏎ ")
diff --git a/jigai/server/__init__.py b/jigai/server/__init__.py
new file mode 100644
index 0000000..738e9b7
--- /dev/null
+++ b/jigai/server/__init__.py
@@ -0,0 +1 @@
+"""JigAi server — FastAPI WebSocket server for mobile notifications."""
diff --git a/jigai/server/app.py b/jigai/server/app.py
new file mode 100644
index 0000000..5be45ba
--- /dev/null
+++ b/jigai/server/app.py
@@ -0,0 +1,208 @@
+"""FastAPI server — receives idle events and broadcasts to mobile clients."""
+
+from __future__ import annotations
+
+import asyncio
+from contextlib import asynccontextmanager
+from datetime import datetime, timezone
+from typing import Any
+
+from fastapi import FastAPI, WebSocket, WebSocketDisconnect
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
+
+from jigai.models import IdleEvent, Session, SessionStatus
+from jigai.server.discovery import ServiceBroadcaster
+from jigai.server.ws_manager import ConnectionManager
+
+
+# Global state
+manager = ConnectionManager()
+broadcaster = ServiceBroadcaster()
+sessions: dict[str, dict[str, Any]] = {}
+event_history: list[dict[str, Any]] = []
+MAX_HISTORY = 100
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+ """Start/stop mDNS broadcasting with the server lifecycle."""
+ port = app.state.port if hasattr(app.state, "port") else 9384
+ broadcaster.port = port
+ broadcaster.start()
+ yield
+ broadcaster.stop()
+
+
+app = FastAPI(
+ title="JigAi Server",
+ description="Terminal notification hub for AI coding agents",
+ version="0.1.0",
+ lifespan=lifespan,
+)
+
+# Allow CORS for mobile app
+app.add_middleware(
+ CORSMiddleware,
+ allow_origins=["*"],
+ allow_credentials=True,
+ allow_methods=["*"],
+ allow_headers=["*"],
+)
+
+
+# ── REST Endpoints ──────────────────────────────────────────
+
+
+@app.get("/api/health")
+async def health():
+ """Health check endpoint."""
+ return {
+ "status": "ok",
+ "version": "0.1.0",
+ "clients": manager.client_count,
+ "sessions": len(sessions),
+ }
+
+
+@app.get("/api/sessions")
+async def list_sessions():
+ """List all active watched sessions."""
+ return {"sessions": list(sessions.values())}
+
+
+@app.get("/api/events")
+async def list_events(limit: int = 20):
+ """List recent idle events."""
+ return {"events": event_history[-limit:]}
+
+
+class IdleEventRequest(BaseModel):
+ """Incoming idle event from a watcher."""
+
+ session_id: str
+ tool_name: str
+ working_dir: str = ""
+ last_output: str = ""
+ idle_seconds: float = 0.0
+ detection_method: str = "pattern"
+
+
+@app.post("/api/events")
+async def receive_event(event: IdleEventRequest):
+ """Receive an idle event from a watcher and broadcast to clients."""
+ event_data = {
+ "type": "idle_detected",
+ "session_id": event.session_id,
+ "tool_name": event.tool_name,
+ "working_dir": event.working_dir,
+ "last_output": event.last_output,
+ "idle_seconds": event.idle_seconds,
+ "detection_method": event.detection_method,
+ "timestamp": datetime.now(timezone.utc).isoformat(),
+ }
+
+ # Update session registry
+ sessions[event.session_id] = {
+ "session_id": event.session_id,
+ "tool_name": event.tool_name,
+ "working_dir": event.working_dir,
+ "status": "idle",
+ "last_event": event_data,
+ }
+
+ # Store in history
+ event_history.append(event_data)
+ if len(event_history) > MAX_HISTORY:
+ event_history.pop(0)
+
+ # Broadcast to all WebSocket clients
+ await manager.broadcast(event_data)
+
+ return {"status": "ok", "clients_notified": manager.client_count}
+
+
+class SessionRegisterRequest(BaseModel):
+ """Register a new watched session."""
+
+ session_id: str
+ tool_name: str
+ command: list[str] = []
+ working_dir: str = ""
+
+
+@app.post("/api/sessions")
+async def register_session(req: SessionRegisterRequest):
+ """Register a new watched session."""
+ sessions[req.session_id] = {
+ "session_id": req.session_id,
+ "tool_name": req.tool_name,
+ "command": req.command,
+ "working_dir": req.working_dir,
+ "status": "active",
+ "registered_at": datetime.now(timezone.utc).isoformat(),
+ }
+
+ await manager.broadcast({
+ "type": "session_started",
+ "session_id": req.session_id,
+ "tool_name": req.tool_name,
+ "working_dir": req.working_dir,
+ })
+
+ return {"status": "ok"}
+
+
+@app.delete("/api/sessions/{session_id}")
+async def unregister_session(session_id: str):
+ """Remove a watched session."""
+ if session_id in sessions:
+ del sessions[session_id]
+
+ await manager.broadcast({
+ "type": "session_stopped",
+ "session_id": session_id,
+ })
+
+ return {"status": "ok"}
+
+
+# ── WebSocket Endpoint ──────────────────────────────────────
+
+
+@app.websocket("/ws")
+async def websocket_endpoint(websocket: WebSocket):
+ """WebSocket endpoint for mobile clients."""
+ await manager.connect(websocket)
+
+ # Send current state on connect
+ try:
+ await websocket.send_json({
+ "type": "connected",
+ "sessions": list(sessions.values()),
+ "server_version": "0.1.0",
+ })
+
+ # Keep connection alive
+ while True:
+ try:
+ # Wait for messages (ping/pong, or future commands)
+ data = await asyncio.wait_for(websocket.receive_text(), timeout=30.0)
+ except asyncio.TimeoutError:
+ # Send heartbeat
+ try:
+ await websocket.send_json({"type": "heartbeat"})
+ except Exception:
+ break
+ except WebSocketDisconnect:
+ pass
+ except Exception:
+ pass
+ finally:
+ await manager.disconnect(websocket)
+
+
+def create_app(port: int = 9384) -> FastAPI:
+ """Create the app with the given port for mDNS."""
+ app.state.port = port
+ return app
diff --git a/jigai/server/client.py b/jigai/server/client.py
new file mode 100644
index 0000000..214c5e3
--- /dev/null
+++ b/jigai/server/client.py
@@ -0,0 +1,88 @@
+"""HTTP client for pushing events from watchers to the JigAi server."""
+
+from __future__ import annotations
+
+import json
+import urllib.request
+import urllib.error
+from typing import Optional
+
+from jigai.models import IdleEvent
+
+
+class ServerClient:
+ """
+ Lightweight HTTP client for pushing events to the JigAi server.
+
+ Uses stdlib urllib to avoid adding httpx/requests as a dependency.
+ """
+
+ def __init__(self, base_url: str = "http://localhost:9384"):
+ self.base_url = base_url.rstrip("/")
+
+ def push_event(self, event: IdleEvent) -> bool:
+ """Push an idle event to the server. Returns True on success."""
+ url = f"{self.base_url}/api/events"
+ data = json.dumps(event.model_dump(), default=str).encode("utf-8")
+
+ req = urllib.request.Request(
+ url,
+ data=data,
+ headers={"Content-Type": "application/json"},
+ method="POST",
+ )
+
+ try:
+ with urllib.request.urlopen(req, timeout=3) as resp:
+ return resp.status == 200
+ except (urllib.error.URLError, OSError, TimeoutError):
+ return False
+
+ def register_session(
+ self,
+ session_id: str,
+ tool_name: str,
+ command: list[str],
+ working_dir: str,
+ ) -> bool:
+ """Register a session with the server."""
+ url = f"{self.base_url}/api/sessions"
+ data = json.dumps({
+ "session_id": session_id,
+ "tool_name": tool_name,
+ "command": command,
+ "working_dir": working_dir,
+ }).encode("utf-8")
+
+ req = urllib.request.Request(
+ url,
+ data=data,
+ headers={"Content-Type": "application/json"},
+ method="POST",
+ )
+
+ try:
+ with urllib.request.urlopen(req, timeout=3) as resp:
+ return resp.status == 200
+ except (urllib.error.URLError, OSError, TimeoutError):
+ return False
+
+ def unregister_session(self, session_id: str) -> bool:
+ """Unregister a session from the server."""
+ url = f"{self.base_url}/api/sessions/{session_id}"
+ req = urllib.request.Request(url, method="DELETE")
+
+ try:
+ with urllib.request.urlopen(req, timeout=3) as resp:
+ return resp.status == 200
+ except (urllib.error.URLError, OSError, TimeoutError):
+ return False
+
+ def is_server_running(self) -> bool:
+ """Check if the server is reachable."""
+ url = f"{self.base_url}/api/health"
+ try:
+ with urllib.request.urlopen(url, timeout=2) as resp:
+ return resp.status == 200
+ except (urllib.error.URLError, OSError, TimeoutError):
+ return False
diff --git a/jigai/server/discovery.py b/jigai/server/discovery.py
new file mode 100644
index 0000000..ccc4472
--- /dev/null
+++ b/jigai/server/discovery.py
@@ -0,0 +1,83 @@
+"""mDNS/Bonjour service discovery for JigAi server."""
+
+from __future__ import annotations
+
+import socket
+from typing import Optional
+
+from rich.console import Console
+
+console = Console(stderr=True)
+
+
+def get_local_ip() -> str:
+ """Get the local IP address of this machine on the LAN."""
+ try:
+ # Connect to a public address to determine local interface
+ # (no data is actually sent)
+ s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+ s.connect(("8.8.8.8", 80))
+ ip = s.getsockname()[0]
+ s.close()
+ return ip
+ except Exception:
+ return "127.0.0.1"
+
+
+class ServiceBroadcaster:
+ """Broadcasts JigAi server via mDNS/Bonjour for auto-discovery."""
+
+ def __init__(self, port: int = 9384):
+ self.port = port
+ self._zeroconf = None
+ self._info = None
+
+ def start(self) -> bool:
+ """Start broadcasting the service. Returns True on success."""
+ try:
+ from zeroconf import ServiceInfo, Zeroconf
+
+ local_ip = get_local_ip()
+ hostname = socket.gethostname()
+
+ self._info = ServiceInfo(
+ "_jigai._tcp.local.",
+ f"JigAi on {hostname}._jigai._tcp.local.",
+ addresses=[socket.inet_aton(local_ip)],
+ port=self.port,
+ properties={
+ "version": "0.1.0",
+ "hostname": hostname,
+ },
+ server=f"{hostname}.local.",
+ )
+
+ self._zeroconf = Zeroconf()
+ self._zeroconf.register_service(self._info)
+
+ console.print(
+ f" [dim]mDNS: Broadcasting as [cyan]_jigai._tcp.local.[/cyan] "
+ f"at {local_ip}:{self.port}[/dim]"
+ )
+ return True
+
+ except ImportError:
+ console.print(
+ " [dim yellow]mDNS: zeroconf not installed, "
+ "mobile auto-discovery disabled[/dim yellow]"
+ )
+ return False
+ except Exception as e:
+ console.print(f" [dim yellow]mDNS: Failed to start ({e})[/dim yellow]")
+ return False
+
+ def stop(self) -> None:
+ """Stop broadcasting."""
+ if self._zeroconf and self._info:
+ try:
+ self._zeroconf.unregister_service(self._info)
+ self._zeroconf.close()
+ except Exception:
+ pass
+ self._zeroconf = None
+ self._info = None
diff --git a/jigai/server/ws_manager.py b/jigai/server/ws_manager.py
new file mode 100644
index 0000000..a68b14d
--- /dev/null
+++ b/jigai/server/ws_manager.py
@@ -0,0 +1,55 @@
+"""WebSocket connection manager for broadcasting events to mobile clients."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from datetime import datetime, timezone
+from typing import Any
+
+from fastapi import WebSocket
+
+
+class ConnectionManager:
+ """Manages WebSocket connections from mobile clients."""
+
+ def __init__(self):
+ self.active_connections: list[WebSocket] = []
+ self._lock = asyncio.Lock()
+
+ async def connect(self, websocket: WebSocket) -> None:
+ """Accept and register a new WebSocket connection."""
+ await websocket.accept()
+ async with self._lock:
+ self.active_connections.append(websocket)
+
+ async def disconnect(self, websocket: WebSocket) -> None:
+ """Remove a WebSocket connection."""
+ async with self._lock:
+ if websocket in self.active_connections:
+ self.active_connections.remove(websocket)
+
+ async def broadcast(self, data: dict[str, Any]) -> None:
+ """Broadcast a message to all connected clients."""
+ if not self.active_connections:
+ return
+
+ # Add server timestamp
+ data["server_time"] = datetime.now(timezone.utc).isoformat()
+ message = json.dumps(data, default=str)
+
+ # Send to all, remove dead connections
+ dead: list[WebSocket] = []
+ async with self._lock:
+ for ws in self.active_connections:
+ try:
+ await ws.send_text(message)
+ except Exception:
+ dead.append(ws)
+
+ for ws in dead:
+ self.active_connections.remove(ws)
+
+ @property
+ def client_count(self) -> int:
+ return len(self.active_connections)
diff --git a/jigai/watcher/__init__.py b/jigai/watcher/__init__.py
new file mode 100644
index 0000000..67cdb2f
--- /dev/null
+++ b/jigai/watcher/__init__.py
@@ -0,0 +1 @@
+"""JigAi watcher — terminal monitoring and idle detection."""
diff --git a/jigai/watcher/detector.py b/jigai/watcher/detector.py
new file mode 100644
index 0000000..5317c32
--- /dev/null
+++ b/jigai/watcher/detector.py
@@ -0,0 +1,139 @@
+"""Detection engine — determines when a terminal session has gone idle."""
+
+from __future__ import annotations
+
+import re
+import time
+from collections import deque
+from dataclasses import dataclass, field
+from typing import Callable, Optional
+
+from jigai.watcher.patterns import PatternRegistry
+
+
+# Regex to strip ANSI escape codes from terminal output
+_ANSI_RE = re.compile(r"\x1b\[[0-9;]*[a-zA-Z]|\x1b\].*?\x07|\x1b\[.*?[@-~]")
+
+
+def strip_ansi(text: str) -> str:
+ """Remove ANSI escape codes from text."""
+ return _ANSI_RE.sub("", text)
+
+
+@dataclass
+class DetectorState:
+ """Tracks the state of idle detection for a single session."""
+
+ last_output_time: float = field(default_factory=time.time)
+ last_idle_notification: float = 0.0
+ output_buffer: deque = field(default_factory=lambda: deque(maxlen=50))
+ is_idle: bool = False
+ detected_tool: Optional[str] = None
+
+
+class Detector:
+ """
+ Idle detection engine.
+
+ Combines pattern matching and timeout-based detection.
+ Feeds terminal output line-by-line and emits idle events via callback.
+ """
+
+ def __init__(
+ self,
+ registry: PatternRegistry,
+ on_idle: Callable[[str, str, float, list[str]], None],
+ tool_hint: Optional[str] = None,
+ ):
+ """
+ Args:
+ registry: Pattern registry with tool patterns loaded.
+ on_idle: Callback(detection_method, tool_key, idle_seconds, recent_lines).
+ tool_hint: Optional tool key hint from command detection.
+ """
+ self.registry = registry
+ self.on_idle = on_idle
+ self.tool_hint = tool_hint
+ self.state = DetectorState()
+
+ # Redaction patterns (loaded externally)
+ self._redact_patterns: list[re.Pattern] = []
+
+ def set_redact_patterns(self, patterns: list[str]) -> None:
+ """Set patterns for redacting sensitive info from output."""
+ self._redact_patterns = []
+ for pat in patterns:
+ try:
+ self._redact_patterns.append(re.compile(pat))
+ except re.error:
+ pass
+
+ def _redact(self, line: str) -> str:
+ """Redact sensitive information from a line."""
+ for pat in self._redact_patterns:
+ line = pat.sub("[REDACTED]", line)
+ return line
+
+ def feed_line(self, raw_line: str) -> None:
+ """
+ Feed a single line of terminal output to the detector.
+
+ This is called for every line of stdout from the watched process.
+ """
+ now = time.time()
+ clean = strip_ansi(raw_line).strip()
+
+ if not clean:
+ return
+
+ # Store in buffer (redacted)
+ self.state.output_buffer.append(self._redact(clean))
+ self.state.last_output_time = now
+ self.state.is_idle = False
+
+ # Try pattern matching
+ # If we have a tool hint, check that tool first
+ matched_tool = None
+ if self.tool_hint and self.tool_hint in self.registry.tools:
+ tool = self.registry.tools[self.tool_hint]
+ if tool.matches(clean):
+ matched_tool = self.tool_hint
+
+ # If no match from hinted tool, try all tools
+ if matched_tool is None:
+ matched_tool = self.registry.match_any(clean)
+
+ if matched_tool is not None:
+ self._trigger_idle("pattern", matched_tool, now)
+
+ def check_timeout(self) -> None:
+ """
+ Check if the timeout-based idle detection should trigger.
+
+ Call this periodically (e.g., every second) from the watcher loop.
+ """
+ now = time.time()
+ elapsed = now - self.state.last_output_time
+
+ if elapsed >= self.registry.timeout_seconds and not self.state.is_idle:
+ tool_key = self.tool_hint or "unknown"
+ self._trigger_idle("timeout", tool_key, now)
+
+ def _trigger_idle(self, method: str, tool_key: str, now: float) -> None:
+ """Trigger an idle event if cooldown has passed."""
+ cooldown = self.registry.cooldown_seconds
+ if now - self.state.last_idle_notification < cooldown:
+ return
+
+ self.state.is_idle = True
+ self.state.last_idle_notification = now
+ self.state.detected_tool = tool_key
+
+ idle_seconds = now - self.state.last_output_time
+ recent = list(self.state.output_buffer)[-10:] # Last 10 lines for context
+
+ self.on_idle(method, tool_key, idle_seconds, recent)
+
+ def get_recent_output(self, n: int = 3) -> list[str]:
+ """Get the last N lines of (redacted) output."""
+ return list(self.state.output_buffer)[-n:]
diff --git a/jigai/watcher/patterns.py b/jigai/watcher/patterns.py
new file mode 100644
index 0000000..93712d9
--- /dev/null
+++ b/jigai/watcher/patterns.py
@@ -0,0 +1,128 @@
+"""Pattern loading and management for idle detection."""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Optional
+
+from jigai.config import BUILTIN_PATTERNS_FILE, USER_PATTERNS_FILE, load_yaml
+
+
+@dataclass
+class ToolPattern:
+ """Compiled patterns for a single tool."""
+
+ name: str
+ key: str
+ patterns: list[re.Pattern] = field(default_factory=list)
+
+ def matches(self, line: str) -> bool:
+ """Check if a line matches any of this tool's idle patterns."""
+ return any(p.search(line) for p in self.patterns)
+
+
+@dataclass
+class PatternRegistry:
+ """Registry of all loaded tool patterns."""
+
+ tools: dict[str, ToolPattern] = field(default_factory=dict)
+ timeout_seconds: int = 30
+ cooldown_seconds: int = 5
+
+ def match_any(self, line: str) -> Optional[str]:
+ """Check if a line matches any tool's idle pattern. Returns tool key or None."""
+ for key, tool in self.tools.items():
+ if tool.matches(line):
+ return key
+ return None
+
+ def get_tool_name(self, key: str) -> str:
+ """Get the display name for a tool key."""
+ if key in self.tools:
+ return self.tools[key].name
+ return key
+
+
+def _compile_patterns(raw_patterns: list[str]) -> list[re.Pattern]:
+ """Compile a list of regex strings, skipping invalid ones."""
+ compiled = []
+ for pat_str in raw_patterns:
+ try:
+ compiled.append(re.compile(pat_str))
+ except re.error:
+ # Skip invalid patterns silently — log in future
+ pass
+ return compiled
+
+
+def load_patterns() -> PatternRegistry:
+ """Load patterns from built-in defaults and user overrides."""
+ registry = PatternRegistry()
+
+ # Load built-in patterns
+ builtin = load_yaml(BUILTIN_PATTERNS_FILE)
+ if "tools" in builtin:
+ for key, tool_data in builtin["tools"].items():
+ name = tool_data.get("name", key)
+ raw = tool_data.get("idle_patterns", [])
+ registry.tools[key] = ToolPattern(
+ name=name,
+ key=key,
+ patterns=_compile_patterns(raw),
+ )
+
+ if "defaults" in builtin:
+ registry.timeout_seconds = builtin["defaults"].get(
+ "timeout_seconds", registry.timeout_seconds
+ )
+ registry.cooldown_seconds = builtin["defaults"].get(
+ "cooldown_seconds", registry.cooldown_seconds
+ )
+
+ # Load user patterns (override/extend)
+ user = load_yaml(USER_PATTERNS_FILE)
+ if "custom_tools" in user:
+ for key, tool_data in user["custom_tools"].items():
+ name = tool_data.get("name", key)
+ raw = tool_data.get("idle_patterns", [])
+ registry.tools[key] = ToolPattern(
+ name=name,
+ key=key,
+ patterns=_compile_patterns(raw),
+ )
+
+ if "overrides" in user:
+ overrides = user["overrides"]
+ if "timeout_seconds" in overrides:
+ registry.timeout_seconds = overrides["timeout_seconds"]
+ if "cooldown_seconds" in overrides:
+ registry.cooldown_seconds = overrides["cooldown_seconds"]
+
+ return registry
+
+
+def detect_tool_from_command(command: list[str], registry: PatternRegistry) -> str:
+ """Try to detect the tool name from the command being run."""
+ if not command:
+ return "unknown"
+
+ cmd_str = " ".join(command).lower()
+
+ # Map common command names to tool keys
+ tool_hints: dict[str, list[str]] = {
+ "claude_code": ["claude"],
+ "codex": ["codex"],
+ "gemini_cli": ["gemini"],
+ "aider": ["aider"],
+ "opencode": ["opencode"],
+ }
+
+ for tool_key, hints in tool_hints.items():
+ if tool_key in registry.tools:
+ for hint in hints:
+ if hint in cmd_str:
+ return tool_key
+
+ return "unknown"
diff --git a/jigai/watcher/pty_proxy.py b/jigai/watcher/pty_proxy.py
new file mode 100644
index 0000000..ce59e8c
--- /dev/null
+++ b/jigai/watcher/pty_proxy.py
@@ -0,0 +1,244 @@
+"""PTY proxy — transparent terminal wrapper for monitoring AI tool output."""
+
+from __future__ import annotations
+
+import errno
+import fcntl
+import os
+import pty
+import select
+import signal
+import struct
+import sys
+import termios
+import time
+import tty
+from typing import Callable, Optional
+
+
+def _set_nonblocking(fd: int) -> None:
+ """Set a file descriptor to non-blocking mode."""
+ flags = fcntl.fcntl(fd, fcntl.F_GETFL)
+ fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)
+
+
+def _get_terminal_size() -> tuple[int, int]:
+ """Get current terminal size (rows, cols)."""
+ try:
+ size = os.get_terminal_size()
+ return (size.lines, size.columns)
+ except OSError:
+ return (24, 80)
+
+
+def _set_pty_size(fd: int, rows: int, cols: int) -> None:
+ """Set the PTY window size."""
+ winsize = struct.pack("HHHH", rows, cols, 0, 0)
+ fcntl.ioctl(fd, termios.TIOCSWINSZ, winsize)
+
+
+class PtyProxy:
+ """
+ Transparent PTY proxy that wraps a child process.
+
+ All I/O passes through unchanged — the child process behaves identically
+ to running directly in the terminal. Output is simultaneously fed to a
+ callback for idle detection.
+ """
+
+ def __init__(
+ self,
+ command: list[str],
+ on_output: Callable[[bytes], None],
+ on_exit: Optional[Callable[[int], None]] = None,
+ ):
+ """
+ Args:
+ command: Command + args to spawn (e.g., ["claude"]).
+ on_output: Called with raw bytes from child stdout.
+ on_exit: Called with exit code when child terminates.
+ """
+ self.command = command
+ self.on_output = on_output
+ self.on_exit = on_exit
+ self._master_fd: Optional[int] = None
+ self._child_pid: Optional[int] = None
+ self._running = False
+ self._old_termios: Optional[list] = None
+
+ @property
+ def child_pid(self) -> Optional[int]:
+ return self._child_pid
+
+ def run(self) -> int:
+ """
+ Run the child process in a PTY proxy. Blocks until child exits.
+
+ Returns the child's exit code.
+ """
+ # Save terminal state and switch to raw mode
+ stdin_fd = sys.stdin.fileno()
+ try:
+ self._old_termios = termios.tcgetattr(stdin_fd)
+ except termios.error:
+ self._old_termios = None
+
+ # Create PTY pair
+ self._master_fd, slave_fd = pty.openpty()
+
+ # Set PTY size to match current terminal
+ rows, cols = _get_terminal_size()
+ _set_pty_size(self._master_fd, rows, cols)
+
+ # Fork the child process
+ self._child_pid = os.fork()
+
+ if self._child_pid == 0:
+ # === CHILD PROCESS ===
+ os.close(self._master_fd)
+ os.setsid()
+
+ # Set the slave as the controlling terminal
+ fcntl.ioctl(slave_fd, termios.TIOCSCTTY, 0)
+
+ # Redirect stdin/stdout/stderr to slave PTY
+ os.dup2(slave_fd, 0)
+ os.dup2(slave_fd, 1)
+ os.dup2(slave_fd, 2)
+
+ if slave_fd > 2:
+ os.close(slave_fd)
+
+ # Execute the command
+ os.execvp(self.command[0], self.command)
+
+ # === PARENT PROCESS ===
+ os.close(slave_fd)
+ self._running = True
+
+ # Handle SIGWINCH (terminal resize)
+ def _handle_resize(signum, frame):
+ rows, cols = _get_terminal_size()
+ if self._master_fd is not None:
+ try:
+ _set_pty_size(self._master_fd, rows, cols)
+ except OSError:
+ pass
+
+ signal.signal(signal.SIGWINCH, _handle_resize)
+
+ # Switch stdin to raw mode so keystrokes pass through immediately
+ if self._old_termios is not None:
+ try:
+ tty.setraw(stdin_fd)
+ except termios.error:
+ pass
+
+ exit_code = self._io_loop(stdin_fd)
+
+ # Restore terminal
+ self._restore_terminal(stdin_fd)
+
+ if self.on_exit:
+ self.on_exit(exit_code)
+
+ return exit_code
+
+ def _io_loop(self, stdin_fd: int) -> int:
+ """Main I/O loop — proxy data between user and child."""
+ master_fd = self._master_fd
+ assert master_fd is not None
+
+ _set_nonblocking(master_fd)
+ _set_nonblocking(stdin_fd)
+
+ exit_code = 0
+
+ while self._running:
+ try:
+ rlist, _, _ = select.select([master_fd, stdin_fd], [], [], 1.0)
+ except (select.error, ValueError, OSError):
+ break
+
+ if master_fd in rlist:
+ # Data from child → write to user's terminal + feed to detector
+ try:
+ data = os.read(master_fd, 16384)
+ if not data:
+ break
+ os.write(sys.stdout.fileno(), data)
+ self.on_output(data)
+ except OSError as e:
+ if e.errno == errno.EIO:
+ break # Child closed PTY
+ if e.errno != errno.EAGAIN:
+ break
+
+ if stdin_fd in rlist:
+ # Data from user → write to child
+ try:
+ data = os.read(stdin_fd, 16384)
+ if not data:
+ break
+ os.write(master_fd, data)
+ except OSError as e:
+ if e.errno != errno.EAGAIN:
+ break
+
+ # Check if child is still alive
+ try:
+ pid, status = os.waitpid(self._child_pid, os.WNOHANG)
+ if pid != 0:
+ if os.WIFEXITED(status):
+ exit_code = os.WEXITSTATUS(status)
+ else:
+ exit_code = -1
+ self._running = False
+ except ChildProcessError:
+ self._running = False
+
+ # Drain any remaining output
+ try:
+ while True:
+ data = os.read(master_fd, 16384)
+ if not data:
+ break
+ os.write(sys.stdout.fileno(), data)
+ self.on_output(data)
+ except OSError:
+ pass
+
+ # Wait for child if still running
+ if self._running:
+ try:
+ _, status = os.waitpid(self._child_pid, 0)
+ if os.WIFEXITED(status):
+ exit_code = os.WEXITSTATUS(status)
+ except ChildProcessError:
+ pass
+
+ # Clean up
+ try:
+ os.close(master_fd)
+ except OSError:
+ pass
+ self._master_fd = None
+
+ return exit_code
+
+ def _restore_terminal(self, stdin_fd: int) -> None:
+ """Restore original terminal settings."""
+ if self._old_termios is not None:
+ try:
+ termios.tcsetattr(stdin_fd, termios.TCSADRAIN, self._old_termios)
+ except termios.error:
+ pass
+
+ def stop(self) -> None:
+ """Stop the proxy and kill the child process."""
+ self._running = False
+ if self._child_pid:
+ try:
+ os.kill(self._child_pid, signal.SIGTERM)
+ except ProcessLookupError:
+ pass
diff --git a/jigai/watcher/watcher.py b/jigai/watcher/watcher.py
new file mode 100644
index 0000000..d581647
--- /dev/null
+++ b/jigai/watcher/watcher.py
@@ -0,0 +1,234 @@
+"""Watcher — orchestrates PTY proxy + detection engine for a single session."""
+
+from __future__ import annotations
+
+import os
+import threading
+import time
+from typing import Optional
+
+from rich.console import Console
+
+from jigai.config import JigAiConfig, load_config
+from jigai.models import IdleEvent, Session, SessionStatus
+from jigai.watcher.detector import Detector, strip_ansi
+from jigai.watcher.patterns import PatternRegistry, detect_tool_from_command, load_patterns
+from jigai.watcher.pty_proxy import PtyProxy
+
+console = Console(stderr=True)
+
+
+class Watcher:
+ """
+ Watches a single command via PTY proxy and detects idle state.
+
+ Combines the PTY proxy (transparent I/O) with the detector (pattern + timeout)
+ and emits notifications when idle is detected.
+ """
+
+ def __init__(
+ self,
+ command: list[str],
+ tool_override: Optional[str] = None,
+ config: Optional[JigAiConfig] = None,
+ registry: Optional[PatternRegistry] = None,
+ on_idle_event: Optional[callable] = None,
+ ):
+ self.command = command
+ self.config = config or load_config()
+ self.registry = registry or load_patterns()
+
+ # Detect tool from command
+ if tool_override:
+ self.tool_key = tool_override
+ else:
+ self.tool_key = detect_tool_from_command(command, self.registry)
+
+ tool_name = self.registry.get_tool_name(self.tool_key)
+
+ # Create session
+ self.session = Session(
+ tool_name=tool_name,
+ command=command,
+ working_dir=os.getcwd(),
+ )
+
+ # External callback for idle events (e.g., server push)
+ self._on_idle_event = on_idle_event
+
+ # Line buffer for partial line accumulation
+ self._line_buffer = ""
+
+ # Create detector
+ self.detector = Detector(
+ registry=self.registry,
+ on_idle=self._handle_idle,
+ tool_hint=self.tool_key,
+ )
+
+ # Set redaction patterns
+ self.detector.set_redact_patterns(self.config.notifications.redact_patterns)
+
+ # Timeout checker thread
+ self._timeout_thread: Optional[threading.Thread] = None
+ self._running = False
+
+ def _handle_output(self, data: bytes) -> None:
+ """Called by PTY proxy with raw bytes from child stdout."""
+ try:
+ text = data.decode("utf-8", errors="replace")
+ except Exception:
+ return
+
+ # Accumulate into lines and feed to detector
+ self._line_buffer += text
+ while "\n" in self._line_buffer:
+ line, self._line_buffer = self._line_buffer.split("\n", 1)
+ self.detector.feed_line(line)
+
+ # Also check for prompt-like patterns in partial lines
+ # (prompts often don't end with newline)
+ if self._line_buffer.strip():
+ self.detector.feed_line(self._line_buffer)
+
+ def _handle_idle(
+ self, method: str, tool_key: str, idle_seconds: float, recent_lines: list[str]
+ ) -> None:
+ """Called by detector when idle is detected."""
+ tool_name = self.registry.get_tool_name(tool_key)
+
+ # Get last N lines for notification
+ n = self.config.notifications.output_lines
+ last_output = "\n".join(recent_lines[-n:]) if recent_lines else ""
+
+ # Create idle event
+ event = IdleEvent(
+ session_id=self.session.session_id,
+ tool_name=tool_name,
+ working_dir=self.session.working_dir,
+ last_output=last_output,
+ idle_seconds=idle_seconds,
+ detection_method=method,
+ )
+
+ # Update session
+ self.session.status = SessionStatus.IDLE
+ self.session.last_output = last_output
+ self.session.last_idle_event = event
+
+ # Intentionally no terminal output — notifications are macOS/server only.
+
+ # Fire macOS notification
+ if self.config.notifications.macos:
+ from jigai.notifier.macos import is_terminal_focused, notify_macos
+
+ if self.config.notifications.only_when_away and is_terminal_focused():
+ return # User is looking at a terminal — skip notification
+
+ subtitle = f"Session: {self.session.to_display_name()}"
+ body = _last_meaningful_line(last_output) if last_output else ""
+ if self.session.working_dir:
+ dir_short = _shorten_path(self.session.working_dir)
+ body = f"{body}\n{dir_short}" if body else dir_short
+
+ notify_macos(
+ title=f"{tool_name} is waiting",
+ message=body,
+ subtitle=subtitle,
+ sound=self.config.notifications.sound,
+ group=self.session.session_id if self.config.notifications.group_by_session else None,
+ )
+
+ # External callback (for server push)
+ if self._on_idle_event:
+ self._on_idle_event(event)
+
+ def _handle_exit(self, exit_code: int) -> None:
+ """Called when the child process exits."""
+ self._running = False
+ self.session.status = SessionStatus.STOPPED
+
+ def _timeout_checker(self) -> None:
+ """Background thread that periodically checks for timeout-based idle."""
+ while self._running:
+ time.sleep(1.0)
+ if self._running:
+ self.detector.check_timeout()
+
+ def run(self) -> int:
+ """Run the watcher. Blocks until the wrapped command exits."""
+ display_name = self.session.to_display_name()
+ console.print(
+ f"[bold green]▶ [JigAi][/bold green] "
+ f"Watching [cyan]{' '.join(self.command)}[/cyan] "
+ f"as [yellow]{display_name}[/yellow]"
+ )
+ console.print(
+ f" [dim]Working dir: {self.session.working_dir}[/dim]"
+ )
+ console.print(
+ f" [dim]Timeout: {self.registry.timeout_seconds}s | "
+ f"Cooldown: {self.registry.cooldown_seconds}s[/dim]"
+ )
+ console.print()
+
+ # Start timeout checker thread
+ self._running = True
+ self._timeout_thread = threading.Thread(target=self._timeout_checker, daemon=True)
+ self._timeout_thread.start()
+
+ # Run PTY proxy (blocks)
+ proxy = PtyProxy(
+ command=self.command,
+ on_output=self._handle_output,
+ on_exit=self._handle_exit,
+ )
+ self.session.pid = proxy.child_pid
+
+ try:
+ exit_code = proxy.run()
+ except KeyboardInterrupt:
+ proxy.stop()
+ exit_code = 130
+ finally:
+ self._running = False
+
+ return exit_code
+
+
+def _last_meaningful_line(text: str) -> str:
+ """
+ Return the last line with real readable content from a block of text.
+
+ AI tool TUIs output lots of box-drawing separators and prompt chars.
+ This skips those and strips decorative characters, returning only
+ lines with actual human-readable text.
+ """
+ import re
+ # Pure separator lines — skip entirely
+ _SEPARATOR_RE = re.compile(r"^[\s\u2500-\u257F\-=_|*~\u2014\u2013]+$")
+ # Decorative chars to strip from inside meaningful lines
+ _DECOR_RE = re.compile(r"[\u2500-\u257F\u2580-\u259F\u25A0-\u25FF\u2600-\u26FF●✻⚡✓►▶⚠\-─━╭╮╰╯│]")
+ # A line is only meaningful if it has 3+ consecutive letters after cleaning
+ _HAS_ALPHA = re.compile(r"[a-zA-Z]{3,}")
+
+ for line in reversed(text.split("\n")):
+ stripped = line.strip()
+ if not stripped or _SEPARATOR_RE.match(stripped):
+ continue
+ cleaned = _DECOR_RE.sub("", stripped).strip()
+ if _HAS_ALPHA.search(cleaned):
+ return cleaned
+ return ""
+
+
+def _shorten_path(path: str, max_len: int = 40) -> str:
+ """Shorten a path for display, replacing home dir with ~."""
+ home = os.path.expanduser("~")
+ if path.startswith(home):
+ path = "~" + path[len(home):]
+ if len(path) > max_len:
+ parts = path.split(os.sep)
+ if len(parts) > 3:
+ path = os.sep.join([parts[0], "...", *parts[-2:]])
+ return path
diff --git a/patterns/defaults.yaml b/patterns/defaults.yaml
new file mode 100644
index 0000000..22f5372
--- /dev/null
+++ b/patterns/defaults.yaml
@@ -0,0 +1,57 @@
+# JigAi — Built-in idle detection patterns
+# These patterns match the idle/prompt state of known AI coding tools.
+# Patterns are tested against each line of terminal output (ANSI codes stripped).
+#
+# To add your own patterns, create ~/.jigai/patterns.yaml
+# See: https://github.com/jigai/jigai#custom-patterns
+
+tools:
+ claude_code:
+ name: "Claude Code"
+ idle_patterns:
+ # Claude Code prompt patterns
+ - '>>\s*$' # Claude Code's >> prompt
+ - '^\s*>\s*$' # Simple > prompt
+ - '(?i)waiting for.*input'
+ - '(?i)what would you like'
+ - '(?i)how can i help'
+ - '\$\s*$' # Shell-like prompt after completion
+
+ codex:
+ name: "OpenAI Codex CLI"
+ idle_patterns:
+ - '(?i)codex>\s*$'
+ - '(?i)what would you like'
+ - '(?i)enter a command'
+ - '(?i)how can i help'
+
+ gemini_cli:
+ name: "Gemini CLI"
+ idle_patterns:
+ - '(?i)gemini>\s*$'
+ - '(?i)what would you like'
+ - '(?i)how can i help'
+
+ aider:
+ name: "Aider"
+ idle_patterns:
+ - '(?i)aider>\s*$'
+ - '(?i)what change.*would you like'
+
+ opencode:
+ name: "OpenCode"
+ idle_patterns:
+ - '(?i)opencode>\s*$'
+
+ generic:
+ name: "Generic AI Agent"
+ idle_patterns:
+ # Common patterns across many agents
+ - '(?i)press enter to continue'
+ - '(?i)\(y/n\)\s*$'
+ - '(?i)\[y/N\]\s*$'
+ - '(?i)\[Y/n\]\s*$'
+
+defaults:
+ timeout_seconds: 30
+ cooldown_seconds: 5
diff --git a/pyproject.toml b/pyproject.toml
new file mode 100644
index 0000000..6616c2e
--- /dev/null
+++ b/pyproject.toml
@@ -0,0 +1,80 @@
+[build-system]
+requires = ["setuptools>=68.0", "setuptools-scm>=8.0"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "jigai"
+version = "0.1.0"
+description = "জিগাই — Tool-agnostic terminal notification system for AI coding agents"
+readme = "README.md"
+license = { text = "MIT" }
+requires-python = ">=3.10"
+authors = [{ name = "Nafis Tiham", email = "nafistiham@gmail.com" }]
+keywords = [
+ "terminal", "notifications", "ai", "coding-agent",
+ "claude-code", "codex", "gemini", "aider", "productivity",
+]
+classifiers = [
+ "Development Status :: 3 - Alpha",
+ "Environment :: Console",
+ "Intended Audience :: Developers",
+ "License :: OSI Approved :: MIT License",
+ "Operating System :: MacOS",
+ "Programming Language :: Python :: 3",
+ "Programming Language :: Python :: 3.10",
+ "Programming Language :: Python :: 3.11",
+ "Programming Language :: Python :: 3.12",
+ "Programming Language :: Python :: 3.13",
+ "Topic :: Software Development :: Tools",
+ "Topic :: Utilities",
+]
+dependencies = [
+ "typer[all]>=0.9.0",
+ "fastapi>=0.104.0",
+ "uvicorn[standard]>=0.24.0",
+ "websockets>=12.0",
+ "zeroconf>=0.131.0",
+ "pyyaml>=6.0",
+ "rich>=13.0",
+ "pydantic>=2.0",
+]
+
+[project.optional-dependencies]
+dev = [
+ "pytest>=7.0",
+ "pytest-asyncio>=0.21",
+ "pytest-cov>=4.0",
+ "ruff>=0.1.0",
+ "mypy>=1.0",
+]
+
+[project.urls]
+Homepage = "https://github.com/nafistiham/jigai"
+Repository = "https://github.com/nafistiham/jigai"
+Issues = "https://github.com/nafistiham/jigai/issues"
+Changelog = "https://github.com/nafistiham/jigai/releases"
+
+[project.scripts]
+jigai = "jigai.cli:app"
+
+[tool.setuptools.packages.find]
+include = ["jigai*"]
+
+[tool.setuptools.package-data]
+"*" = ["patterns/*.yaml"]
+
+[tool.ruff]
+target-version = "py310"
+line-length = 100
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "N", "W", "UP"]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+asyncio_mode = "auto"
+
+[tool.mypy]
+python_version = "3.10"
+warn_return_any = true
+warn_unused_configs = true
diff --git a/tests/__init__.py b/tests/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/test_config.py b/tests/test_config.py
new file mode 100644
index 0000000..d6a2818
--- /dev/null
+++ b/tests/test_config.py
@@ -0,0 +1,51 @@
+"""Tests for configuration management."""
+
+import pytest
+
+from jigai.config import JigAiConfig, NotificationConfig, DetectionConfig, ServerConfig
+
+
+class TestJigAiConfig:
+ def test_defaults(self):
+ config = JigAiConfig()
+ assert config.server.port == 9384
+ assert config.server.bind == "0.0.0.0"
+ assert config.notifications.macos is True
+ assert config.notifications.sound == "Ping"
+ assert config.notifications.show_last_output is True
+ assert config.notifications.output_lines == 3
+ assert config.detection.timeout_seconds == 30
+ assert config.detection.cooldown_seconds == 5
+
+ def test_custom_values(self):
+ config = JigAiConfig(
+ server=ServerConfig(port=8080),
+ detection=DetectionConfig(timeout_seconds=60),
+ )
+ assert config.server.port == 8080
+ assert config.detection.timeout_seconds == 60
+
+ def test_notification_redact_patterns(self):
+ config = JigAiConfig()
+ assert len(config.notifications.redact_patterns) > 0
+
+ def test_serialization_roundtrip(self):
+ config = JigAiConfig()
+ data = config.model_dump()
+ restored = JigAiConfig(**data)
+ assert restored.server.port == config.server.port
+ assert restored.detection.timeout_seconds == config.detection.timeout_seconds
+
+
+class TestNotificationConfig:
+ def test_defaults(self):
+ nc = NotificationConfig()
+ assert nc.show_last_output is True
+ assert nc.output_lines == 3
+ assert nc.group_by_session is True
+
+ def test_custom_redact(self):
+ nc = NotificationConfig(
+ redact_patterns=[r"SECRET_\w+", r"token=\S+"]
+ )
+ assert len(nc.redact_patterns) == 2
diff --git a/tests/test_detector.py b/tests/test_detector.py
new file mode 100644
index 0000000..a85ddf4
--- /dev/null
+++ b/tests/test_detector.py
@@ -0,0 +1,287 @@
+"""Tests for the idle detection engine."""
+
+import time
+
+import pytest
+
+from jigai.watcher.detector import Detector, strip_ansi
+from jigai.watcher.patterns import PatternRegistry, ToolPattern, _compile_patterns
+
+
+# ── Helpers ─────────────────────────────────────────────────
+
+
+def make_registry(**kwargs) -> PatternRegistry:
+ """Create a minimal test registry."""
+ registry = PatternRegistry(
+ timeout_seconds=kwargs.get("timeout", 5),
+ cooldown_seconds=kwargs.get("cooldown", 0),
+ )
+ registry.tools["claude_code"] = ToolPattern(
+ name="Claude Code",
+ key="claude_code",
+ patterns=_compile_patterns([
+ r">>\s*$",
+ r"(?i)waiting for.*input",
+ ]),
+ )
+ registry.tools["codex"] = ToolPattern(
+ name="Codex",
+ key="codex",
+ patterns=_compile_patterns([
+ r"(?i)codex>\s*$",
+ ]),
+ )
+ return registry
+
+
+# ── Tests: strip_ansi ───────────────────────────────────────
+
+
+class TestStripAnsi:
+ def test_plain_text(self):
+ assert strip_ansi("hello world") == "hello world"
+
+ def test_color_codes(self):
+ assert strip_ansi("\x1b[32mgreen\x1b[0m") == "green"
+
+ def test_complex_sequences(self):
+ assert strip_ansi("\x1b[1;34mBold Blue\x1b[0m text") == "Bold Blue text"
+
+ def test_osc_sequences(self):
+ assert strip_ansi("\x1b]0;Window Title\x07rest") == "rest"
+
+ def test_empty_string(self):
+ assert strip_ansi("") == ""
+
+
+# ── Tests: Pattern Matching ─────────────────────────────────
+
+
+class TestPatternMatching:
+ def test_claude_prompt_match(self):
+ registry = make_registry()
+ tool = registry.tools["claude_code"]
+ assert tool.matches(">> ")
+ assert tool.matches(">>")
+
+ def test_claude_waiting_match(self):
+ registry = make_registry()
+ tool = registry.tools["claude_code"]
+ assert tool.matches("Waiting for your input")
+ assert tool.matches("waiting for user input")
+
+ def test_codex_prompt_match(self):
+ registry = make_registry()
+ tool = registry.tools["codex"]
+ assert tool.matches("codex> ")
+ assert tool.matches("Codex>")
+
+ def test_no_match(self):
+ registry = make_registry()
+ tool = registry.tools["claude_code"]
+ assert not tool.matches("Installing packages...")
+ assert not tool.matches("Running tests...")
+
+ def test_match_any(self):
+ registry = make_registry()
+ assert registry.match_any(">> ") == "claude_code"
+ assert registry.match_any("codex> ") == "codex"
+ assert registry.match_any("random output") is None
+
+
+# ── Tests: Detector ─────────────────────────────────────────
+
+
+class TestDetector:
+ def test_pattern_detection(self):
+ """Detector should fire on_idle when a pattern matches."""
+ registry = make_registry(cooldown=0)
+ events = []
+
+ def on_idle(method, tool_key, idle_seconds, recent):
+ events.append((method, tool_key))
+
+ detector = Detector(registry=registry, on_idle=on_idle)
+ detector.feed_line("Some normal output")
+ detector.feed_line("More output here")
+ detector.feed_line(">> ")
+
+ assert len(events) == 1
+ assert events[0] == ("pattern", "claude_code")
+
+ def test_tool_hint_prioritized(self):
+ """Detector with tool_hint should check that tool first."""
+ registry = make_registry(cooldown=0)
+ events = []
+
+ def on_idle(method, tool_key, idle_seconds, recent):
+ events.append(tool_key)
+
+ detector = Detector(
+ registry=registry, on_idle=on_idle, tool_hint="claude_code"
+ )
+ detector.feed_line(">> ")
+
+ assert events == ["claude_code"]
+
+ def test_cooldown_prevents_rapid_fire(self):
+ """Detector should respect cooldown between notifications."""
+ registry = make_registry(cooldown=10)
+ events = []
+
+ def on_idle(method, tool_key, idle_seconds, recent):
+ events.append(tool_key)
+
+ detector = Detector(registry=registry, on_idle=on_idle)
+ detector.feed_line(">> ")
+ detector.feed_line(">> ")
+ detector.feed_line(">> ")
+
+ # Only one event due to cooldown
+ assert len(events) == 1
+
+ def test_output_buffer(self):
+ """Detector should maintain a buffer of recent output."""
+ registry = make_registry(cooldown=0)
+ events = []
+ recent_lines = []
+
+ def on_idle(method, tool_key, idle_seconds, recent):
+ events.append(tool_key)
+ recent_lines.extend(recent)
+
+ detector = Detector(registry=registry, on_idle=on_idle)
+ detector.feed_line("Line 1")
+ detector.feed_line("Line 2")
+ detector.feed_line("Line 3")
+ detector.feed_line(">> ")
+
+ assert "Line 1" in recent_lines
+ assert "Line 2" in recent_lines
+ assert "Line 3" in recent_lines
+
+ def test_empty_lines_ignored(self):
+ """Empty lines should not trigger detection."""
+ registry = make_registry(cooldown=0)
+ events = []
+
+ def on_idle(method, tool_key, idle_seconds, recent):
+ events.append(tool_key)
+
+ detector = Detector(registry=registry, on_idle=on_idle)
+ detector.feed_line("")
+ detector.feed_line(" ")
+ detector.feed_line("\n")
+
+ assert len(events) == 0
+
+ def test_ansi_stripped_before_matching(self):
+ """ANSI codes should be stripped before pattern matching."""
+ registry = make_registry(cooldown=0)
+ events = []
+
+ def on_idle(method, tool_key, idle_seconds, recent):
+ events.append(tool_key)
+
+ detector = Detector(registry=registry, on_idle=on_idle)
+ detector.feed_line("\x1b[32m>> \x1b[0m")
+
+ assert len(events) == 1
+ assert events[0] == "claude_code"
+
+ def test_redaction(self):
+ """Sensitive data should be redacted in output buffer."""
+ registry = make_registry(cooldown=0)
+ recent_lines = []
+
+ def on_idle(method, tool_key, idle_seconds, recent):
+ recent_lines.extend(recent)
+
+ detector = Detector(registry=registry, on_idle=on_idle)
+ detector.set_redact_patterns([r"(?i)(token|password)=\S+"])
+
+ detector.feed_line("Setting token=abc123secret")
+ detector.feed_line("password=hunter2")
+ detector.feed_line(">> ")
+
+ # Check redacted content
+ assert any("[REDACTED]" in line for line in recent_lines)
+ assert not any("abc123secret" in line for line in recent_lines)
+ assert not any("hunter2" in line for line in recent_lines)
+
+ def test_get_recent_output(self):
+ """get_recent_output should return last N lines."""
+ registry = make_registry()
+ detector = Detector(
+ registry=registry, on_idle=lambda *args: None
+ )
+
+ for i in range(10):
+ detector.feed_line(f"Line {i}")
+
+ recent = detector.get_recent_output(3)
+ assert len(recent) == 3
+ assert recent[-1] == "Line 9"
+
+
+# ── Tests: Timeout Detection ───────────────────────────────
+
+
+class TestTimeoutDetection:
+ def test_timeout_triggers_after_silence(self):
+ """Timeout should trigger when no output for timeout_seconds."""
+ registry = make_registry(timeout=1, cooldown=0)
+ events = []
+
+ def on_idle(method, tool_key, idle_seconds, recent):
+ events.append(method)
+
+ detector = Detector(registry=registry, on_idle=on_idle)
+ detector.feed_line("Some output")
+
+ # Manually set last_output_time to simulate passage of time
+ detector.state.last_output_time = time.time() - 2
+
+ detector.check_timeout()
+
+ assert len(events) == 1
+ assert events[0] == "timeout"
+
+ def test_timeout_does_not_retrigger_while_idle(self):
+ """Timeout should not re-trigger while already idle."""
+ registry = make_registry(timeout=1, cooldown=0)
+ events = []
+
+ def on_idle(method, tool_key, idle_seconds, recent):
+ events.append(method)
+
+ detector = Detector(registry=registry, on_idle=on_idle)
+ detector.feed_line("Some output")
+ detector.state.last_output_time = time.time() - 2
+
+ detector.check_timeout()
+ detector.check_timeout()
+ detector.check_timeout()
+
+ # Only one event — is_idle prevents retrigger
+ assert len(events) == 1
+
+ def test_new_output_resets_idle(self):
+ """New output should reset the idle state."""
+ registry = make_registry(timeout=1, cooldown=0)
+ events = []
+
+ def on_idle(method, tool_key, idle_seconds, recent):
+ events.append(method)
+
+ detector = Detector(registry=registry, on_idle=on_idle)
+ detector.feed_line("Some output")
+ detector.state.last_output_time = time.time() - 2
+ detector.check_timeout()
+
+ assert len(events) == 1
+
+ # New output resets idle
+ detector.feed_line("New output arrived")
+ assert detector.state.is_idle is False
diff --git a/tests/test_models.py b/tests/test_models.py
new file mode 100644
index 0000000..7a187cb
--- /dev/null
+++ b/tests/test_models.py
@@ -0,0 +1,39 @@
+"""Tests for data models."""
+
+from jigai.models import IdleEvent, Session, SessionStatus
+
+
+class TestSession:
+ def test_default_session(self):
+ session = Session()
+ assert len(session.session_id) == 8
+ assert session.tool_name == "unknown"
+ assert session.status == SessionStatus.ACTIVE
+
+ def test_display_name(self):
+ session = Session(tool_name="Claude Code", session_id="abc123")
+ assert session.to_display_name() == "Claude Code-abc123"
+
+
+class TestIdleEvent:
+ def test_creation(self):
+ event = IdleEvent(
+ session_id="test123",
+ tool_name="Claude Code",
+ working_dir="/home/user/project",
+ last_output="Tests passed",
+ idle_seconds=5.2,
+ )
+ assert event.session_id == "test123"
+ assert event.tool_name == "Claude Code"
+ assert event.detection_method == "pattern"
+
+ def test_serialization(self):
+ event = IdleEvent(
+ session_id="test",
+ tool_name="test",
+ working_dir="/tmp",
+ )
+ data = event.model_dump()
+ assert "session_id" in data
+ assert "timestamp" in data
diff --git a/tests/test_patterns.py b/tests/test_patterns.py
new file mode 100644
index 0000000..7f3a357
--- /dev/null
+++ b/tests/test_patterns.py
@@ -0,0 +1,102 @@
+"""Tests for pattern loading and tool detection."""
+
+import pytest
+
+from jigai.watcher.patterns import (
+ PatternRegistry,
+ ToolPattern,
+ _compile_patterns,
+ detect_tool_from_command,
+ load_patterns,
+)
+
+
+class TestCompilePatterns:
+ def test_valid_patterns(self):
+ patterns = _compile_patterns([r"hello", r"\d+", r"^test$"])
+ assert len(patterns) == 3
+
+ def test_invalid_pattern_skipped(self):
+ patterns = _compile_patterns([r"valid", r"[invalid", r"also_valid"])
+ assert len(patterns) == 2
+
+ def test_empty_list(self):
+ assert _compile_patterns([]) == []
+
+
+class TestLoadPatterns:
+ def test_loads_builtin_patterns(self):
+ registry = load_patterns()
+ # Should have at least the built-in tools
+ assert "claude_code" in registry.tools
+ assert "codex" in registry.tools
+ assert "gemini_cli" in registry.tools
+
+ def test_builtin_patterns_compile(self):
+ registry = load_patterns()
+ for key, tool in registry.tools.items():
+ assert len(tool.patterns) > 0, f"{key} has no compiled patterns"
+
+ def test_defaults_loaded(self):
+ registry = load_patterns()
+ assert registry.timeout_seconds > 0
+ assert registry.cooldown_seconds >= 0
+
+
+class TestDetectToolFromCommand:
+ def test_detect_claude(self):
+ registry = load_patterns()
+ assert detect_tool_from_command(["claude"], registry) == "claude_code"
+
+ def test_detect_codex(self):
+ registry = load_patterns()
+ assert detect_tool_from_command(["codex"], registry) == "codex"
+
+ def test_detect_gemini(self):
+ registry = load_patterns()
+ assert detect_tool_from_command(["gemini"], registry) == "gemini_cli"
+
+ def test_detect_aider(self):
+ registry = load_patterns()
+ assert detect_tool_from_command(["aider"], registry) == "aider"
+
+ def test_unknown_command(self):
+ registry = load_patterns()
+ assert detect_tool_from_command(["python", "my_script.py"], registry) == "unknown"
+
+ def test_empty_command(self):
+ registry = load_patterns()
+ assert detect_tool_from_command([], registry) == "unknown"
+
+ def test_command_with_args(self):
+ registry = load_patterns()
+ assert detect_tool_from_command(
+ ["claude", "--model", "sonnet"], registry
+ ) == "claude_code"
+
+
+class TestPatternRegistry:
+ def test_match_any_returns_first_match(self):
+ registry = PatternRegistry()
+ registry.tools["tool_a"] = ToolPattern(
+ name="Tool A",
+ key="tool_a",
+ patterns=_compile_patterns([r"prompt_a>"]),
+ )
+ registry.tools["tool_b"] = ToolPattern(
+ name="Tool B",
+ key="tool_b",
+ patterns=_compile_patterns([r"prompt_b>"]),
+ )
+
+ assert registry.match_any("prompt_a> ") == "tool_a"
+ assert registry.match_any("prompt_b> ") == "tool_b"
+ assert registry.match_any("random text") is None
+
+ def test_get_tool_name(self):
+ registry = PatternRegistry()
+ registry.tools["test"] = ToolPattern(
+ name="Test Tool", key="test", patterns=[]
+ )
+ assert registry.get_tool_name("test") == "Test Tool"
+ assert registry.get_tool_name("nonexistent") == "nonexistent"