diff --git a/README.md b/README.md index 5b438e72..4fca5d5b 100644 --- a/README.md +++ b/README.md @@ -33,11 +33,11 @@

-> **Status:** v0.1.0 · macOS only · early alpha +> **Status:** v0.1.0 · macOS + Windows 11 · early alpha OpenChronicle gives AI agents a local, inspectable memory built from real screen and app context. -It runs on your Mac, captures structured context from what you're doing, and turns it into persistent Markdown memory: what you're working on, what you've decided, which tools you use, and which people or projects matter. +It runs on your Mac or Windows PC, captures structured context from what you're doing, and turns it into persistent Markdown memory: what you're working on, what you've decided, which tools you use, and which people or projects matter. Any agent that can call tools can use it. MCP clients work especially well today, but OpenChronicle is meant to be a general memory layer for tool-using agents - not something tied to one protocol, one model provider, or one app. @@ -120,7 +120,7 @@ The core idea is simple: ## What you get -* **Event-driven capture** from macOS AX events +* **Cross-platform capture** from macOS AX events or Windows UI Automation polling * **Session-aware memory writing** instead of noisy per-snapshot logs * **Human-readable Markdown memory** * **Local SQLite indexing** @@ -133,14 +133,22 @@ The core idea is simple: ## Install -Requires **macOS 13+** and **Xcode Command Line Tools** (`xcode-select --install`). +Requires either **macOS 13+** with Xcode Command Line Tools (`xcode-select --install`) or **Windows 11**. +macOS: ```bash git clone https://github.com/Einsia/OpenChronicle.git cd openchronicle bash install.sh ``` +Windows PowerShell: +```powershell +git clone https://github.com/Einsia/OpenChronicle.git +cd OpenChronicle +.\install.ps1 +``` + --- ## Run diff --git a/README_ADB_AGENT.md b/README_ADB_AGENT.md new file mode 100644 index 00000000..f6eb167c --- /dev/null +++ b/README_ADB_AGENT.md @@ -0,0 +1,191 @@ +# OpenChronicle ADB Agent + +This document explains the minimal Android ADB control layer added on top of +OpenChronicle. + +Architecture: + +```text +Codex / Claude Code / Hermes Agent + -> MCP tool call + -> OpenChronicle ADB Control MCP Server + -> adb + -> Android device + -> OpenChronicle event memory +``` + +The original OpenChronicle memory MCP server is unchanged. The ADB server is a +separate stdio MCP server started with: + +```powershell +uv run openchronicle adb-mcp +``` + +## Tools + +The ADB MCP server exposes these tools: + +- `adb_list_devices` +- `adb_screenshot` +- `adb_dump_ui` +- `adb_tap` +- `adb_swipe` +- `adb_input_text` +- `adb_keyevent` +- `adb_current_app` +- `adb_open_app` +- `adb_read_logcat` + +Every tool call appends an entry to OpenChronicle's daily +`event-YYYY-MM-DD.md` memory file. Screenshots are written under: + +```text +/adb/screenshots/ +``` + +## Safety Policy + +The ADB command runner denies high-risk operations before subprocess execution. + +Blocked by default: + +- `adb uninstall` +- `adb reboot` +- `adb root`, `adb unroot`, `adb remount` +- `adb disable-verity`, `adb enable-verity` +- `adb shell rm`, `adb shell rmdir` +- `adb shell pm clear` +- `adb shell pm uninstall` +- `adb shell cmd package clear` +- `adb shell cmd package uninstall` +- `adb shell settings put/delete/reset` +- `adb shell content delete` +- `adb shell su` +- `adb shell input keyevent POWER` +- `adb shell input keyevent 26` + +`adb_input_text` rejects shell metacharacters and records only text length in +memory, not the raw text. + +Agent operating rules: + +1. Call `adb_screenshot` or `adb_dump_ui` before `adb_tap`, `adb_swipe`, or + `adb_input_text`. +2. Stop and ask the user before payments, login passwords, SMS codes, privacy + grants, account changes, or destructive workflows. +3. Do not ask for a generic adb shell. Use the fixed tools only. + +## Connect a Phone + +1. On the Android phone, enable Developer options. +2. Enable USB debugging. +3. Connect the phone by USB. +4. Accept the RSA debugging prompt on the phone. +5. Verify from PowerShell: + +```powershell +E:\ai\product\.tooling\platform-tools\adb.exe devices -l +``` + +Expected output: + +```text +List of devices attached + device ... +``` + +If the device is `unauthorized`, unlock the phone and accept the USB debugging +prompt. If it is `offline`, unplug/replug USB or restart the adb server. + +## ADB Path Resolution + +The server looks for adb in this order: + +1. `OPENCHRONICLE_ADB_PATH` +2. `ADB_PATH` +3. `ANDROID_HOME/platform-tools` +4. `ANDROID_SDK_ROOT/platform-tools` +5. `PATH` +6. local `.tooling/platform-tools` or `.tool/platform-tools` directories in the + current working directory or one of its parents + +Windows 11 example: + +```powershell +$env:OPENCHRONICLE_ADB_PATH = "E:\ai\product\.tooling\platform-tools\adb.exe" +uv run openchronicle adb-mcp +``` + +WSL2 example using the Windows adb.exe: + +```bash +export OPENCHRONICLE_ADB_PATH=/mnt/e/ai/product/.tooling/platform-tools/adb.exe +uv run openchronicle adb-mcp +``` + +Native Linux adb inside WSL2 can also work, but USB needs to be attached to WSL +with `usbipd-win`; using Windows `adb.exe` is simpler for this MVP. + +## Codex / Claude / Hermes MCP Config + +Generic stdio MCP config: + +```json +{ + "mcpServers": { + "openchronicle-adb": { + "command": "uv", + "args": ["run", "openchronicle", "adb-mcp"] + } + } +} +``` + +Codex CLI can also register a stdio server from the repo: + +```powershell +codex mcp add openchronicle-adb -- uv run openchronicle adb-mcp +``` + +Claude Code can register the same stdio command: + +```powershell +claude mcp add openchronicle-adb -- uv run openchronicle adb-mcp +``` + +Keep the existing OpenChronicle memory MCP configured separately if you also +want memory search tools. Use `openchronicle adb-mcp` only for phone control. + +## Quick Self-Test + +Run these from the repository root: + +```powershell +E:\ai\product\.tooling\platform-tools\adb.exe devices -l +uv run pytest tests/test_adb_control.py +uv run openchronicle adb-mcp +``` + +MCP client smoke test: + +```python +import asyncio +from mcp import ClientSession, StdioServerParameters +from mcp.client.stdio import stdio_client + +async def main(): + params = StdioServerParameters( + command="uv", + args=["run", "openchronicle", "adb-mcp"], + ) + async with stdio_client(params) as (read, write): + async with ClientSession(read, write) as session: + await session.initialize() + tools = await session.list_tools() + print([tool.name for tool in tools.tools]) + +asyncio.run(main()) +``` + +To call the phone through MCP, use `adb_list_devices` first. That call should +return the visible device list and create an `event-YYYY-MM-DD.md` memory entry. diff --git a/docs/architecture.md b/docs/architecture.md index bb47b0db..17cdfa74 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -4,7 +4,7 @@ OpenChronicle is a single daemon that ingests capture events, compresses them th ```mermaid flowchart LR - W[mac-ax-watcher
Swift binary] + W[platform capture source
mac-ax-watcher · windows-uia-poller] subgraph capture [Capture Layer] direction TB @@ -60,7 +60,7 @@ A typical 5-minute flush window, showing how one AX event propagates through to ```mermaid sequenceDiagram - participant W as mac-ax-watcher + participant W as capture source participant S0 as S0 dispatcher participant S1 as S1 parser participant BUF as capture-buffer @@ -109,7 +109,7 @@ Defined in `src/openchronicle/daemon.py`. | Task | Purpose | |---|---| -| `capture` | Consumes `mac-ax-watcher` events, debounces, writes enriched JSON captures (incl. S1 fields) to `~/.openchronicle/capture-buffer/`. Heartbeat catches quiet periods. Also calls `SessionManager.on_event` on every capture so the session cutter sees the same signal. | +| `capture` | Consumes macOS `mac-ax-watcher` events or Windows UI Automation poll events, debounces, writes enriched JSON captures (incl. S1 fields) to the capture buffer. Heartbeat catches quiet periods. Also calls `SessionManager.on_event` on every capture so the session cutter sees the same signal. | | `timeline` | Every 60s scans for closed wall-clock windows (default 1 min) and runs the `timeline` LLM stage to normalize each window while preserving authored text verbatim. Cleans buffer files older than the newest block. | | `session` | Every `session.tick_seconds` (default 30), calls `SessionManager.check_cuts()` so idle-gap and timeout cuts fire even when the dispatcher is quiet. | | `flush` | Every `session.flush_minutes` (default 5, clamped to 5-min floor), runs the reducer incrementally over the active session's newly closed timeline blocks (~5 of them at defaults) and appends `[flush]`-tagged partial entries to today's event-daily. | @@ -133,8 +133,11 @@ Force-end is also called on daemon shutdown and on the 23:55 safety net, so a se ## On-disk state +macOS/Linux: `~/.openchronicle/` +Windows: `%LOCALAPPDATA%\OpenChronicle\` + ``` -~/.openchronicle/ +/ ├── config.toml # single source of truth for runtime config ├── .pid # daemon PID; absence ⇒ stopped ├── .paused # sentinel — capture skips while present @@ -163,10 +166,11 @@ src/openchronicle/ ├── cli.py # Typer entry point ├── daemon.py # Async task orchestration ├── config.py # TOML loader, per-stage ModelConfig inheritance -├── paths.py # ~/.openchronicle/* paths +├── paths.py # platform data-root paths ├── logger.py # Rotating file sinks per component ├── capture/ │ ├── watcher.py # Spawns mac-ax-watcher, parses JSONL +│ ├── windows_uia.py # Windows UI Automation provider + poller │ ├── event_dispatcher.py # Debounce / dedup / min-gap │ ├── ax_capture.py # One-shot mac-ax-helper invocation │ ├── ax_models.py # ax_tree_to_markdown, prune helpers diff --git a/docs/capture.md b/docs/capture.md index 3948d05c..36315f79 100644 --- a/docs/capture.md +++ b/docs/capture.md @@ -1,16 +1,18 @@ # Capture -Capture is the only layer that touches the outside world. It produces one JSON file per observation into `~/.openchronicle/capture-buffer/`; nothing above it ever talks to macOS directly. +Capture is the only layer that touches the outside world. It produces one JSON file per observation into the platform capture buffer (`~/.openchronicle/capture-buffer/` on macOS, `%LOCALAPPDATA%\OpenChronicle\capture-buffer\` on Windows); nothing above it talks to OS APIs directly. -## Two signal sources +## Platform signal sources **`mac-ax-watcher`** (primary, event-driven). A vendored Swift binary that subscribes to AX notifications across all running apps: window focus, value changes (typing), title changes, app activation. It emits one JSON object per event on stdout. The Python side reads that stream line-by-line in `capture/watcher.py` → `capture/event_dispatcher.py`. +**Windows UI Automation poller** (Windows 11). A Python polling source reads the foreground app/window through Win32 APIs, emits macOS-style event names (`AXApplicationActivated`, `AXFocusedWindowChanged`, `AXValueChanged`), and lets the existing dispatcher apply the same debounce, dedup, and min-gap rules. It does not install low-level hooks. + **Heartbeat timer** (fallback). Every `heartbeat_minutes` (default 10), the scheduler fires a capture even if no event arrived — so long idle periods leave a trail. Set `heartbeat_minutes = 0` to disable entirely (watcher-only); values `>0` are clamped to a 60-second floor. Both funnel into `capture_once` in `capture/scheduler.py`, which runs: -1. `ax_capture.capture_frontmost(focused_window_only=True)` — one-shot invocation of `mac-ax-helper` for the current window, pruned to `ax_depth` layers. +1. `ax_capture.capture_frontmost(focused_window_only=True)` — one-shot invocation of `mac-ax-helper` on macOS or Windows UI Automation on Windows, pruned to `ax_depth` layers. 2. `s1_parser.enrich()` — extracts `focused_element`, `visible_text`, and `url` from the AX tree (see [S1 fields](#s1-fields) below). 3. `screenshot.grab()` — unless `include_screenshot = false`. 4. `window_meta.active_window()` — app name, title, bundle_id via `NSRunningApplication`. @@ -39,9 +41,9 @@ On top of the time-based knobs, the scheduler compares each built capture agains This catches the case the time knobs can't: a screen that doesn't change (lock screen overnight, a paused video, an idle IDE) keeps generating AX events with the same content indefinitely. Without content-dedup those would both fill the buffer and keep the current session from ever idling out. Timestamps, triggers, and screenshots are excluded from the fingerprint so only meaningful changes count. -## AX depth — the #1 footgun +## AX / UIA depth — the #1 footgun -AX Trees for native Cocoa apps are shallow (5–15 layers). Electron apps (Claude Desktop, VS Code, Slack, Notion) nest user content 20–60 layers deep under chrome. +AX Trees for native Cocoa apps are shallow (5–15 layers). Electron apps (Claude Desktop, VS Code, Slack, Notion) and browser UIA trees can nest user content 20–60 layers deep under chrome. **Default `ax_depth = 100`** was chosen after diagnosing silent capture misses: a 90-second Claude Desktop conversation about an interview at 18:00 was producing captures where "18:00" appeared at character 5639 of the tree — past any reasonable prune limit. At depth 8, the tree contained only window chrome and sidebar headers; at depth 100, the full conversation was there. @@ -91,7 +93,7 @@ A 10×+ ratio means there's content past depth 30 you'd miss. `trigger` is `{"event_type": "heartbeat"}` for timer captures and `{"event_type": "manual"}` for `capture-once`. Screenshot is omitted entirely when `include_screenshot = false`. -Secure fields (password inputs) are replaced with `"[REDACTED]"` at the helper level — the Python side never sees them. +Secure fields (password inputs) are replaced with `"[REDACTED]"` at the macOS helper level. Windows UIA capture reads only the accessibility values exposed by the foreground app. ## S1 fields @@ -99,7 +101,7 @@ Ported from Einsia-Partner's `s1_collector`. These are what downstream LLM stage - **`focused_element`** — `{role, title, value, is_editable, value_length}` for the currently focused AX element. This is the user's cursor context: what they're typing into, which sidebar row is selected, etc. - **`visible_text`** — a length-capped markdown rendering of the AX tree (up to ~10 k chars). What the user is currently reading on screen. -- **`url`** — regex-extracted from `visible_text` when present; `null` otherwise. +- **`url`** — regex-extracted from browser address fields when present; `null` otherwise. Chrome and Edge are supported on Windows first. Screenshots live in the capture JSON but are **not** passed to the timeline / reducer / classifier prompts. They exist for future vision-model paths and for debugging. @@ -151,7 +153,7 @@ openchronicle rebuild-captures-index openchronicle pause ``` -Drops a `~/.openchronicle/.paused` sentinel. The watcher keeps streaming but `capture_once` short-circuits on sentinel presence. `resume` removes the sentinel. +Drops a `.paused` sentinel in the platform data root. The watcher/poller keeps running but `capture_once` short-circuits on sentinel presence. `resume` removes the sentinel. ## Smoke test diff --git a/docs/config.md b/docs/config.md index d6e5cb32..cf0bb2cb 100644 --- a/docs/config.md +++ b/docs/config.md @@ -1,6 +1,6 @@ # Configuration -Runtime config lives at `~/.openchronicle/config.toml` (or `$OPENCHRONICLE_ROOT/config.toml`). It's created with sensible defaults the first time you run `openchronicle status`. +Runtime config lives at `~/.openchronicle/config.toml` on macOS/Linux, `%LOCALAPPDATA%\OpenChronicle\config.toml` on Windows, or `$OPENCHRONICLE_ROOT/config.toml` when overridden. It's created with sensible defaults the first time you run `openchronicle status`. View the resolved config any time with: diff --git a/install.ps1 b/install.ps1 new file mode 100644 index 00000000..e28e97e7 --- /dev/null +++ b/install.ps1 @@ -0,0 +1,213 @@ +param( + [string]$Python = "3.12", + [string]$InstallHome = "", + [string]$BinDir = "", + [switch]$Start, + [switch]$NoPathUpdate +) + +$ErrorActionPreference = "Stop" + +$RootDir = Split-Path -Parent $MyInvocation.MyCommand.Path +if (-not $InstallHome) { + if (-not $env:LOCALAPPDATA) { + throw "LOCALAPPDATA is not set; pass -InstallHome explicitly." + } + $InstallHome = Join-Path $env:LOCALAPPDATA "OpenChronicle" +} +$VenvDir = Join-Path $InstallHome "venv" + +function Write-Log { + param([string]$Message) + Write-Host "[openchronicle-install] $Message" +} + +function Die { + param([string]$Message) + Write-Error "[openchronicle-install] Error: $Message" + exit 1 +} + +function Require-RepoRoot { + if (-not (Test-Path (Join-Path $RootDir "pyproject.toml"))) { + Die "run this script from the repository root" + } + if (-not (Test-Path (Join-Path $RootDir "src\openchronicle"))) { + Die "repository layout looks incomplete" + } +} + +function Check-Windows11 { + if ($PSVersionTable.Platform -and $PSVersionTable.Platform -ne "Win32NT") { + Die "install.ps1 supports Windows 11 only" + } + $os = Get-CimInstance Win32_OperatingSystem + $build = [int]$os.BuildNumber + if ($build -lt 22000) { + Die "Windows 11 required (found $($os.Caption), build $build)" + } +} + +function Resolve-Uv { + $cmd = Get-Command uv -ErrorAction SilentlyContinue + if ($cmd) { + return $cmd.Source + } + + Write-Log "uv not found; installing it" + Invoke-RestMethod https://astral.sh/uv/install.ps1 | Invoke-Expression + + $candidates = @( + (Join-Path $env:USERPROFILE ".local\bin\uv.exe"), + (Join-Path $env:USERPROFILE ".cargo\bin\uv.exe") + ) + foreach ($candidate in $candidates) { + if (Test-Path $candidate) { + $env:Path = "$(Split-Path -Parent $candidate);$env:Path" + return $candidate + } + } + Die "uv installation finished but uv.exe was not found" +} + +function Install-Package { + param([string]$UvBin) + + New-Item -ItemType Directory -Force -Path $InstallHome | Out-Null + if (Test-Path $VenvDir) { + Write-Log "removing existing virtualenv at $VenvDir" + Remove-Item -Recurse -Force $VenvDir + } + + Write-Log "installing Python $Python via uv if needed" + & $UvBin python install $Python + if ($LASTEXITCODE -ne 0) { + Die "failed to install Python $Python via uv" + } + + Write-Log "creating virtualenv at $VenvDir" + & $UvBin venv $VenvDir --python $Python + if ($LASTEXITCODE -ne 0) { + Die "failed to create virtualenv" + } + + $venvPython = Join-Path $VenvDir "Scripts\python.exe" + Write-Log "installing OpenChronicle into the virtualenv" + & $UvBin pip install --python $venvPython $RootDir + if ($LASTEXITCODE -ne 0) { + Die "failed to install OpenChronicle" + } + + $openchronicleExe = Join-Path $VenvDir "Scripts\openchronicle.exe" + if (-not (Test-Path $openchronicleExe)) { + Die "expected CLI not found at $openchronicleExe" + } + return $openchronicleExe +} + +function Resolve-BinDir { + if ($BinDir) { + New-Item -ItemType Directory -Force -Path $BinDir | Out-Null + return (Resolve-Path $BinDir).Path + } + $default = Join-Path $env:USERPROFILE ".local\bin" + New-Item -ItemType Directory -Force -Path $default | Out-Null + return $default +} + +function Quote-PowerShellString { + param([string]$Value) + return "'" + $Value.Replace("'", "''") + "'" +} + +function Install-Shims { + param( + [string]$OpenChronicleExe, + [string]$TargetBinDir + ) + + $ps1Path = Join-Path $TargetBinDir "openchronicle.ps1" + $cmdPath = Join-Path $TargetBinDir "openchronicle.cmd" + $quotedExe = Quote-PowerShellString $OpenChronicleExe + + @" +`$OpenChronicleBin = $quotedExe +& `$OpenChronicleBin @args +exit `$LASTEXITCODE +"@ | Set-Content -Encoding UTF8 $ps1Path + + @" +@echo off +"$OpenChronicleExe" %* +"@ | Set-Content -Encoding ASCII $cmdPath + + Write-Log "installed openchronicle shims at $TargetBinDir" +} + +function Ensure-Path { + param([string]$TargetBinDir) + $env:Path = "$TargetBinDir;$env:Path" + if ($NoPathUpdate) { + return + } + $userPath = [Environment]::GetEnvironmentVariable("Path", "User") + $parts = @() + if ($userPath) { + $parts = $userPath -split ";" + } + if ($parts -notcontains $TargetBinDir) { + $newPath = if ($userPath) { "$userPath;$TargetBinDir" } else { $TargetBinDir } + [Environment]::SetEnvironmentVariable("Path", $newPath, "User") + Write-Log "added $TargetBinDir to the user PATH" + } +} + +function Verify-Install { + param([string]$TargetBinDir) + $cli = Join-Path $TargetBinDir "openchronicle.cmd" + $oldMock = $env:OPENCHRONICLE_LLM_MOCK + $env:OPENCHRONICLE_LLM_MOCK = "1" + try { + & $cli status | Out-Null + if ($LASTEXITCODE -ne 0) { + Die "installation verification failed ('openchronicle status' did not succeed)" + } + } + finally { + $env:OPENCHRONICLE_LLM_MOCK = $oldMock + } +} + +function Print-Summary { + param([string]$TargetBinDir) + Write-Host "" + Write-Host "OpenChronicle installed successfully." + Write-Host "" + Write-Host "Data root : $InstallHome" + Write-Host "Virtualenv : $VenvDir" + Write-Host "CLI shim : $(Join-Path $TargetBinDir 'openchronicle.cmd')" + Write-Host "" + Write-Host "Next steps:" + Write-Host " openchronicle start" + Write-Host " openchronicle status" +} + +Require-RepoRoot +Check-Windows11 +$uvBin = Resolve-Uv +$openchronicleExe = Install-Package $uvBin +$targetBinDir = Resolve-BinDir +Install-Shims $openchronicleExe $targetBinDir +Ensure-Path $targetBinDir +Verify-Install $targetBinDir + +if ($Start) { + $cli = Join-Path $targetBinDir "openchronicle.cmd" + Write-Log "starting OpenChronicle daemon" + & $cli start + if ($LASTEXITCODE -ne 0) { + Die "openchronicle start failed" + } +} + +Print-Summary $targetBinDir diff --git a/pyproject.toml b/pyproject.toml index 0681e023..9f8e0e93 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,6 +12,7 @@ classifiers = [ "Environment :: Console", "License :: OSI Approved :: MIT License", "Operating System :: MacOS", + "Operating System :: Microsoft :: Windows", "Programming Language :: Python :: 3", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", @@ -26,6 +27,7 @@ dependencies = [ "Pillow>=10.0", "mcp>=1.0", "httpx[socks]>=0.27", + "uiautomation>=2.0; platform_system == 'Windows'", "tomli>=2.0; python_version < '3.11'", ] diff --git a/src/openchronicle/adb/__init__.py b/src/openchronicle/adb/__init__.py new file mode 100644 index 00000000..af778c93 --- /dev/null +++ b/src/openchronicle/adb/__init__.py @@ -0,0 +1,12 @@ +"""Android ADB control helpers for OpenChronicle.""" + +from .client import ADBClient, ADBCommandResult, ADBError, ADBNotFoundError +from .tools import ADBController + +__all__ = [ + "ADBClient", + "ADBCommandResult", + "ADBController", + "ADBError", + "ADBNotFoundError", +] diff --git a/src/openchronicle/adb/client.py b/src/openchronicle/adb/client.py new file mode 100644 index 00000000..c717799e --- /dev/null +++ b/src/openchronicle/adb/client.py @@ -0,0 +1,165 @@ +"""Small, safe ADB subprocess wrapper.""" + +from __future__ import annotations + +import os +import platform +import shutil +import subprocess +from collections.abc import Sequence +from dataclasses import dataclass +from pathlib import Path + +from . import safety + + +class ADBError(RuntimeError): + """Raised when an adb subprocess fails.""" + + def __init__( + self, + message: str, + *, + args: Sequence[str] | None = None, + returncode: int | None = None, + stdout: str | bytes = "", + stderr: str = "", + ) -> None: + super().__init__(message) + self.args_list = list(args or []) + self.returncode = returncode + self.stdout = stdout + self.stderr = stderr + + +class ADBNotFoundError(ADBError): + """Raised when no adb executable can be started.""" + + +@dataclass(frozen=True) +class ADBCommandResult: + args: list[str] + returncode: int + stdout: str | bytes + stderr: str + + @property + def stdout_text(self) -> str: + if isinstance(self.stdout, bytes): + return self.stdout.decode("utf-8", errors="replace") + return self.stdout + + @property + def stdout_bytes(self) -> bytes: + if isinstance(self.stdout, bytes): + return self.stdout + return self.stdout.encode("utf-8") + + +def _tooling_candidates() -> list[Path]: + names = ("adb.exe", "adb") if platform.system() == "Windows" else ("adb", "adb.exe") + layouts = ( + (".tooling", "platform-tools"), + (".tooling", "android-sdk", "platform-tools"), + (".tool", "platform-tools"), + (".tool", "android-sdk", "platform-tools"), + ) + candidates: list[Path] = [] + for parent in (Path.cwd(), *Path.cwd().parents): + for layout in layouts: + base = parent.joinpath(*layout) + for name in names: + candidates.append(base / name) + return candidates + + +def find_adb_path() -> str: + """Resolve adb from env, PATH, or a local .tooling/.tool platform-tools dir.""" + for env_name in ("OPENCHRONICLE_ADB_PATH", "ADB_PATH"): + value = os.environ.get(env_name) + if value: + return str(Path(value).expanduser()) + + for env_name in ("ANDROID_HOME", "ANDROID_SDK_ROOT"): + value = os.environ.get(env_name) + if not value: + continue + sdk = Path(value).expanduser() + for name in ("adb.exe", "adb"): + candidate = sdk / "platform-tools" / name + if candidate.exists(): + return str(candidate) + + for name in ("adb.exe", "adb"): + found = shutil.which(name) + if found: + return found + + for candidate in _tooling_candidates(): + if candidate.exists(): + return str(candidate) + + return "adb" + + +class ADBClient: + """Run adb commands through a single safety gate.""" + + def __init__(self, adb_path: str | None = None) -> None: + self.adb_path = adb_path or find_adb_path() + + def command_for_display(self, args: Sequence[str], device_id: str | None = None) -> list[str]: + cmd = [self.adb_path] + if device_id: + cmd.extend(["-s", device_id]) + cmd.extend(str(a) for a in args) + return cmd + + def run( + self, + args: Sequence[str], + *, + device_id: str | None = None, + timeout: float = 30.0, + binary: bool = False, + check: bool = True, + ) -> ADBCommandResult: + safety.assert_safe([str(a) for a in args]) + cmd = self.command_for_display(args, device_id) + try: + completed = subprocess.run( + cmd, + capture_output=True, + check=False, + timeout=timeout, + text=not binary, + encoding=None if binary else "utf-8", + errors=None if binary else "replace", + ) + except FileNotFoundError as exc: + raise ADBNotFoundError( + f"adb executable not found: {self.adb_path!r}. Set OPENCHRONICLE_ADB_PATH.", + args=cmd, + ) from exc + except subprocess.TimeoutExpired as exc: + raise ADBError(f"adb command timed out after {timeout:g}s", args=cmd) from exc + + stderr = completed.stderr + if isinstance(stderr, bytes): + stderr = stderr.decode("utf-8", errors="replace") + result = ADBCommandResult( + args=cmd, + returncode=completed.returncode, + stdout=completed.stdout, + stderr=stderr or "", + ) + if check and result.returncode != 0: + message = (result.stderr or result.stdout_text or "adb command failed").strip() + raise ADBError( + message, + args=cmd, + returncode=result.returncode, + stdout=result.stdout, + stderr=result.stderr, + ) + return result diff --git a/src/openchronicle/adb/memory.py b/src/openchronicle/adb/memory.py new file mode 100644 index 00000000..013e5620 --- /dev/null +++ b/src/openchronicle/adb/memory.py @@ -0,0 +1,89 @@ +"""OpenChronicle memory adapter for ADB control events.""" + +from __future__ import annotations + +import contextlib +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any + +from .. import paths +from ..store import entries as entries_mod +from ..store import files as files_mod +from ..store import fts + + +def _truncate(value: str, limit: int = 1200) -> str: + text = value.strip() + if len(text) <= limit: + return text + return text[:limit].rstrip() + "\n... [truncated]" + + +def _event_daily_name(now: datetime) -> str: + return f"event-{now.strftime('%Y-%m-%d')}.md" + + +def _ensure_event_file(conn, name: str, *, day: str) -> None: + if files_mod.memory_path(name).exists(): + return + with contextlib.suppress(FileExistsError): + entries_mod.create_file( + conn, + name=name, + description=( + f"Activity log for {day}, including OpenChronicle sessions and " + "Android ADB agent operations." + ), + tags=["event", "session", "daily", "adb"], + ) + + +@dataclass +class ADBMemoryRecorder: + """Append every ADB tool attempt to the daily OpenChronicle event file.""" + + preview_limit: int = 1200 + default_tags: list[str] = field(default_factory=lambda: ["adb", "android"]) + + def record( + self, + *, + tool_name: str, + status: str, + device_id: str | None, + command: list[str], + summary: str, + params: dict[str, Any] | None = None, + output_preview: str = "", + artifact_path: str = "", + error: str = "", + ) -> str: + paths.ensure_dirs() + now = datetime.now().astimezone() + name = _event_daily_name(now) + command_text = " ".join(command) + + body_parts = [ + f"**ADB tool {tool_name}** ({now.strftime('%H:%M')})", + "", + f"- Status: {status}", + f"- Device: {device_id or 'auto'}", + f"- Command: `{command_text}`", + f"- Summary: {summary}", + ] + if artifact_path: + body_parts.append(f"- Artifact: `{artifact_path}`") + if params: + safe_params = {k: v for k, v in params.items() if v not in (None, "")} + if safe_params: + body_parts.append(f"- Params: `{safe_params}`") + if output_preview: + body_parts.extend(["", "Output preview:", "```text", _truncate(output_preview, self.preview_limit), "```"]) + if error: + body_parts.extend(["", "Error:", "```text", _truncate(error, self.preview_limit), "```"]) + + tags = [*self.default_tags, f"tool:{tool_name}"] + with fts.cursor() as conn: + _ensure_event_file(conn, name, day=now.strftime("%Y-%m-%d")) + return entries_mod.append_entry(conn, name=name, content="\n".join(body_parts), tags=tags) diff --git a/src/openchronicle/adb/safety.py b/src/openchronicle/adb/safety.py new file mode 100644 index 00000000..14a59434 --- /dev/null +++ b/src/openchronicle/adb/safety.py @@ -0,0 +1,138 @@ +"""Safety policy for Android ADB commands. + +The ADB MCP server intentionally exposes only a small set of fixed tools, but +this module is still called by the command runner so future helpers cannot +accidentally bypass the same denylist. +""" + +from __future__ import annotations + +import shlex +from collections.abc import Sequence + + +class ADBSafetyError(ValueError): + """Raised when an ADB command violates the local safety policy.""" + + +_BLOCKED_TOP_LEVEL = { + "disable-verity", + "enable-verity", + "reboot", + "remount", + "root", + "sideload", + "uninstall", + "unroot", +} + +_BLOCKED_SHELL_COMMANDS = { + "reboot", + "rm", + "rmdir", + "su", + "wipe", +} + +_BLOCKED_SEQUENCES = ( + ("cmd", "package", "clear"), + ("cmd", "package", "uninstall"), + ("content", "delete"), + ("pm", "clear"), + ("pm", "uninstall"), + ("settings", "delete"), + ("settings", "put"), + ("settings", "reset"), +) + +_BLOCKED_KEYEVENTS = { + "26", # KEYCODE_POWER + "223", # KEYCODE_SLEEP + "224", # KEYCODE_WAKEUP + "keycode_power", + "keycode_sleep", + "keycode_wakeup", + "power", + "sleep", + "wakeup", +} + + +def _split_token(token: str) -> list[str]: + try: + return shlex.split(token) + except ValueError: + return [token] + + +def _flatten(args: Sequence[str]) -> list[str]: + out: list[str] = [] + for arg in args: + text = str(arg).strip() + if not text: + continue + out.extend(_split_token(text)) + return [part.lower() for part in out if part] + + +def _contains_sequence(tokens: Sequence[str], sequence: Sequence[str]) -> bool: + if not sequence or len(sequence) > len(tokens): + return False + last_start = len(tokens) - len(sequence) + return any(tuple(tokens[i : i + len(sequence)]) == tuple(sequence) for i in range(last_start + 1)) + + +def assert_safe(args: Sequence[str]) -> None: + """Reject destructive or privilege-escalating ADB commands. + + Blocked categories: + - device reboot/root/remount/verity changes + - app uninstall or app data clearing + - shell deletion commands + - system settings mutation + - POWER/SLEEP/WAKEUP keyevents + """ + tokens = _flatten(args) + if not tokens: + raise ADBSafetyError("empty adb command is not allowed") + + for token in tokens: + if token in _BLOCKED_TOP_LEVEL: + raise ADBSafetyError(f"blocked high-risk adb command: {token}") + + shell_tokens = tokens + if "shell" in tokens: + shell_tokens = tokens[tokens.index("shell") + 1 :] + + for token in shell_tokens: + if token in _BLOCKED_SHELL_COMMANDS: + raise ADBSafetyError(f"blocked high-risk adb shell command: {token}") + + for sequence in _BLOCKED_SEQUENCES: + if _contains_sequence(shell_tokens, sequence): + raise ADBSafetyError("blocked high-risk adb shell command: " + " ".join(sequence)) + + if _contains_sequence(shell_tokens, ("input", "keyevent")): + idx = shell_tokens.index("keyevent") + if idx + 1 < len(shell_tokens): + key = shell_tokens[idx + 1].removeprefix("keycode_") + original = shell_tokens[idx + 1] + if original in _BLOCKED_KEYEVENTS or key in _BLOCKED_KEYEVENTS: + raise ADBSafetyError(f"blocked high-risk keyevent: {shell_tokens[idx + 1]}") + + +def validate_input_text(text: str) -> str: + """Return Android input-text syntax for safe text. + + `adb shell input text` runs through the device shell. Reject shell + metacharacters instead of trying to quote arbitrary text across host and + device shells. Spaces are encoded as `%s`, which is Android's input syntax. + """ + if text == "": + raise ADBSafetyError("adb_input_text requires non-empty text") + blocked = set("\r\n;&|<>`$(){}[]\\\"'") + bad = sorted({ch for ch in text if ch in blocked}) + if bad: + shown = " ".join(repr(ch) for ch in bad) + raise ADBSafetyError(f"adb_input_text rejected shell metacharacter(s): {shown}") + return text.replace(" ", "%s") diff --git a/src/openchronicle/adb/tools.py b/src/openchronicle/adb/tools.py new file mode 100644 index 00000000..bce36fd5 --- /dev/null +++ b/src/openchronicle/adb/tools.py @@ -0,0 +1,554 @@ +"""High-level Android ADB operations exposed through MCP.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from datetime import datetime +from pathlib import Path +from typing import Any + +from .. import paths +from . import safety +from .client import ADBClient, ADBError +from .memory import ADBMemoryRecorder + +_PACKAGE_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_]*(?:\.[A-Za-z][A-Za-z0-9_]*)+$") +_ACTIVITY_RE = re.compile(r"^[A-Za-z0-9_.$/]+$") +_COMPONENT_RE = re.compile(r"(?P[A-Za-z0-9_.$]+)/(?P[A-Za-z0-9_.$]+)") + + +def _parse_devices(output: str) -> list[dict[str, Any]]: + devices: list[dict[str, Any]] = [] + for raw_line in output.splitlines()[1:]: + line = raw_line.strip() + if not line: + continue + parts = line.split() + if len(parts) < 2: + continue + qualifiers: dict[str, str] = {} + for item in parts[2:]: + if ":" in item: + key, value = item.split(":", 1) + qualifiers[key] = value + devices.append( + { + "serial": parts[0], + "state": parts[1], + "qualifiers": qualifiers, + "raw": line, + } + ) + return devices + + +def _artifact_dir(name: str) -> Path: + path = paths.root() / "adb" / name + path.mkdir(parents=True, exist_ok=True) + return path + + +def _require_non_negative_int(name: str, value: int) -> int: + integer = int(value) + if integer < 0: + raise safety.ADBSafetyError(f"{name} must be non-negative") + if integer > 10000: + raise safety.ADBSafetyError(f"{name} is implausibly large: {integer}") + return integer + + +def _bounded_lines(lines: int) -> int: + value = int(lines) + if value < 1: + return 1 + return min(value, 2000) + + +def _command_with_redacted_tail(command: list[str]) -> list[str]: + if not command: + return command + return [*command[:-1], ""] + + +def _parse_component(text: str) -> dict[str, str]: + match = _COMPONENT_RE.search(text) + if not match: + return {"package": "", "activity": "", "raw": text.strip()} + return { + "package": match.group("package"), + "activity": match.group("activity"), + "raw": text.strip(), + } + + +@dataclass +class ADBController: + """ADB operation facade that always records into OpenChronicle memory.""" + + client: ADBClient = field(default_factory=ADBClient) + recorder: ADBMemoryRecorder = field(default_factory=ADBMemoryRecorder) + + def _record_success( + self, + *, + tool_name: str, + device_id: str | None, + command: list[str], + summary: str, + params: dict[str, Any] | None = None, + output_preview: str = "", + artifact_path: str = "", + ) -> str: + return self.recorder.record( + tool_name=tool_name, + status="ok", + device_id=device_id, + command=command, + summary=summary, + params=params, + output_preview=output_preview, + artifact_path=artifact_path, + ) + + def _record_failure( + self, + *, + tool_name: str, + status: str, + device_id: str | None, + command: list[str], + summary: str, + params: dict[str, Any] | None = None, + error: str = "", + ) -> str: + return self.recorder.record( + tool_name=tool_name, + status=status, + device_id=device_id, + command=command, + summary=summary, + params=params, + error=error, + ) + + def _adb_error_payload( + self, + *, + tool_name: str, + device_id: str | None, + command: list[str], + params: dict[str, Any] | None, + exc: ADBError, + ) -> dict[str, Any]: + entry_id = self._record_failure( + tool_name=tool_name, + status="error", + device_id=device_id, + command=command, + summary=f"{tool_name} failed", + params=params, + error=str(exc), + ) + return { + "ok": False, + "tool": tool_name, + "error": str(exc), + "returncode": exc.returncode, + "memory_entry_id": entry_id, + } + + def _blocked_payload( + self, + *, + tool_name: str, + device_id: str | None, + command: list[str], + params: dict[str, Any] | None, + exc: safety.ADBSafetyError, + ) -> dict[str, Any]: + entry_id = self._record_failure( + tool_name=tool_name, + status="blocked", + device_id=device_id, + command=command, + summary=f"{tool_name} blocked by safety policy", + params=params, + error=str(exc), + ) + return { + "ok": False, + "tool": tool_name, + "blocked": True, + "error": str(exc), + "memory_entry_id": entry_id, + } + + def list_devices(self) -> dict[str, Any]: + tool = "adb_list_devices" + command = ["devices", "-l"] + try: + result = self.client.run(command, timeout=15) + except safety.ADBSafetyError as exc: + return self._blocked_payload( + tool_name=tool, device_id=None, command=command, params=None, exc=exc + ) + except ADBError as exc: + return self._adb_error_payload( + tool_name=tool, device_id=None, command=command, params=None, exc=exc + ) + devices = _parse_devices(result.stdout_text) + entry_id = self._record_success( + tool_name=tool, + device_id=None, + command=self.client.command_for_display(command), + summary=f"Listed {len(devices)} Android device(s).", + output_preview=result.stdout_text, + ) + return {"ok": True, "tool": tool, "count": len(devices), "devices": devices, "memory_entry_id": entry_id} + + def screenshot(self, device_id: str | None = None) -> dict[str, Any]: + tool = "adb_screenshot" + command = ["exec-out", "screencap", "-p"] + params = {"device_id": device_id} + display = self.client.command_for_display(command, device_id) + try: + result = self.client.run(command, device_id=device_id, timeout=20, binary=True) + except safety.ADBSafetyError as exc: + return self._blocked_payload( + tool_name=tool, device_id=device_id, command=display, params=params, exc=exc + ) + except ADBError as exc: + return self._adb_error_payload( + tool_name=tool, device_id=device_id, command=display, params=params, exc=exc + ) + + now = datetime.now().astimezone() + out_path = _artifact_dir("screenshots") / f"{now.strftime('%Y%m%d-%H%M%S-%f')}.png" + out_path.write_bytes(result.stdout_bytes) + entry_id = self._record_success( + tool_name=tool, + device_id=device_id, + command=display, + summary=f"Captured Android screenshot ({len(result.stdout_bytes)} bytes).", + params=params, + artifact_path=str(out_path), + ) + return { + "ok": True, + "tool": tool, + "path": str(out_path), + "bytes": len(result.stdout_bytes), + "memory_entry_id": entry_id, + } + + def dump_ui(self, device_id: str | None = None) -> dict[str, Any]: + tool = "adb_dump_ui" + dump_path = "/sdcard/window.xml" + command = ["shell", "uiautomator", "dump", dump_path] + cat_command = ["exec-out", "cat", dump_path] + params = {"device_id": device_id} + display = self.client.command_for_display(command, device_id) + try: + dump_result = self.client.run(command, device_id=device_id, timeout=20) + xml_result = self.client.run(cat_command, device_id=device_id, timeout=20) + except safety.ADBSafetyError as exc: + return self._blocked_payload( + tool_name=tool, device_id=device_id, command=display, params=params, exc=exc + ) + except ADBError as exc: + return self._adb_error_payload( + tool_name=tool, device_id=device_id, command=display, params=params, exc=exc + ) + xml = xml_result.stdout_text.strip() + preview = dump_result.stdout_text + "\n" + xml[:1000] + entry_id = self._record_success( + tool_name=tool, + device_id=device_id, + command=display, + summary=f"Dumped Android UI XML ({len(xml)} characters).", + params=params, + output_preview=preview, + ) + return {"ok": True, "tool": tool, "xml": xml, "memory_entry_id": entry_id} + + def tap(self, x: int, y: int, device_id: str | None = None) -> dict[str, Any]: + tool = "adb_tap" + params = {"x": x, "y": y, "device_id": device_id} + try: + sx = _require_non_negative_int("x", x) + sy = _require_non_negative_int("y", y) + command = ["shell", "input", "tap", str(sx), str(sy)] + display = self.client.command_for_display(command, device_id) + result = self.client.run(command, device_id=device_id, timeout=10) + except safety.ADBSafetyError as exc: + command = ["shell", "input", "tap", str(x), str(y)] + display = self.client.command_for_display(command, device_id) + return self._blocked_payload( + tool_name=tool, device_id=device_id, command=display, params=params, exc=exc + ) + except ADBError as exc: + return self._adb_error_payload( + tool_name=tool, device_id=device_id, command=display, params=params, exc=exc + ) + entry_id = self._record_success( + tool_name=tool, + device_id=device_id, + command=display, + summary=f"Tapped Android screen at ({sx}, {sy}).", + params=params, + output_preview=result.stdout_text, + ) + return {"ok": True, "tool": tool, "x": sx, "y": sy, "memory_entry_id": entry_id} + + def swipe( + self, + x1: int, + y1: int, + x2: int, + y2: int, + duration_ms: int = 300, + device_id: str | None = None, + ) -> dict[str, Any]: + tool = "adb_swipe" + params = { + "x1": x1, + "y1": y1, + "x2": x2, + "y2": y2, + "duration_ms": duration_ms, + "device_id": device_id, + } + try: + sx1 = _require_non_negative_int("x1", x1) + sy1 = _require_non_negative_int("y1", y1) + sx2 = _require_non_negative_int("x2", x2) + sy2 = _require_non_negative_int("y2", y2) + duration = min(max(int(duration_ms), 0), 60000) + command = [ + "shell", + "input", + "swipe", + str(sx1), + str(sy1), + str(sx2), + str(sy2), + str(duration), + ] + display = self.client.command_for_display(command, device_id) + result = self.client.run(command, device_id=device_id, timeout=15) + except safety.ADBSafetyError as exc: + command = ["shell", "input", "swipe", str(x1), str(y1), str(x2), str(y2), str(duration_ms)] + display = self.client.command_for_display(command, device_id) + return self._blocked_payload( + tool_name=tool, device_id=device_id, command=display, params=params, exc=exc + ) + except ADBError as exc: + return self._adb_error_payload( + tool_name=tool, device_id=device_id, command=display, params=params, exc=exc + ) + entry_id = self._record_success( + tool_name=tool, + device_id=device_id, + command=display, + summary=f"Swiped Android screen from ({sx1}, {sy1}) to ({sx2}, {sy2}).", + params=params, + output_preview=result.stdout_text, + ) + return { + "ok": True, + "tool": tool, + "x1": sx1, + "y1": sy1, + "x2": sx2, + "y2": sy2, + "duration_ms": duration, + "memory_entry_id": entry_id, + } + + def input_text(self, text: str, device_id: str | None = None) -> dict[str, Any]: + tool = "adb_input_text" + params = {"text_length": len(text), "device_id": device_id} + try: + encoded = safety.validate_input_text(text) + command = ["shell", "input", "text", encoded] + display = self.client.command_for_display(_command_with_redacted_tail(command), device_id) + result = self.client.run(command, device_id=device_id, timeout=10) + except safety.ADBSafetyError as exc: + command = ["shell", "input", "text", ""] + display = self.client.command_for_display(command, device_id) + return self._blocked_payload( + tool_name=tool, device_id=device_id, command=display, params=params, exc=exc + ) + except ADBError as exc: + return self._adb_error_payload( + tool_name=tool, device_id=device_id, command=display, params=params, exc=exc + ) + entry_id = self._record_success( + tool_name=tool, + device_id=device_id, + command=display, + summary=f"Entered {len(text)} character(s) of text on Android device.", + params=params, + output_preview=result.stdout_text, + ) + return {"ok": True, "tool": tool, "text_length": len(text), "memory_entry_id": entry_id} + + def keyevent(self, keyevent: str | int, device_id: str | None = None) -> dict[str, Any]: + tool = "adb_keyevent" + key = str(keyevent).strip() + params = {"keyevent": key, "device_id": device_id} + command = ["shell", "input", "keyevent", key] + display = self.client.command_for_display(command, device_id) + try: + if not key: + raise safety.ADBSafetyError("keyevent is required") + result = self.client.run(command, device_id=device_id, timeout=10) + except safety.ADBSafetyError as exc: + return self._blocked_payload( + tool_name=tool, device_id=device_id, command=display, params=params, exc=exc + ) + except ADBError as exc: + return self._adb_error_payload( + tool_name=tool, device_id=device_id, command=display, params=params, exc=exc + ) + entry_id = self._record_success( + tool_name=tool, + device_id=device_id, + command=display, + summary=f"Sent Android keyevent {key}.", + params=params, + output_preview=result.stdout_text, + ) + return {"ok": True, "tool": tool, "keyevent": key, "memory_entry_id": entry_id} + + def current_app(self, device_id: str | None = None) -> dict[str, Any]: + tool = "adb_current_app" + command = ["shell", "dumpsys", "window"] + params = {"device_id": device_id} + display = self.client.command_for_display(command, device_id) + try: + result = self.client.run(command, device_id=device_id, timeout=20) + except safety.ADBSafetyError as exc: + return self._blocked_payload( + tool_name=tool, device_id=device_id, command=display, params=params, exc=exc + ) + except ADBError as exc: + return self._adb_error_payload( + tool_name=tool, device_id=device_id, command=display, params=params, exc=exc + ) + + raw_line = "" + for line in result.stdout_text.splitlines(): + if "mCurrentFocus" in line or "mFocusedApp" in line: + raw_line = line.strip() + break + parsed = _parse_component(raw_line) + entry_id = self._record_success( + tool_name=tool, + device_id=device_id, + command=display, + summary=f"Read current Android app: {parsed.get('package') or 'unknown'}.", + params=params, + output_preview=raw_line, + ) + return {"ok": True, "tool": tool, **parsed, "memory_entry_id": entry_id} + + def open_app( + self, + package_name: str, + activity: str | None = None, + device_id: str | None = None, + ) -> dict[str, Any]: + tool = "adb_open_app" + params = {"package_name": package_name, "activity": activity, "device_id": device_id} + try: + if not _PACKAGE_RE.match(package_name): + raise safety.ADBSafetyError(f"invalid Android package name: {package_name!r}") + if activity: + if not _ACTIVITY_RE.match(activity): + raise safety.ADBSafetyError(f"invalid Android activity name: {activity!r}") + component = activity if "/" in activity else f"{package_name}/{activity}" + command = ["shell", "am", "start", "-n", component] + else: + command = [ + "shell", + "monkey", + "-p", + package_name, + "-c", + "android.intent.category.LAUNCHER", + "1", + ] + display = self.client.command_for_display(command, device_id) + result = self.client.run(command, device_id=device_id, timeout=20) + except safety.ADBSafetyError as exc: + command = ["shell", "monkey", "-p", package_name, "-c", "android.intent.category.LAUNCHER", "1"] + display = self.client.command_for_display(command, device_id) + return self._blocked_payload( + tool_name=tool, device_id=device_id, command=display, params=params, exc=exc + ) + except ADBError as exc: + return self._adb_error_payload( + tool_name=tool, device_id=device_id, command=display, params=params, exc=exc + ) + entry_id = self._record_success( + tool_name=tool, + device_id=device_id, + command=display, + summary=f"Opened Android app {package_name}.", + params=params, + output_preview=result.stdout_text, + ) + return { + "ok": True, + "tool": tool, + "package_name": package_name, + "activity": activity or "", + "memory_entry_id": entry_id, + } + + def read_logcat( + self, + lines: int = 200, + filter_expr: str | None = None, + device_id: str | None = None, + ) -> dict[str, Any]: + tool = "adb_read_logcat" + line_count = _bounded_lines(lines) + params = {"lines": line_count, "filter_expr": filter_expr, "device_id": device_id} + command = ["logcat", "-d", "-t", str(line_count)] + if filter_expr: + if not re.match(r"^[A-Za-z0-9_.*:\-\s]+$", filter_expr): + exc = safety.ADBSafetyError("filter_expr contains unsupported characters") + display = self.client.command_for_display(command, device_id) + return self._blocked_payload( + tool_name=tool, device_id=device_id, command=display, params=params, exc=exc + ) + command.extend(filter_expr.split()) + display = self.client.command_for_display(command, device_id) + try: + result = self.client.run(command, device_id=device_id, timeout=30) + except safety.ADBSafetyError as exc: + return self._blocked_payload( + tool_name=tool, device_id=device_id, command=display, params=params, exc=exc + ) + except ADBError as exc: + return self._adb_error_payload( + tool_name=tool, device_id=device_id, command=display, params=params, exc=exc + ) + entry_id = self._record_success( + tool_name=tool, + device_id=device_id, + command=display, + summary=f"Read last {line_count} Android logcat line(s).", + params=params, + output_preview=result.stdout_text, + ) + return { + "ok": True, + "tool": tool, + "lines": line_count, + "logcat": result.stdout_text, + "memory_entry_id": entry_id, + } diff --git a/src/openchronicle/capture/__init__.py b/src/openchronicle/capture/__init__.py index 0edaea18..1fde4657 100644 --- a/src/openchronicle/capture/__init__.py +++ b/src/openchronicle/capture/__init__.py @@ -1 +1 @@ -"""Capture layer — AX Tree + screenshot + window metadata on a timer.""" +"""Capture layer — accessibility tree + screenshot + window metadata.""" diff --git a/src/openchronicle/capture/ax_capture.py b/src/openchronicle/capture/ax_capture.py index f5c297f9..d34ea3a8 100644 --- a/src/openchronicle/capture/ax_capture.py +++ b/src/openchronicle/capture/ax_capture.py @@ -1,8 +1,8 @@ -"""Cross-platform-stub AX Tree capture (macOS only in v1). +"""Cross-platform AX Tree capture provider selection. Wraps the vendored `mac-ax-helper` Swift binary. Ported from Einsia-Partner's -backend/core/capture/ax_capture_service.py with Windows branch removed and -resource resolution adapted for a uv/pip-installable package. +backend/core/capture/ax_capture_service.py with resource resolution adapted +for a uv/pip-installable package. """ from __future__ import annotations @@ -238,6 +238,10 @@ def _run( def create_provider(*, depth: int = 8, timeout: int = 3, raw: bool = False) -> AXProvider: + if platform.system() == "Windows": + from .windows_uia import create_provider as create_windows_provider + + return create_windows_provider(depth=depth, timeout=timeout, raw=raw) if platform.system() != "Darwin": return UnavailableAXProvider(f"unsupported platform: {platform.system()}") helper = _resolve_helper_path() diff --git a/src/openchronicle/capture/s1_parser.py b/src/openchronicle/capture/s1_parser.py index f8466616..3bbbb784 100644 --- a/src/openchronicle/capture/s1_parser.py +++ b/src/openchronicle/capture/s1_parser.py @@ -14,6 +14,7 @@ from __future__ import annotations import re +from collections.abc import Iterable from dataclasses import asdict, dataclass from typing import Any @@ -27,6 +28,11 @@ "company.thebrowser.Browser", "com.brave.Browser", "com.operasoftware.Opera", + "chrome.exe", + "msedge.exe", + "firefox.exe", + "brave.exe", + "opera.exe", } _URL_RE = re.compile(r"https?://\S+") @@ -89,25 +95,32 @@ def _extract_focused_element(app_data: dict[str, Any]) -> FocusedElement: for window in app_data.get("windows", []): if not window.get("focused"): continue - for el in window.get("elements", []): - role = el.get("role", "") or "" - if role in _EDITABLE_ROLES: - return FocusedElement( - role=role, - title=(el.get("title") or "")[:_FOCUS_TITLE_MAX], - value=(el.get("value") or "")[:_FOCUS_VALUE_MAX], - is_editable=True, - ) - if role in _STATIC_ROLES: - return FocusedElement( - role=role, - title=(el.get("title") or "")[:_FOCUS_TITLE_MAX], - value=(el.get("value") or el.get("title") or "")[:_FOCUS_VALUE_MAX], - is_editable=False, - ) + + elements = list(_walk_elements(window.get("elements", []))) + for el in elements: + if el.get("focused") and (el.get("role") or "") in (_EDITABLE_ROLES | _STATIC_ROLES): + return _focused_from_element(el) + for el in elements: + if (el.get("role") or "") in _EDITABLE_ROLES: + return _focused_from_element(el) + for el in elements: + if (el.get("role") or "") in _STATIC_ROLES: + return _focused_from_element(el) return FocusedElement() +def _focused_from_element(el: dict[str, Any]) -> FocusedElement: + role = el.get("role", "") or "" + is_editable = role in _EDITABLE_ROLES + value = el.get("value") or ("" if is_editable else el.get("title") or "") + return FocusedElement( + role=role, + title=(el.get("title") or "")[:_FOCUS_TITLE_MAX], + value=(value or "")[:_FOCUS_VALUE_MAX], + is_editable=is_editable, + ) + + def _render_visible_text(app_data: dict[str, Any]) -> str: md = ax_app_to_markdown(app_data) if len(md) > _VISIBLE_TEXT_MAX: @@ -116,18 +129,36 @@ def _render_visible_text(app_data: dict[str, Any]) -> str: def _extract_url(app_data: dict[str, Any]) -> str | None: - bundle = app_data.get("bundle_id", "") - if bundle not in _BROWSER_BUNDLES: + bundle = (app_data.get("bundle_id", "") or "").lower() + if bundle not in {b.lower() for b in _BROWSER_BUNDLES}: return None for window in app_data.get("windows", []): - for el in window.get("elements", []): - if el.get("role") != "AXTextField": + for el in _walk_elements(window.get("elements", [])): + if el.get("role") not in ("AXTextField", "AXComboBox", "AXWebArea"): continue value = (el.get("value") or "").strip() if not value: continue - if _URL_RE.search(value): - return value - if "." in value and " " not in value: + match = _URL_RE.search(value) + if match: + return match.group(0).rstrip(".,);]") + if _looks_like_bare_url(value): return f"https://{value}" return None + + +def _walk_elements(elements: list[dict[str, Any]]) -> Iterable[dict[str, Any]]: + for el in elements: + yield el + children = el.get("children") or [] + if isinstance(children, list): + yield from _walk_elements(children) + + +def _looks_like_bare_url(value: str) -> bool: + if any(ch.isspace() for ch in value): + return False + if "." not in value or len(value) > 300: + return False + lowered = value.lower() + return not lowered.startswith(("about:", "file:", "edge:", "chrome:")) diff --git a/src/openchronicle/capture/scheduler.py b/src/openchronicle/capture/scheduler.py index bcfe00dc..1c7eb1ec 100644 --- a/src/openchronicle/capture/scheduler.py +++ b/src/openchronicle/capture/scheduler.py @@ -6,6 +6,7 @@ import contextlib import hashlib import json +import platform import queue import threading import time @@ -87,7 +88,7 @@ def _write_capture(out: dict[str, Any]) -> Path: """Persist a built capture dict to the buffer, index it for search, and log.""" ts = out["timestamp"] path = paths.capture_buffer_dir() / f"{_safe_filename(ts)}.json" - path.write_text(json.dumps(out, ensure_ascii=False)) + path.write_text(json.dumps(out, ensure_ascii=False), encoding="utf-8") _index_capture(path.stem, out) meta = out.get("window_meta") or {} logger.info( @@ -301,7 +302,7 @@ async def run_forever( runner = _CaptureRunner(cfg, provider, pre_capture_hook=pre_capture_hook) runner.start_worker() - watcher: AXWatcherProcess | None = None + watcher: Any | None = None dispatcher: EventDispatcher | None = None def _on_capture(trigger: dict[str, Any] | None) -> None: @@ -310,7 +311,12 @@ def _on_capture(trigger: dict[str, Any] | None) -> None: runner.run_threaded(trigger) if cfg.event_driven: - watcher = AXWatcherProcess() + if platform.system() == "Windows": + from .windows_uia import WindowsPollingWatcher + + watcher = WindowsPollingWatcher(interval_seconds=cfg.poll_interval_seconds) + else: + watcher = AXWatcherProcess() if watcher.available: dispatcher = EventDispatcher( _on_capture, @@ -470,7 +476,7 @@ def _delete_captures_from_fts(stems: list[str]) -> None: def _strip_screenshot_inplace(path: Path) -> bool: """Rewrite a capture JSON without its ``screenshot`` field. Returns True if stripped.""" try: - raw = path.read_text() + raw = path.read_text(encoding="utf-8") except OSError: return False try: @@ -482,7 +488,7 @@ def _strip_screenshot_inplace(path: Path) -> bool: data.pop("screenshot", None) data["screenshot_stripped"] = True try: - path.write_text(json.dumps(data, ensure_ascii=False)) + path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8") return True except OSError: return False diff --git a/src/openchronicle/capture/window_meta.py b/src/openchronicle/capture/window_meta.py index 96e1faa1..8c46fe8f 100644 --- a/src/openchronicle/capture/window_meta.py +++ b/src/openchronicle/capture/window_meta.py @@ -1,13 +1,15 @@ -"""Foreground app / window metadata via osascript. macOS only in v1. +"""Foreground app / window metadata. Extracted from Einsia-Partner's capture_service.get_active_window_macos(). """ from __future__ import annotations +import ctypes import platform import subprocess from dataclasses import dataclass +from pathlib import Path from ..logger import get @@ -31,6 +33,14 @@ end tell """ +_WINDOWS_DISPLAY_NAMES = { + "chrome.exe": "Chrome", + "msedge.exe": "Edge", + "firefox.exe": "Firefox", + "brave.exe": "Brave", + "opera.exe": "Opera", +} + @dataclass class WindowMeta: @@ -40,7 +50,10 @@ class WindowMeta: def active_window() -> WindowMeta: - if platform.system() != "Darwin": + system = platform.system() + if system == "Windows": + return _active_window_windows() + if system != "Darwin": return WindowMeta() try: proc = subprocess.run( @@ -60,3 +73,44 @@ def active_window() -> WindowMeta: title=parts[1] if len(parts) > 1 else "", bundle_id=parts[2] if len(parts) > 2 else "", ) + + +def _active_window_windows() -> WindowMeta: + user32 = ctypes.windll.user32 + hwnd = user32.GetForegroundWindow() + if not hwnd: + return WindowMeta() + + title_length = user32.GetWindowTextLengthW(hwnd) + title_buffer = ctypes.create_unicode_buffer(title_length + 1) + user32.GetWindowTextW(hwnd, title_buffer, title_length + 1) + + pid = ctypes.c_ulong() + user32.GetWindowThreadProcessId(hwnd, ctypes.byref(pid)) + exe_name = _process_exe_name(pid.value) + bundle_id = exe_name.lower() + app_name = _WINDOWS_DISPLAY_NAMES.get(bundle_id) or ( + Path(exe_name).stem if exe_name else "" + ) + return WindowMeta(app_name=app_name, title=title_buffer.value, bundle_id=bundle_id) + + +def _process_exe_name(pid: int) -> str: + if not pid: + return "" + + kernel32 = ctypes.windll.kernel32 + process_query_limited_information = 0x1000 + handle = kernel32.OpenProcess(process_query_limited_information, False, pid) + if not handle: + return "" + + try: + size = ctypes.c_ulong(32768) + buffer = ctypes.create_unicode_buffer(size.value) + ok = kernel32.QueryFullProcessImageNameW(handle, 0, buffer, ctypes.byref(size)) + if not ok: + return "" + return Path(buffer.value).name + finally: + kernel32.CloseHandle(handle) diff --git a/src/openchronicle/capture/windows_uia.py b/src/openchronicle/capture/windows_uia.py new file mode 100644 index 00000000..cbddbb43 --- /dev/null +++ b/src/openchronicle/capture/windows_uia.py @@ -0,0 +1,388 @@ +"""Windows UI Automation capture and polling event source. + +This module adapts Windows UI Automation controls into the same AX-shaped JSON +that the macOS helper emits, so the existing S1 parser and downstream memory +pipeline do not need a Windows-specific path. +""" + +from __future__ import annotations + +import contextlib +import platform +import threading +from collections.abc import Callable, Iterable +from dataclasses import dataclass +from typing import Any + +from ..logger import get +from .ax_models import AXCaptureResult +from .window_meta import WindowMeta, active_window + +logger = get("openchronicle.capture") + +_MAX_CHILDREN_PER_NODE = 80 +_MAX_TOTAL_NODES = 2_000 + + +def _import_uia() -> tuple[Any | None, str | None]: + if platform.system() != "Windows": + return None, f"unsupported platform: {platform.system()}" + try: + import uiautomation as auto # type: ignore[import-not-found] + except ImportError as exc: + return None, f"missing optional dependency 'uiautomation': {exc}" + return auto, None + + +def _call_or_value(obj: Any, name: str, default: Any = "") -> Any: + try: + value = getattr(obj, name) + except Exception: # noqa: BLE001 - third-party COM wrappers raise varied errors + return default + if callable(value): + try: + return value() + except Exception: # noqa: BLE001 + return default + return value + + +def _text(value: Any, *, limit: int = 2_000) -> str: + if value is None: + return "" + if not isinstance(value, str): + value = str(value) + value = value.replace("\x00", "").strip() + if len(value) > limit: + return value[:limit] + "...(truncated)" + return value + + +def _control_value(control: Any) -> str: + for attr in ("Value", "value"): + value = _text(_call_or_value(control, attr, default="")) + if value: + return value + + for method in ("GetValuePattern", "GetLegacyIAccessiblePattern"): + getter = getattr(control, method, None) + if not callable(getter): + continue + try: + pattern = getter() + except Exception: # noqa: BLE001 + continue + for attr in ("Value", "Name", "Description"): + value = _text(_call_or_value(pattern, attr, default="")) + if value: + return value + return "" + + +def _control_name(control: Any) -> str: + return _text(_call_or_value(control, "Name", default=""), limit=1_000) + + +def _control_type(control: Any) -> str: + return _text(_call_or_value(control, "ControlTypeName", default=""), limit=200) + + +def _automation_id(control: Any) -> str: + return _text(_call_or_value(control, "AutomationId", default=""), limit=200) + + +def _class_name(control: Any) -> str: + return _text(_call_or_value(control, "ClassName", default=""), limit=200) + + +def _role_for_control(control_type: str) -> str: + compact = control_type.replace(" ", "").lower() + if "edit" in compact: + return "AXTextField" + if "document" in compact: + return "AXWebArea" + if "text" in compact: + return "AXStaticText" + if "button" in compact: + return "AXButton" + if "combobox" in compact: + return "AXComboBox" + if "menuitem" in compact: + return "AXMenuItem" + if "tabitem" in compact: + return "AXTab" + if "listitem" in compact: + return "AXRow" + if "window" in compact: + return "AXWindow" + if "pane" in compact or "group" in compact: + return "AXGroup" + return f"AX{control_type}" if control_type else "AXUnknown" + + +def _iter_children(control: Any) -> Iterable[Any]: + get_children = getattr(control, "GetChildren", None) + if callable(get_children): + try: + yield from list(get_children())[:_MAX_CHILDREN_PER_NODE] + return + except Exception: # noqa: BLE001 + pass + + first_child = getattr(control, "GetFirstChildControl", None) + if not callable(first_child): + return + try: + child = first_child() + except Exception: # noqa: BLE001 + return + + count = 0 + while child is not None and count < _MAX_CHILDREN_PER_NODE: + yield child + count += 1 + next_sibling = getattr(child, "GetNextSiblingControl", None) + if not callable(next_sibling): + break + try: + child = next_sibling() + except Exception: # noqa: BLE001 + break + + +@dataclass +class _Budget: + remaining: int = _MAX_TOTAL_NODES + + def take(self) -> bool: + if self.remaining <= 0: + return False + self.remaining -= 1 + return True + + +def _element_from_control(control: Any, *, depth: int, budget: _Budget) -> dict[str, Any] | None: + if depth < 0 or not budget.take(): + return None + + control_type = _control_type(control) + name = _control_name(control) + value = _control_value(control) + role = _role_for_control(control_type) + + children: list[dict[str, Any]] = [] + if depth > 0: + for child in _iter_children(control): + child_el = _element_from_control(child, depth=depth - 1, budget=budget) + if child_el is not None: + children.append(child_el) + + element: dict[str, Any] = { + "role": role, + "title": name, + "value": value, + } + automation_id = _automation_id(control) + class_name = _class_name(control) + if automation_id: + element["identifier"] = automation_id + if class_name: + element["class_name"] = class_name + with contextlib.suppress(Exception): + if bool(_call_or_value(control, "HasKeyboardFocus", default=False)): + element["focused"] = True + if children: + element["children"] = children + + if not name and not value and not children: + return None + return element + + +def _safe_foreground_control(auto: Any) -> Any | None: + getter = getattr(auto, "GetForegroundControl", None) + if not callable(getter): + return None + try: + return getter() + except Exception as exc: # noqa: BLE001 + logger.warning("Windows UIA foreground control failed: %s", exc) + return None + + +def _safe_focused_control(auto: Any) -> Any | None: + getter = getattr(auto, "GetFocusedControl", None) + if not callable(getter): + return None + try: + return getter() + except Exception: # noqa: BLE001 + return None + + +def _control_identity(control: Any | None) -> tuple[str, str, str, str]: + if control is None: + return ("", "", "", "") + return ( + _text(_call_or_value(control, "ProcessId", default=""), limit=50), + _control_type(control), + _automation_id(control), + _control_name(control), + ) + + +class WindowsUIAProvider: + """One-shot Windows UI Automation provider.""" + + def __init__(self, *, depth: int, timeout: int, auto_module: Any | None = None) -> None: + auto, reason = (auto_module, None) if auto_module is not None else _import_uia() + self._auto = auto + self.reason = reason or "" + self._depth = max(depth, 0) + self._timeout = timeout + + @property + def available(self) -> bool: + return self._auto is not None + + def capture_frontmost(self, *, focused_window_only: bool = True) -> AXCaptureResult | None: + return self._capture() + + def capture_all_visible(self) -> AXCaptureResult | None: + return self._capture() + + def capture_app( + self, app_name: str, *, focused_window_only: bool = True + ) -> AXCaptureResult | None: + return self._capture() + + def _capture(self) -> AXCaptureResult | None: + if self._auto is None: + return None + + meta = active_window() + foreground = _safe_foreground_control(self._auto) + if foreground is None: + return None + + budget = _Budget() + root_el = _element_from_control(foreground, depth=self._depth, budget=budget) + elements: list[dict[str, Any]] = [] + focused = _safe_focused_control(self._auto) + focused_el = None + if _control_identity(focused) != _control_identity(foreground): + focused_el = _element_from_control(focused, depth=min(self._depth, 3), budget=budget) + if focused_el is not None: + elements.append(focused_el) + if root_el is not None: + elements.extend(root_el.pop("children", []) or [root_el]) + + title = meta.title or _control_name(foreground) + app_name = meta.app_name or "Windows App" + bundle_id = meta.bundle_id or "unknown.exe" + tree = { + "timestamp": "", + "apps": [ + { + "name": app_name, + "bundle_id": bundle_id, + "is_frontmost": True, + "windows": [ + { + "title": title, + "focused": True, + "elements": elements, + } + ], + } + ], + } + return AXCaptureResult( + raw_json=tree, + timestamp="", + apps=tree["apps"], + metadata={ + "mode": "frontmost", + "depth": self._depth, + "platform": "windows", + "provider": "uiautomation", + "timeout": self._timeout, + }, + ) + + +class WindowsPollingWatcher: + """Polling event source for Windows. + + It emits macOS-style event names so the existing EventDispatcher can keep + applying the same debounce and dedup rules. + """ + + def __init__(self, *, interval_seconds: float = 5.0) -> None: + self._interval = max(interval_seconds, 1.0) + self._callback: Callable[[dict[str, Any]], None] | None = None + self._thread: threading.Thread | None = None + self._stop_event = threading.Event() + self._last: WindowMeta | None = None + + @property + def available(self) -> bool: + return platform.system() == "Windows" + + @property + def running(self) -> bool: + return self._thread is not None and self._thread.is_alive() + + def on_event(self, callback: Callable[[dict[str, Any]], None]) -> None: + self._callback = callback + + def start(self) -> None: + if not self.available: + logger.warning("Windows polling watcher unavailable on %s", platform.system()) + return + self._stop_event.clear() + self._thread = threading.Thread( + target=self._run_loop, daemon=True, name="windows-uia-poller" + ) + self._thread.start() + logger.info("Windows polling capture started (interval=%.1fs)", self._interval) + + def stop(self, *, join_timeout: float = 5.0) -> None: + self._stop_event.set() + if self._thread is not None and self._thread.is_alive(): + self._thread.join(timeout=join_timeout) + if self._thread.is_alive(): + logger.warning("Windows polling watcher did not exit within %.1fs", join_timeout) + self._thread = None + logger.info("Windows polling capture stopped") + + def _run_loop(self) -> None: + while not self._stop_event.is_set(): + self._poll_once() + self._stop_event.wait(self._interval) + + def _poll_once(self) -> None: + meta = active_window() + if not meta.app_name and not meta.title and not meta.bundle_id: + return + + event_type = "AXValueChanged" + if self._last is None or meta.bundle_id != self._last.bundle_id: + event_type = "AXApplicationActivated" + elif meta.title != self._last.title: + event_type = "AXFocusedWindowChanged" + self._last = meta + + if self._callback is not None: + self._callback( + { + "event_type": event_type, + "app": meta.app_name, + "bundle_id": meta.bundle_id, + "window_title": meta.title, + } + ) + + +def create_provider(*, depth: int = 8, timeout: int = 3, raw: bool = False) -> WindowsUIAProvider: + return WindowsUIAProvider(depth=depth, timeout=timeout) diff --git a/src/openchronicle/cli.py b/src/openchronicle/cli.py index 69af91cf..82d09886 100644 --- a/src/openchronicle/cli.py +++ b/src/openchronicle/cli.py @@ -5,9 +5,12 @@ import contextlib import json import os +import platform import shutil import signal import subprocess +import sys +import time from datetime import datetime from pathlib import Path @@ -38,7 +41,22 @@ def _init() -> config_mod.Config: return config_mod.load() +def _init_silent() -> config_mod.Config: + """Initialize OpenChronicle without writing to stdout. + + stdio MCP servers must keep stdout reserved for JSON-RPC messages. + """ + paths.ensure_dirs() + config_mod.write_default_if_missing() + logger_mod.setup(console=False) + return config_mod.load() + + def _is_pid_alive(pid: int) -> bool: + if pid <= 0: + return False + if platform.system() == "Windows": + return _is_pid_alive_windows(pid) try: os.kill(pid, 0) except ProcessLookupError: @@ -48,9 +66,28 @@ def _is_pid_alive(pid: int) -> bool: return True +def _is_pid_alive_windows(pid: int) -> bool: + import ctypes + + kernel32 = ctypes.windll.kernel32 + process_query_limited_information = 0x1000 + still_active = 259 + handle = kernel32.OpenProcess(process_query_limited_information, False, pid) + if not handle: + error_access_denied = 5 + return kernel32.GetLastError() == error_access_denied + try: + exit_code = ctypes.c_ulong() + if not kernel32.GetExitCodeProcess(handle, ctypes.byref(exit_code)): + return False + return exit_code.value == still_active + finally: + kernel32.CloseHandle(handle) + + def _read_pid() -> int | None: try: - pid = int(paths.pid_file().read_text().strip()) + pid = int(paths.pid_file().read_text(encoding="utf-8").strip()) except (FileNotFoundError, ValueError): return None return pid if _is_pid_alive(pid) else None @@ -121,6 +158,62 @@ def _health_status(pid: int | None, last_ts: str | None) -> tuple[str, str]: return "stale (no captures in >5m)", "yellow" +def _wait_for_pid(timeout: float = 5.0) -> int | None: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + pid = _read_pid() + if pid: + return pid + time.sleep(0.1) + return _read_pid() + + +def _start_background_windows(*, capture_only: bool) -> int | None: + cmd = [sys.executable, "-m", "openchronicle.cli", "start", "--foreground"] + if capture_only: + cmd.append("--capture-only") + + creationflags = 0 + for name in ("CREATE_NEW_PROCESS_GROUP", "DETACHED_PROCESS", "CREATE_NO_WINDOW"): + creationflags |= getattr(subprocess, name, 0) + + subprocess.Popen( # noqa: S603 + cmd, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + close_fds=True, + creationflags=creationflags, + ) + return _wait_for_pid() + + +def _start_background_posix(cfg: config_mod.Config, *, capture_only: bool) -> None: + from . import daemon + + if os.fork() != 0: + console.print("[green]OpenChronicle started in background.[/green]") + console.print(f"Logs: {paths.logs_dir()}") + return + os.setsid() + if os.fork() != 0: + os._exit(0) + # Redirect stdio to /dev/null. After dup2 the original fd is no longer + # needed; closing it avoids leaking one descriptor per daemon start. + devnull = os.open(os.devnull, os.O_RDWR) + for fd in (0, 1, 2): + os.dup2(devnull, fd) + if devnull > 2: + os.close(devnull) + daemon.run(cfg, capture_only=capture_only) + os._exit(0) + + +def _remove_stale_pid() -> None: + with contextlib.suppress(FileNotFoundError): + paths.pid_file().unlink() + + # ─── commands ───────────────────────────────────────────────────────────── @app.command() @@ -134,6 +227,8 @@ def start( if pid: console.print(f"[yellow]Already running (pid {pid})[/yellow]") raise typer.Exit(1) + if paths.pid_file().exists(): + _remove_stale_pid() from . import daemon @@ -142,23 +237,15 @@ def start( daemon.run(cfg, capture_only=capture_only) return - # Background: double-fork - if os.fork() != 0: + if platform.system() == "Windows": + started_pid = _start_background_windows(capture_only=capture_only) console.print("[green]OpenChronicle started in background.[/green]") console.print(f"Logs: {paths.logs_dir()}") + if started_pid: + console.print(f"PID: {started_pid}") return - os.setsid() - if os.fork() != 0: - os._exit(0) - # Redirect stdio to /dev/null. After dup2 the original fd is no longer - # needed; closing it avoids leaking one descriptor per daemon start. - devnull = os.open(os.devnull, os.O_RDWR) - for fd in (0, 1, 2): - os.dup2(devnull, fd) - if devnull > 2: - os.close(devnull) - daemon.run(cfg, capture_only=capture_only) - os._exit(0) + + _start_background_posix(cfg, capture_only=capture_only) @app.command() @@ -171,13 +258,19 @@ def stop() -> None: raise typer.Exit(1) os.kill(pid, signal.SIGTERM) console.print(f"[green]Sent SIGTERM to pid {pid}.[/green]") + if platform.system() == "Windows": + deadline = time.monotonic() + 5.0 + while time.monotonic() < deadline and _is_pid_alive(pid): + time.sleep(0.1) + if not _is_pid_alive(pid): + _remove_stale_pid() @app.command() def pause() -> None: """Pause capture (daemon stays up but skips captures).""" paths.ensure_dirs() - paths.paused_flag().write_text(datetime.now().isoformat()) + paths.paused_flag().write_text(datetime.now().isoformat(), encoding="utf-8") console.print("[yellow]Capture paused.[/yellow]") @@ -339,6 +432,15 @@ def mcp() -> None: mcp_server.run_stdio() +@app.command("adb-mcp") +def adb_mcp() -> None: + """Run the Android ADB control MCP server (stdio).""" + _init_silent() + from .mcp import adb_server + + adb_server.run_stdio() + + install_app = typer.Typer(help="Register the MCP server with common LLM clients.") app.add_typer(install_app, name="install") @@ -419,7 +521,7 @@ def _load_claude_desktop_config(path: Path) -> dict: if not path.exists(): return {} try: - data = json.loads(path.read_text()) + data = json.loads(path.read_text(encoding="utf-8")) except json.JSONDecodeError as exc: console.print( f"[red]Could not parse {path}:[/red] {exc}\n" @@ -479,7 +581,7 @@ def install_claude_desktop( "args": ["mcp"], } - cfg_path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n") + cfg_path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") verb = "Updated" if replaced else "Registered" console.print(f"[green]{verb} {name!r} in Claude Desktop config.[/green]") @@ -553,7 +655,7 @@ def _load_opencode_config(path: Path) -> dict: if not path.exists(): return {} try: - data = json.loads(path.read_text()) + data = json.loads(path.read_text(encoding="utf-8")) except json.JSONDecodeError as exc: console.print( f"[red]Could not parse {path}:[/red] {exc}\n" @@ -622,7 +724,7 @@ def install_opencode( "enabled": True, } - cfg_path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n") + cfg_path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") verb = "Updated" if replaced else "Registered" console.print(f"[green]{verb} {name!r} in opencode config.[/green]") @@ -676,7 +778,7 @@ def install_mcp_json( summary = f"stdio → {openchronicle_bin} mcp" payload = {"mcpServers": {name: entry}} - out_path.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n") + out_path.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") console.print(f"[green]Wrote {out_path}[/green]") console.print(f" server: {name} ({summary})") @@ -781,7 +883,7 @@ def uninstall_opencode( return del servers[name] - cfg_path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n") + cfg_path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") console.print(f"[green]Removed {name!r} from opencode config.[/green]") @@ -809,7 +911,7 @@ def uninstall_claude_desktop( return del servers[name] - cfg_path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n") + cfg_path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") console.print(f"[green]Removed {name!r} from Claude Desktop config.[/green]") _restart_reminder("finalize the removal") @@ -929,7 +1031,7 @@ def rebuild_captures_index() -> None: with fts.cursor() as conn: for p in files: try: - data = json.loads(p.read_text()) + data = json.loads(p.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: skipped += 1 console.print(f"[yellow]skip {p.name}: {exc}[/yellow]") @@ -968,7 +1070,7 @@ def config() -> None: _init() p = paths.config_file() console.print(f"[bold]{p}[/bold]") - console.print(p.read_text()) + console.print(p.read_text(encoding="utf-8")) clean_app = typer.Typer(help="Delete past data. Destructive — use with care.") diff --git a/src/openchronicle/config.py b/src/openchronicle/config.py index 21794c89..45f2aea8 100644 --- a/src/openchronicle/config.py +++ b/src/openchronicle/config.py @@ -23,8 +23,9 @@ class ModelConfig: @dataclass class CaptureConfig: # Event-driven capture knobs - event_driven: bool = True # consume mac-ax-watcher events + event_driven: bool = True # macOS AX watcher or Windows UIA poller heartbeat_minutes: int = 10 # periodic capture even without events + poll_interval_seconds: float = 5.0 # Windows UIA polling interval debounce_seconds: float = 3.0 # for AXValueChanged bursts min_capture_gap_seconds: float = 2.0 # between consecutive captures dedup_interval_seconds: float = 1.0 # per-event-type dedup window @@ -228,8 +229,9 @@ def load(path: Path | None = None) -> Config: # Accuracy-sensitive — pick a capable model. [capture] -event_driven = true # capture on window/app/typing events via mac-ax-watcher +event_driven = true # macOS AX watcher or Windows UIA poller heartbeat_minutes = 10 # periodic capture even when nothing happens +poll_interval_seconds = 5.0 # Windows UI Automation polling interval debounce_seconds = 3.0 # for AXValueChanged bursts min_capture_gap_seconds = 2.0 # minimum gap between consecutive captures dedup_interval_seconds = 1.0 # per-event-type dedup window @@ -289,5 +291,5 @@ def write_default_if_missing(path: Path | None = None) -> bool: if path.exists(): return False path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(DEFAULT_CONFIG_TEMPLATE) + path.write_text(DEFAULT_CONFIG_TEMPLATE, encoding="utf-8") return True diff --git a/src/openchronicle/daemon.py b/src/openchronicle/daemon.py index de0af579..ce72b4ae 100644 --- a/src/openchronicle/daemon.py +++ b/src/openchronicle/daemon.py @@ -50,7 +50,7 @@ async def _mcp_loop(cfg: Config) -> None: async def _run(cfg: Config, *, capture_only: bool = False) -> None: paths.ensure_dirs() - paths.pid_file().write_text(str(os.getpid())) + paths.pid_file().write_text(str(os.getpid()), encoding="utf-8") # SessionManager observes every capture-worthy event and fires the # reducer via its on_session_end callback. Built even when diff --git a/src/openchronicle/mcp/adb_server.py b/src/openchronicle/mcp/adb_server.py new file mode 100644 index 00000000..ad9b7c89 --- /dev/null +++ b/src/openchronicle/mcp/adb_server.py @@ -0,0 +1,124 @@ +"""MCP server exposing safe Android ADB control tools.""" + +from __future__ import annotations + +import json + +from ..adb import ADBController +from ..config import Config + +_SERVER_INSTRUCTIONS = """\ +# OpenChronicle ADB Control + +This MCP server lets an agent operate an Android device through a small, safe +ADB tool surface. Every tool call is appended to OpenChronicle's daily +event-YYYY-MM-DD.md memory file. + +Rules for agents: + +1. Observe before acting: call adb_screenshot or adb_dump_ui before tap/swipe/text. +2. Do not request delete, uninstall, clear-data, root, reboot, remount, or settings writes. +3. Stop and ask the user before payments, passwords, SMS codes, private data export, or account changes. +4. Prefer package/activity based app launch over blind navigation when the package is known. +""" + + +def build_server(cfg: Config | None = None, controller: ADBController | None = None): + """Construct and return a FastMCP server instance for ADB control.""" + from mcp.server.fastmcp import FastMCP + + del cfg # The ADB server is stdio-first and currently has no config section. + adb = controller or ADBController() + server = FastMCP("openchronicle-adb", instructions=_SERVER_INSTRUCTIONS) + + @server.tool() + def adb_list_devices() -> str: + """List Android devices visible to adb and record the operation in memory.""" + return json.dumps(adb.list_devices(), ensure_ascii=False) + + @server.tool() + def adb_screenshot(device_id: str | None = None) -> str: + """Capture a PNG screenshot from the Android device and return its local path.""" + return json.dumps(adb.screenshot(device_id=device_id), ensure_ascii=False) + + @server.tool() + def adb_dump_ui(device_id: str | None = None) -> str: + """Dump the Android UI Automator XML tree for the current screen.""" + return json.dumps(adb.dump_ui(device_id=device_id), ensure_ascii=False) + + @server.tool() + def adb_tap(x: int, y: int, device_id: str | None = None) -> str: + """Tap an Android screen coordinate.""" + return json.dumps(adb.tap(x=x, y=y, device_id=device_id), ensure_ascii=False) + + @server.tool() + def adb_swipe( + x1: int, + y1: int, + x2: int, + y2: int, + duration_ms: int = 300, + device_id: str | None = None, + ) -> str: + """Swipe from one Android screen coordinate to another.""" + return json.dumps( + adb.swipe( + x1=x1, + y1=y1, + x2=x2, + y2=y2, + duration_ms=duration_ms, + device_id=device_id, + ), + ensure_ascii=False, + ) + + @server.tool() + def adb_input_text(text: str, device_id: str | None = None) -> str: + """Input text through Android's input command. Text is redacted in memory.""" + return json.dumps(adb.input_text(text=text, device_id=device_id), ensure_ascii=False) + + @server.tool() + def adb_keyevent(keyevent: str, device_id: str | None = None) -> str: + """Send a safe Android keyevent, such as BACK, HOME, ENTER, or a numeric keycode.""" + return json.dumps(adb.keyevent(keyevent=keyevent, device_id=device_id), ensure_ascii=False) + + @server.tool() + def adb_current_app(device_id: str | None = None) -> str: + """Return the foreground Android package/activity when adb can determine it.""" + return json.dumps(adb.current_app(device_id=device_id), ensure_ascii=False) + + @server.tool() + def adb_open_app( + package_name: str, + activity: str | None = None, + device_id: str | None = None, + ) -> str: + """Open an Android app by package name, optionally with an explicit activity.""" + return json.dumps( + adb.open_app(package_name=package_name, activity=activity, device_id=device_id), + ensure_ascii=False, + ) + + @server.tool() + def adb_read_logcat( + lines: int = 200, + filter_expr: str | None = None, + device_id: str | None = None, + ) -> str: + """Read bounded logcat output. Defaults to the last 200 lines.""" + return json.dumps( + adb.read_logcat(lines=lines, filter_expr=filter_expr, device_id=device_id), + ensure_ascii=False, + ) + + return server + + +def run_stdio() -> None: + """Run the ADB MCP server on stdio.""" + build_server().run() + + +if __name__ == "__main__": + run_stdio() diff --git a/src/openchronicle/mcp/captures.py b/src/openchronicle/mcp/captures.py index 9c56cc94..3b2a9fd1 100644 --- a/src/openchronicle/mcp/captures.py +++ b/src/openchronicle/mcp/captures.py @@ -78,7 +78,7 @@ def _matches( def _load_capture(path: Path) -> dict[str, Any] | None: try: - return json.loads(path.read_text()) + return json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return None diff --git a/src/openchronicle/paths.py b/src/openchronicle/paths.py index 7c2d95bc..dbf436fb 100644 --- a/src/openchronicle/paths.py +++ b/src/openchronicle/paths.py @@ -1,8 +1,9 @@ -"""Single source of truth for on-disk locations under ~/.openchronicle/.""" +"""Single source of truth for on-disk OpenChronicle locations.""" from __future__ import annotations import os +import platform from pathlib import Path @@ -10,6 +11,11 @@ def root() -> Path: override = os.environ.get("OPENCHRONICLE_ROOT") if override: return Path(override).expanduser().resolve() + if platform.system() == "Windows": + local_app_data = os.environ.get("LOCALAPPDATA") + if local_app_data: + return Path(local_app_data) / "OpenChronicle" + return Path.home() / "AppData" / "Local" / "OpenChronicle" return Path.home() / ".openchronicle" diff --git a/src/openchronicle/store/entries.py b/src/openchronicle/store/entries.py index 6778e895..4c64592c 100644 --- a/src/openchronicle/store/entries.py +++ b/src/openchronicle/store/entries.py @@ -97,7 +97,7 @@ def append_entry( # write — both writes claim "+1 entry" but only one entry survives # while the FTS index keeps both, leaving file/index inconsistent. with files_mod.file_lock(path): - post = frontmatter.load(path) + post = frontmatter.loads(path.read_text(encoding="utf-8")) current = post.content.rstrip() new_block = f"\n\n{heading}\n{body}\n" if current else f"{heading}\n{body}\n" post.content = current + new_block @@ -178,7 +178,7 @@ def supersede_entry( ) # Modify file text directly to preserve formatting - text = path.read_text() + text = path.read_text(encoding="utf-8") # 1) append #superseded-by to old heading (only if not already present) old_heading = target.heading_line if f"superseded-by:{new_id}" not in old_heading: diff --git a/src/openchronicle/store/files.py b/src/openchronicle/store/files.py index 95538731..cadabd1f 100644 --- a/src/openchronicle/store/files.py +++ b/src/openchronicle/store/files.py @@ -7,6 +7,7 @@ import re import tempfile import threading +import time from collections.abc import Iterator from dataclasses import dataclass, field from datetime import date @@ -48,9 +49,9 @@ def atomic_write_text(path: Path, content: str) -> None: os.fsync(f.fileno()) # Preserve permissions of the existing file so updates don't # silently flip group/other-read bits set by the user. - with contextlib.suppress(FileNotFoundError): + with contextlib.suppress(OSError): os.chmod(tmp_path, path.stat().st_mode & 0o7777) - os.replace(tmp_path, path) + _replace_with_retry(tmp_path, path) # Persist the directory entry so a power loss right after the # rename can't leave the dir pointing at neither old nor new. # macOS APFS sometimes returns EINVAL on directory fsync; the @@ -67,6 +68,18 @@ def atomic_write_text(path: Path, content: str) -> None: tmp_path.unlink() raise + +def _replace_with_retry(tmp_path: Path, path: Path) -> None: + attempts = 8 if os.name == "nt" else 1 + for attempt in range(attempts): + try: + os.replace(tmp_path, path) + return + except PermissionError: + if attempt == attempts - 1: + raise + time.sleep(0.05 * (attempt + 1)) + VALID_PREFIXES = ("user-", "project-", "tool-", "topic-", "person-", "org-", "event-") @@ -184,7 +197,7 @@ def write_file(path: Path, fm: dict[str, Any], body: str) -> None: def read_file(path: Path) -> ParsedFile: if not path.exists(): raise FileNotFoundError(path) - post = frontmatter.load(path) + post = frontmatter.loads(path.read_text(encoding="utf-8")) fm = dict(post.metadata) body = post.content entries = _parse_entries(body) @@ -251,7 +264,7 @@ def render_file( def update_frontmatter(path: Path, updates: dict[str, Any]) -> None: with file_lock(path): - post = frontmatter.load(path) + post = frontmatter.loads(path.read_text(encoding="utf-8")) post.metadata.update(updates) atomic_write_text(path, frontmatter.dumps(post) + "\n") diff --git a/src/openchronicle/writer/compact.py b/src/openchronicle/writer/compact.py index 2e644308..0bb8a74e 100644 --- a/src/openchronicle/writer/compact.py +++ b/src/openchronicle/writer/compact.py @@ -45,7 +45,7 @@ def compact_file(cfg: Config, conn: sqlite3.Connection, *, name: str) -> Compact if not path.exists(): return CompactResult(name, False, 0, 0, 0, 0, 0.0, "file missing") - original = path.read_text() + original = path.read_text(encoding="utf-8") before_unique = _unique_tokens(original) before_tokens = len(original) // 4 diff --git a/tests/test_adb_control.py b/tests/test_adb_control.py new file mode 100644 index 00000000..c0dbff63 --- /dev/null +++ b/tests/test_adb_control.py @@ -0,0 +1,114 @@ +from __future__ import annotations + +import pytest + +from openchronicle.adb.client import ADBCommandResult +from openchronicle.adb.memory import ADBMemoryRecorder +from openchronicle.adb.safety import ADBSafetyError, assert_safe +from openchronicle.adb.tools import ADBController, _parse_devices +from openchronicle.mcp import adb_server +from openchronicle.store import files as files_mod + + +class FakeADBClient: + adb_path = "adb" + + def __init__(self, stdout: str = "", stdout_bytes: bytes = b"") -> None: + self.stdout = stdout + self.stdout_bytes = stdout_bytes + self.calls: list[tuple[list[str], str | None, bool]] = [] + + def command_for_display(self, args, device_id=None): + command = [self.adb_path] + if device_id: + command.extend(["-s", device_id]) + command.extend(str(a) for a in args) + return command + + def run(self, args, *, device_id=None, timeout=30.0, binary=False, check=True): + del timeout, check + self.calls.append((list(args), device_id, binary)) + stdout = self.stdout_bytes if binary else self.stdout + return ADBCommandResult( + args=self.command_for_display(args, device_id), + returncode=0, + stdout=stdout, + stderr="", + ) + + +def test_safety_blocks_destructive_commands() -> None: + blocked = [ + ["uninstall", "com.example.app"], + ["reboot"], + ["root"], + ["shell", "pm", "clear", "com.example.app"], + ["shell", "settings", "put", "system", "screen_brightness", "1"], + ["shell", "rm", "-rf", "/sdcard/Download"], + ["shell", "input", "keyevent", "POWER"], + ["shell", "input", "keyevent", "26"], + ] + for command in blocked: + with pytest.raises(ADBSafetyError): + assert_safe(command) + + +def test_safety_allows_mvp_commands() -> None: + allowed = [ + ["devices", "-l"], + ["exec-out", "screencap", "-p"], + ["shell", "uiautomator", "dump", "/sdcard/window.xml"], + ["shell", "input", "tap", "100", "200"], + ["shell", "input", "swipe", "100", "900", "100", "100", "300"], + ["shell", "input", "keyevent", "BACK"], + ["shell", "monkey", "-p", "com.example.app", "-c", "android.intent.category.LAUNCHER", "1"], + ] + for command in allowed: + assert_safe(command) + + +def test_parse_devices() -> None: + output = """List of devices attached +emulator-5554 device product:sdk_gphone_x86_64 model:sdk_gphone64 transport_id:1 +R58M123 offline usb:1-1 +""" + devices = _parse_devices(output) + assert devices[0]["serial"] == "emulator-5554" + assert devices[0]["state"] == "device" + assert devices[0]["qualifiers"]["product"] == "sdk_gphone_x86_64" + assert devices[1]["state"] == "offline" + + +def test_list_devices_writes_openchronicle_event(ac_root) -> None: + client = FakeADBClient( + stdout="List of devices attached\nemulator-5554 device product:sdk model:Pixel\n" + ) + controller = ADBController(client=client, recorder=ADBMemoryRecorder()) + + result = controller.list_devices() + + assert result["ok"] is True + assert result["count"] == 1 + event_files = list((ac_root / "memory").glob("event-*.md")) + assert len(event_files) == 1 + parsed = files_mod.read_file(event_files[0]) + assert parsed.entries + assert "ADB tool adb_list_devices" in parsed.entries[-1].body + + +def test_blocked_input_text_is_recorded(ac_root) -> None: + controller = ADBController(client=FakeADBClient(), recorder=ADBMemoryRecorder()) + + result = controller.input_text("hello; rm -rf /") + + assert result["ok"] is False + assert result["blocked"] is True + assert not controller.client.calls + event_file = next((ac_root / "memory").glob("event-*.md")) + parsed = files_mod.read_file(event_file) + assert "blocked by safety policy" in parsed.entries[-1].body + + +def test_adb_mcp_server_builds() -> None: + server = adb_server.build_server(controller=ADBController(client=FakeADBClient())) + assert server is not None diff --git a/tests/test_classifier.py b/tests/test_classifier.py index 5ca1fcaf..3a34c50e 100644 --- a/tests/test_classifier.py +++ b/tests/test_classifier.py @@ -1,13 +1,11 @@ from __future__ import annotations import json -from datetime import datetime, timedelta, timezone +from datetime import timedelta, timezone from pathlib import Path from types import SimpleNamespace from typing import Any -import pytest - from openchronicle import config as config_mod from openchronicle import paths from openchronicle.store import entries as entries_mod @@ -15,7 +13,6 @@ from openchronicle.writer import classifier as classifier_mod from openchronicle.writer import llm as llm_mod - _TZ = timezone(timedelta(hours=8)) @@ -94,11 +91,11 @@ def fake_call_llm(cfg, stage, *, messages, tools=None, json_mode=False): assert "Cursor-over-VSCode" in result.summary # Event-daily was NOT modified. - evt = (paths.memory_dir() / name).read_text() + evt = (paths.memory_dir() / name).read_text(encoding="utf-8") assert evt.count("**Session sess_abc**") == 1 # user-preferences.md got the new entry. - pref = (paths.memory_dir() / "user-preferences.md").read_text() + pref = (paths.memory_dir() / "user-preferences.md").read_text(encoding="utf-8") assert "Cursor over VSCode" in pref diff --git a/tests/test_config.py b/tests/test_config.py index f5e742db..11d6b73e 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -38,7 +38,7 @@ def test_write_default_creates_file(tmp_path: Path) -> None: p = tmp_path / "config.toml" assert config.write_default_if_missing(p) assert p.exists() - assert "[models.default]" in p.read_text() + assert "[models.default]" in p.read_text(encoding="utf-8") # idempotent assert not config.write_default_if_missing(p) diff --git a/tests/test_session_reducer.py b/tests/test_session_reducer.py index cfd70364..315aac44 100644 --- a/tests/test_session_reducer.py +++ b/tests/test_session_reducer.py @@ -4,8 +4,6 @@ from datetime import datetime, timedelta, timezone from pathlib import Path -import pytest - from openchronicle import config as config_mod from openchronicle import paths from openchronicle.session import store as session_store @@ -13,7 +11,6 @@ from openchronicle.timeline import store as timeline_store from openchronicle.writer import session_reducer - _TZ = timezone(timedelta(hours=8)) _SID = "sess_test0000" @@ -91,7 +88,7 @@ def test_reducer_happy_path_writes_event_daily(ac_root: Path, monkeypatch) -> No assert result.path == "event-2026-04-21.md" assert len(result.sub_tasks) == 1 - md = (paths.memory_dir() / "event-2026-04-21.md").read_text() + md = (paths.memory_dir() / "event-2026-04-21.md").read_text(encoding="utf-8") assert "Session sess_test0000" in md assert "[10:00-10:15, Cursor]" in md assert "file_0.py" in md @@ -191,7 +188,7 @@ def test_reducer_exhausted_retries_writes_heuristic(ac_root: Path, monkeypatch) assert result.succeeded is False assert result.written is True - md = (paths.memory_dir() / "event-2026-04-21.md").read_text() + md = (paths.memory_dir() / "event-2026-04-21.md").read_text(encoding="utf-8") assert "Cursor" in md assert "heuristic" in md # tag should be present on the heading @@ -260,7 +257,7 @@ def test_flush_active_session_writes_partial_entry( assert result.is_final is False assert result.written is True - md = (paths.memory_dir() / "event-2026-04-21.md").read_text() + md = (paths.memory_dir() / "event-2026-04-21.md").read_text(encoding="utf-8") assert "Session sess_flush1 [flush]" in md with fts.cursor() as conn: @@ -322,7 +319,7 @@ def test_terminal_reduce_after_flush_covers_trailing_window( assert result.written is True assert result.is_final is True - md = (paths.memory_dir() / "event-2026-04-21.md").read_text() + md = (paths.memory_dir() / "event-2026-04-21.md").read_text(encoding="utf-8") # The terminal entry is NOT tagged as flush. assert "Session sess_flush2 [flush]" not in md assert "Session sess_flush2" in md diff --git a/tests/test_store.py b/tests/test_store.py index 4b7dfbd6..7d9d8122 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -155,6 +155,9 @@ def test_atomic_write_preserves_existing_permissions(tmp_path: Path) -> None: chmod the rename would replace a user's 0o644 file with a 0o600 one, a hidden behavior change from ``Path.write_text``. """ + if os.name == "nt": + pytest.skip("POSIX permission bits are not stable on Windows") + target = tmp_path / "memory.md" target.write_text("original") target.chmod(0o644) diff --git a/tests/test_windows_support.py b/tests/test_windows_support.py new file mode 100644 index 00000000..ac8d6473 --- /dev/null +++ b/tests/test_windows_support.py @@ -0,0 +1,155 @@ +from __future__ import annotations + +from pathlib import Path + +from openchronicle import paths +from openchronicle.capture import s1_parser, windows_uia +from openchronicle.capture.window_meta import WindowMeta + + +class _Pattern: + def __init__(self, value: str) -> None: + self.Value = value + + +class _Control: + def __init__( + self, + *, + name: str, + control_type: str, + value: str = "", + children: list[_Control] | None = None, + focused: bool = False, + automation_id: str = "", + ) -> None: + self.Name = name + self.ControlTypeName = control_type + self.Value = value + self.AutomationId = automation_id + self.ClassName = "" + self.ProcessId = 123 + self.HasKeyboardFocus = focused + self._children = children or [] + + def GetChildren(self) -> list[_Control]: + return self._children + + def GetValuePattern(self) -> _Pattern: + return _Pattern(self.Value) + + +class _Auto: + def __init__(self, foreground: _Control, focused: _Control) -> None: + self._foreground = foreground + self._focused = focused + + def GetForegroundControl(self) -> _Control: + return self._foreground + + def GetFocusedControl(self) -> _Control: + return self._focused + + +def test_windows_root_defaults_to_local_app_data(monkeypatch) -> None: + monkeypatch.delenv("OPENCHRONICLE_ROOT", raising=False) + monkeypatch.setenv("LOCALAPPDATA", r"C:\Users\me\AppData\Local") + monkeypatch.setattr(paths.platform, "system", lambda: "Windows") + + assert paths.root() == Path(r"C:\Users\me\AppData\Local\OpenChronicle") + + +def test_windows_uia_provider_emits_capture_compatible_tree(monkeypatch) -> None: + address = _Control( + name="Address and search bar", + control_type="EditControl", + value="https://www.anthropic.com/news", + focused=True, + automation_id="address-edit", + ) + root = _Control( + name="Anthropic - Google Chrome", + control_type="WindowControl", + children=[ + _Control(name="Toolbar", control_type="PaneControl", children=[address]), + _Control(name="OpenChronicle notes", control_type="TextControl"), + ], + ) + monkeypatch.setattr( + windows_uia, + "active_window", + lambda: WindowMeta(app_name="Chrome", title="Anthropic", bundle_id="chrome.exe"), + ) + + provider = windows_uia.WindowsUIAProvider( + depth=5, timeout=3, auto_module=_Auto(root, address) + ) + result = provider.capture_frontmost() + + assert result is not None + assert result.metadata["platform"] == "windows" + capture = {"ax_tree": result.raw_json} + s1_parser.enrich(capture) + assert capture["url"] == "https://www.anthropic.com/news" + assert capture["focused_element"]["role"] == "AXTextField" + + +def test_s1_extracts_nested_edge_url() -> None: + capture = { + "ax_tree": { + "apps": [ + { + "name": "Edge", + "bundle_id": "msedge.exe", + "is_frontmost": True, + "windows": [ + { + "title": "Example", + "focused": True, + "elements": [ + { + "role": "AXGroup", + "children": [ + { + "role": "AXTextField", + "title": "Address and search bar", + "value": "example.com/path", + } + ], + } + ], + } + ], + } + ] + } + } + + s1_parser.enrich(capture) + + assert capture["url"] == "https://example.com/path" + + +def test_windows_polling_watcher_emits_dispatcher_compatible_events(monkeypatch) -> None: + metas = iter( + [ + WindowMeta(app_name="Chrome", title="A", bundle_id="chrome.exe"), + WindowMeta(app_name="Chrome", title="A", bundle_id="chrome.exe"), + WindowMeta(app_name="Edge", title="B", bundle_id="msedge.exe"), + ] + ) + events = [] + monkeypatch.setattr(windows_uia.platform, "system", lambda: "Windows") + monkeypatch.setattr(windows_uia, "active_window", lambda: next(metas)) + + watcher = windows_uia.WindowsPollingWatcher(interval_seconds=1) + watcher.on_event(events.append) + watcher._poll_once() + watcher._poll_once() + watcher._poll_once() + + assert [e["event_type"] for e in events] == [ + "AXApplicationActivated", + "AXValueChanged", + "AXApplicationActivated", + ] diff --git a/uv.lock b/uv.lock index 7cc439cf..5ecb99e1 100644 --- a/uv.lock +++ b/uv.lock @@ -355,6 +355,15 @@ wheels = [ { url = "https://mirrors.aliyun.com/pypi/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6" }, ] +[[package]] +name = "comtypes" +version = "1.4.16" +source = { registry = "https://mirrors.aliyun.com/pypi/simple/" } +sdist = { url = "https://mirrors.aliyun.com/pypi/packages/c6/2a/65274c13327f637ec13af8d39f2cf579d9ebe7a0e683696b5f05236d2805/comtypes-1.4.16.tar.gz", hash = "sha256:cd66d1add01265cface4df51ba1e31cd1657e04463c281c802e737e79e1ba93c" } +wheels = [ + { url = "https://mirrors.aliyun.com/pypi/packages/5f/7c/0eb685107290b6221c03c46d39214a4e42a124189691cb83ae3228257f46/comtypes-1.4.16-py3-none-any.whl", hash = "sha256:e18d85179ff12955524c5a8c3bc09cb3c0d890f1da4d7123d14244c7b78f84c8" }, +] + [[package]] name = "cryptography" version = "46.0.7" @@ -1161,6 +1170,7 @@ dependencies = [ { name = "python-frontmatter" }, { name = "rich" }, { name = "typer" }, + { name = "uiautomation", marker = "sys_platform == 'win32'" }, ] [package.dev-dependencies] @@ -1181,6 +1191,7 @@ requires-dist = [ { name = "rich", specifier = ">=13.7" }, { name = "tomli", marker = "python_full_version < '3.11'", specifier = ">=2.0" }, { name = "typer", specifier = ">=0.12" }, + { name = "uiautomation", marker = "sys_platform == 'win32'", specifier = ">=2.0" }, ] [package.metadata.requires-dev] @@ -2150,6 +2161,18 @@ wheels = [ { url = "https://mirrors.aliyun.com/pypi/packages/dc/9b/47798a6c91d8bdb567fe2698fe81e0c6b7cb7ef4d13da4114b41d239f65d/typing_inspection-0.4.2-py3-none-any.whl", hash = "sha256:4ed1cacbdc298c220f1bd249ed5287caa16f34d44ef4e9c3d0cbad5b521545e7" }, ] +[[package]] +name = "uiautomation" +version = "2.0.29" +source = { registry = "https://mirrors.aliyun.com/pypi/simple/" } +dependencies = [ + { name = "comtypes" }, +] +sdist = { url = "https://mirrors.aliyun.com/pypi/packages/bc/23/8238b5cb73e54c3618ce4d443c1830a2749264a0d61a9b61637096b8dc7a/uiautomation-2.0.29.tar.gz", hash = "sha256:3c169112043ce21065aead1d79c3baebdafc9cf03bd24ded02b2db11d423d88d" } +wheels = [ + { url = "https://mirrors.aliyun.com/pypi/packages/b0/27/b9c4b33b4129805fa2c437fa13da06c71e74213ae46da098d194d89834fe/uiautomation-2.0.29-py3-none-any.whl", hash = "sha256:5dd51c9e77e70470142a13d903be67f256c445e7cf20b47ada0ece2bdaff9f32" }, +] + [[package]] name = "urllib3" version = "2.6.3"