From 13e72709d9ea68a966c65678f5dd058d8186fd24 Mon Sep 17 00:00:00 2001 From: Ankush <43288948+AnkushMalaker@users.noreply.github.com> Date: Wed, 22 Jul 2026 21:54:51 +0530 Subject: [PATCH 1/2] feat: unify desktop capture and vault sync --- AGENTS.md | 10 + .../services/device_audio_ingest.py | 11 +- docs/README.md | 1 + docs/screenpipe.md | 67 ++++ extras/screenpipe-collector/README.md | 15 +- .../chronicle_screenpipe/collector.py | 10 +- .../chronicle_screenpipe/main.py | 7 + extras/screenpipe-collector/init.py | 53 ++- extras/screenpipe-collector/pyproject.toml | 2 +- .../screenpipe-collector/tests/test_init.py | 20 +- extras/screenpipe-collector/uv.lock | 49 ++- extras/vault-sync/README.md | 15 +- extras/vault-sync/desktop_core.py | 177 ++++++++++ extras/vault-sync/menu_linux.py | 313 ++++++++++++------ extras/vault-sync/menu_vault.py | 175 +--------- extras/vault-sync/service.py | 27 +- extras/vault-sync/start.sh | 5 +- extras/vault-sync/tests/test_menu_linux.py | 72 ++++ extras/vault-sync/vault_core.py | 8 + 19 files changed, 731 insertions(+), 306 deletions(-) create mode 100644 docs/screenpipe.md create mode 100644 extras/vault-sync/desktop_core.py create mode 100644 extras/vault-sync/tests/test_menu_linux.py diff --git a/AGENTS.md b/AGENTS.md index 5b65b3e4..1f22fc0b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -602,11 +602,21 @@ For detailed technical documentation, see: - **[@docs/init-system.md](docs/init-system.md)**: Initialization system and service management - **[@docs/ssl-certificates.md](docs/ssl-certificates.md)**: HTTPS/SSL setup details - **[@docs/podman.md](docs/podman.md)**: Running with Podman instead of Docker (engine selection, rootless/GPU setup) +- **[@docs/screenpipe.md](docs/screenpipe.md)**: ScreenPipe capture-node architecture, services, desktop controls, and troubleshooting - **[@docs/audio-pipeline-architecture.md](docs/audio-pipeline-architecture.md)**: Audio pipeline design - **[@docs/backend/auth.md](docs/backend/auth.md)**: Authentication architecture - **[@docs/backend/memories.md](docs/backend/memories.md)**: Memory system documentation - **[@docs/backend/plugin-development-guide.md](docs/backend/plugin-development-guide.md)**: Plugin development guide +### ScreenPipe Capture Nodes + +Before changing ScreenPipe ingestion, the desktop tray, or capture-node services, read +[@docs/screenpipe.md](docs/screenpipe.md). ScreenPipe owns the high-volume local capture +store; Chronicle's companion sends compact activity metadata and serves bounded +snapshot/OCR requests. The desktop entry point is shared across macOS and Linux, with +platform UI adapters over common state, logging, and vault-sync code. The ScreenPipe UI +is an optional on-demand viewer and must not be required for background capture. + ## Robot Framework Testing **IMPORTANT: When writing or modifying Robot Framework tests, you MUST follow the testing guidelines.** diff --git a/backends/advanced/src/advanced_omi_backend/services/device_audio_ingest.py b/backends/advanced/src/advanced_omi_backend/services/device_audio_ingest.py index f794b216..fd512a23 100644 --- a/backends/advanced/src/advanced_omi_backend/services/device_audio_ingest.py +++ b/backends/advanced/src/advanced_omi_backend/services/device_audio_ingest.py @@ -107,11 +107,12 @@ async def process_device_audio() -> dict[str, Any]: .sort([("source_id", 1), ("captured_at", 1)]) .to_list() ) - by_source: dict[tuple[str, str], list[DeviceInputItem]] = {} + by_source: dict[tuple[str, str, str], list[DeviceInputItem]] = {} for item in pending: - by_source.setdefault((item.user_id, item.source_id), []).append(item) + direction = str(item.metadata.get("direction", "unknown")) + by_source.setdefault((item.user_id, item.source_id, direction), []).append(item) processed = 0 - for (user_id, source_id), source_items in by_source.items(): + for (user_id, source_id, direction), source_items in by_source.items(): try: user = await User.get(PydanticObjectId(user_id)) except Exception: @@ -136,7 +137,7 @@ async def process_device_audio() -> dict[str, Any]: result = await upload_and_process_audio_files( user, [UploadFile(file=handle, filename=output.name)], - device_name=source_id, + device_name=f"{source_id}-{direction}", source="screenpipe", ) if ( @@ -153,7 +154,7 @@ async def process_device_audio() -> dict[str, Any]: conversation.created_at = min( _as_utc(item.captured_at) for item in session ) - conversation.external_source_id = f"screenpipe:{source_id}:{session[0].source_item_id}-{session[-1].source_item_id}" + conversation.external_source_id = f"screenpipe:{source_id}:{direction}:{session[0].source_item_id}-{session[-1].source_item_id}" conversation.external_source_type = "screenpipe" await conversation.save() for item in session: diff --git a/docs/README.md b/docs/README.md index 45c1d88f..66f57fa1 100644 --- a/docs/README.md +++ b/docs/README.md @@ -10,6 +10,7 @@ day-to-day operation, and use [AGENTS.md](../AGENTS.md) for development conventi - [Testing and coverage](testing.md): fast Python lanes, coverage reports, and integration tests - [Audio pipeline](audio-pipeline-architecture.md): session, transcription, and memory flow - [Initialization system](init-system.md): setup wizard and service orchestration +- [ScreenPipe capture nodes](screenpipe.md): local capture, Chronicle ingestion, desktop controls, and logs - [Podman](podman.md): rootless containers, GPU access, and engine migration - [SSL certificates](ssl-certificates.md): HTTPS setup and certificate trust diff --git a/docs/screenpipe.md b/docs/screenpipe.md new file mode 100644 index 00000000..fe9893eb --- /dev/null +++ b/docs/screenpipe.md @@ -0,0 +1,67 @@ +# ScreenPipe capture nodes + +ScreenPipe is Chronicle's local, cross-platform screen/activity capture interface. It +owns the high-volume frame, OCR, accessibility, and optional audio store on each +computer. Chronicle does not mirror that entire store: the companion collector sends +compact application/window transitions and can fulfill bounded snapshot/OCR requests +for the timeline. + +## Component boundaries + +- `screenpipe record` captures and retains source data locally. +- `extras/screenpipe-collector/` pairs the node with Chronicle, uploads activity + metadata, checkpoints progress, and answers bounded media requests. +- Chronicle's backend stores timeline activity and only the media selected for a + durable Chronicle view or memory. +- `extras/vault-sync/` is the single Chronicle desktop entry point. `main.py` selects + the macOS menu bar or Linux system tray adapter; `desktop_core.py` contains their + shared state, logging, and vault synchronization. +- ScreenPipe's own desktop UI is an optional, on-demand local timeline viewer. It is + not the recorder and should not autostart or launch a second recording process. + +Do not scan or upload the complete frame stream, and do not copy ScreenPipe's SQLite +database into Chronicle as a second source of truth. + +## Capture mode + +Microphone and system-output capture are separate sources. A capture node may record +neither, system output only, microphone only, or both. Use +`--audio-transcription-engine disabled` in every enabled mode because Chronicle owns +speech-to-text. ScreenPipe's `--use-system-default-audio true` follows and enrolls both +the default input and output; system-only or microphone-only modes therefore require +an explicit `--audio-device "... (output)"` or `--audio-device "... (input)"` with +`--use-system-default-audio false`. The collector preserves the source direction when +it sends audio to Chronicle. + +The setup wizard's capture-node option delegates to +`extras/screenpipe-collector/init.py`. Pairing and standalone commands are documented +in the [companion README](../extras/screenpipe-collector/README.md). + +## Desktop controls and logs + +On Linux the Chronicle tray controls `screenpipe.service` and +`chronicle-screenpipe.service`, shows local frame/storage statistics, toggles capture +modes, and offers timed pauses. Both Linux and macOS expose **View Logs** for the +current Chronicle desktop process. System-service history remains in the native +service manager: + +```bash +systemctl --user status screenpipe.service chronicle-screenpipe.service chronicle-desktop.service +journalctl --user -u screenpipe.service -u chronicle-screenpipe.service -u chronicle-desktop.service +``` + +On macOS, use `extras/vault-sync/start.sh logs` for the installed desktop service. + +The Linux ScreenPipe UI launcher may set rendering/onboarding environment required by +the locally installed build, but its desktop autostart entry should remain disabled. +Open that UI manually only when the full local ScreenPipe timeline is needed; the +recorder and Chronicle collector remain independently managed background services. + +## Implementation map + +- `extras/screenpipe-collector/`: collector CLI, local client, checkpoints, and service installation +- `backends/advanced/src/routers/device_activity.py`: capture-node ingestion API +- `backends/advanced/src/services/device_activity_service.py`: activity persistence and media requests +- `extras/vault-sync/main.py`: cross-platform desktop entry point +- `extras/vault-sync/desktop_core.py`: shared desktop state, logs, and vault sync +- `extras/vault-sync/menu_linux.py` and `menu_vault.py`: platform-specific UI adapters diff --git a/extras/screenpipe-collector/README.md b/extras/screenpipe-collector/README.md index 596cde94..f0d61bb4 100644 --- a/extras/screenpipe-collector/README.md +++ b/extras/screenpipe-collector/README.md @@ -1,6 +1,9 @@ # Chronicle ScreenPipe companion -This service keeps ScreenPipe as the local capture store while forwarding completed audio chunks and compact application/window transitions to Chronicle. Screen pixels and OCR are retrieved only for bounded jobs requested by Chronicle. +This service keeps ScreenPipe as the local capture store while forwarding compact +application/window transitions and, only when enabled, completed audio chunks to +Chronicle. Screen pixels and OCR are retrieved only for bounded jobs requested by +Chronicle. See the full [capture-node architecture](../../docs/screenpipe.md). ## Pair and run @@ -13,7 +16,9 @@ This service keeps ScreenPipe as the local capture store while forwarding comple --code PAIRING_CODE ``` -3. Start ScreenPipe with Chronicle's privacy-oriented defaults: +3. Start ScreenPipe with Chronicle's privacy-oriented defaults. The example records + both the default microphone and system output while leaving transcription to + Chronicle: ```bash screenpipe record --audio-transcription-engine disabled \ @@ -30,6 +35,12 @@ This service keeps ScreenPipe as the local capture store while forwarding comple Set `SCREENPIPE_API_KEY` for both ScreenPipe and the pairing command so the companion can authenticate bounded local OCR queries. + For system audio without the microphone, replace + `--use-system-default-audio true` with + `--use-system-default-audio false --audio-device "DEVICE (output)"`. Discover the + exact platform device names with `screenpipe audio list --output json`. Use + `--disable-audio` only for screen-only capture. + 4. Run the companion: ```bash diff --git a/extras/screenpipe-collector/chronicle_screenpipe/collector.py b/extras/screenpipe-collector/chronicle_screenpipe/collector.py index 766e33f7..e5c51651 100644 --- a/extras/screenpipe-collector/chronicle_screenpipe/collector.py +++ b/extras/screenpipe-collector/chronicle_screenpipe/collector.py @@ -25,6 +25,7 @@ class Config: screenpipe_dir: Path screenpipe_url: str = "http://127.0.0.1:3030" screenpipe_token: str | None = None + forward_audio: str = "both" poll_seconds: float = 5.0 activity_debounce_seconds: float = 10.0 @@ -245,6 +246,13 @@ def collect_audio(self, connection: sqlite3.Connection) -> int: sent = 0 for row in rows: path = Path(row["file_path"]) + direction = infer_audio_direction(str(path)) + if self.config.forward_audio == "none" or ( + self.config.forward_audio != "both" + and direction != self.config.forward_audio + ): + self.checkpoints.set("audio", row["id"]) + continue if not path.is_file(): captured = timestamp_seconds(iso_timestamp(row["timestamp"])) if time.time() - captured < 120: @@ -284,7 +292,7 @@ def collect_audio(self, connection: sqlite3.Connection) -> int: "captured_at": iso_timestamp(row["timestamp"]), "duration_seconds": str(audio_duration(path)), "device_name": path.stem, - "direction": infer_audio_direction(str(path)), + "direction": direction, "content_hash": digest, }, files={"file": (path.name, handle, content_type)}, diff --git a/extras/screenpipe-collector/chronicle_screenpipe/main.py b/extras/screenpipe-collector/chronicle_screenpipe/main.py index fe8de1bd..fca0f7f4 100644 --- a/extras/screenpipe-collector/chronicle_screenpipe/main.py +++ b/extras/screenpipe-collector/chronicle_screenpipe/main.py @@ -54,6 +54,7 @@ def pair(args: argparse.Namespace) -> None: "screenpipe_dir": str(Path(args.screenpipe_dir).expanduser()), "screenpipe_url": args.screenpipe_url, "screenpipe_token": args.screenpipe_token, + "forward_audio": args.forward_audio, }, indent=2, ), @@ -99,6 +100,12 @@ def main() -> None: default=os.getenv("SCREENPIPE_API_KEY"), help="token used by ScreenPipe's authenticated local API", ) + pair_parser.add_argument( + "--forward-audio", + choices=("none", "output", "input", "both"), + default="both", + help="which locally captured ScreenPipe audio sources Chronicle receives", + ) sub.add_parser("run") sub.add_parser("install-service") args = parser.parse_args() diff --git a/extras/screenpipe-collector/init.py b/extras/screenpipe-collector/init.py index 8328b4e3..20ac5fac 100644 --- a/extras/screenpipe-collector/init.py +++ b/extras/screenpipe-collector/init.py @@ -4,8 +4,10 @@ from __future__ import annotations import argparse +import json import os import secrets +import shlex import shutil import subprocess from pathlib import Path @@ -23,14 +25,37 @@ def screenpipe_command() -> str | None: return shutil.which("screenpipe") -def write_screenpipe_unit(binary: str, api_key: str) -> Path: +def list_audio_devices(binary: str) -> list[str]: + result = subprocess.run( + [binary, "audio", "list", "--output", "json"], + capture_output=True, + text=True, + check=True, + ) + return [entry["name"] for entry in json.loads(result.stdout)["data"]] + + +def audio_arguments(mode: str, devices: list[str]) -> list[str]: + if mode == "off": + return ["--disable-audio"] + if mode == "both": + return ["--use-system-default-audio", "true"] + suffix = "(output)" if mode == "system" else "(input)" + matches = [device for device in devices if device.lower().endswith(suffix)] + if not matches: + raise ValueError(f"no {mode} audio device is available") + return ["--use-system-default-audio", "false", "--audio-device", matches[0]] + + +def write_screenpipe_unit( + binary: str, api_key: str, audio_mode: str = "both", devices: list[str] | None = None +) -> Path: SYSTEMD_USER_DIR.mkdir(parents=True, exist_ok=True) path = SYSTEMD_USER_DIR / "screenpipe.service" args = [ binary, "record", "--audio-transcription-engine", "disabled", - "--use-system-default-audio", "true", "--use-all-monitors", "true", "--use-pii-removal", "true", "--disable-keyboard-capture", @@ -44,11 +69,12 @@ def write_screenpipe_unit(binary: str, api_key: str) -> Path: "--retention-mode", "media", "--api-auth", "true", ] + args.extend(audio_arguments(audio_mode, devices or [])) path.write_text( "[Unit]\nDescription=ScreenPipe local recorder for Chronicle\n" "After=graphical-session.target\n\n[Service]\nType=simple\n" f"Environment=SCREENPIPE_API_KEY={api_key}\n" - f"ExecStart={' '.join(args)}\nRestart=on-failure\nRestartSec=5\n\n" + f"ExecStart={shlex.join(args)}\nRestart=on-failure\nRestartSec=5\n\n" "[Install]\nWantedBy=default.target\n", encoding="utf-8", ) @@ -76,6 +102,24 @@ def main() -> None: console.print(f"[green]✅[/green] ScreenPipe detected: [cyan]{binary}[/cyan]") backend = Prompt.ask("Chronicle backend URL", default=args.backend or "http://127.0.0.1:8000") + devices = list_audio_devices(binary) + audio_mode = Prompt.ask( + "Local audio capture", + choices=("off", "system", "mic", "both"), + default="system", + ) + forward_default = ( + "none" + if audio_mode == "off" + else "output" + if audio_mode == "system" + else audio_mode.replace("mic", "input") + ) + forward_audio = Prompt.ask( + "Audio sent to Chronicle", + choices=("none", "output", "input", "both"), + default=forward_default, + ) console.print( "Open Chronicle → Timeline → Sources and create a pairing code. " "The code expires after 10 minutes." @@ -92,8 +136,9 @@ def main() -> None: "--screenpipe-dir", str(Path.home() / ".screenpipe"), "--screenpipe-url", "http://127.0.0.1:3030", "--screenpipe-token", api_key, + "--forward-audio", forward_audio, ) - write_screenpipe_unit(binary, api_key) + write_screenpipe_unit(binary, api_key, audio_mode, devices) run("uv", "run", "--project", str(PROJECT), "chronicle-screenpipe", "install-service") run("systemctl", "--user", "daemon-reload") run("systemctl", "--user", "enable", "--now", "screenpipe.service") diff --git a/extras/screenpipe-collector/pyproject.toml b/extras/screenpipe-collector/pyproject.toml index 2dc33448..c0007b85 100644 --- a/extras/screenpipe-collector/pyproject.toml +++ b/extras/screenpipe-collector/pyproject.toml @@ -3,7 +3,7 @@ name = "chronicle-screenpipe-collector" version = "0.1.0" description = "Chronicle companion for a local ScreenPipe recorder" requires-python = ">=3.11" -dependencies = ["httpx>=0.28,<1"] +dependencies = ["httpx>=0.28,<1", "rich>=13,<15"] [project.scripts] chronicle-screenpipe = "chronicle_screenpipe.main:main" diff --git a/extras/screenpipe-collector/tests/test_init.py b/extras/screenpipe-collector/tests/test_init.py index 8ecafb45..415c063a 100644 --- a/extras/screenpipe-collector/tests/test_init.py +++ b/extras/screenpipe-collector/tests/test_init.py @@ -10,10 +10,28 @@ def test_screenpipe_unit_uses_privacy_defaults_and_api_auth(tmp_path, monkeypatch): monkeypatch.setattr(MODULE, "SYSTEMD_USER_DIR", tmp_path) - path = MODULE.write_screenpipe_unit("/usr/bin/screenpipe", "local-key") + path = MODULE.write_screenpipe_unit( + "/usr/bin/screenpipe", + "local-key", + "system", + ["Speakers (output)"], + ) text = path.read_text() assert "--audio-transcription-engine disabled" in text assert "--disable-keyboard-capture" in text assert "--disable-clipboard-capture" in text assert "--api-auth true" in text assert "Environment=SCREENPIPE_API_KEY=local-key" in text + assert "--use-system-default-audio false" in text + assert "--audio-device 'Speakers (output)'" in text + + +def test_audio_arguments_keep_microphone_and_system_independent(): + devices = ["Mic (input)", "Speakers (output)"] + + assert MODULE.audio_arguments("system", devices)[-1] == "Speakers (output)" + assert MODULE.audio_arguments("mic", devices)[-1] == "Mic (input)" + assert MODULE.audio_arguments("both", devices) == [ + "--use-system-default-audio", + "true", + ] diff --git a/extras/screenpipe-collector/uv.lock b/extras/screenpipe-collector/uv.lock index c4362d30..abe0d6e8 100644 --- a/extras/screenpipe-collector/uv.lock +++ b/extras/screenpipe-collector/uv.lock @@ -30,10 +30,14 @@ version = "0.1.0" source = { editable = "." } dependencies = [ { name = "httpx" }, + { name = "rich" }, ] [package.metadata] -requires-dist = [{ name = "httpx", specifier = ">=0.28,<1" }] +requires-dist = [ + { name = "httpx", specifier = ">=0.28,<1" }, + { name = "rich", specifier = ">=13,<15" }, +] [[package]] name = "h11" @@ -81,6 +85,49 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/1e/5e/d4e9f1a599fb8e573b7b87160658329fbf28d19eac2718f51fc3def3aa5a/idna-3.18-py3-none-any.whl", hash = "sha256:7f952cbe720b688055e3f87de14f5c3e5fdaa8bc3928985c4077ca689de849a2", size = 65455, upload-time = "2026-06-02T14:34:06.319Z" }, ] +[[package]] +name = "markdown-it-py" +version = "4.2.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "mdurl" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/06/ff/7841249c247aa650a76b9ee4bbaeae59370dc8bfd2f6c01f3630c35eb134/markdown_it_py-4.2.0.tar.gz", hash = "sha256:04a21681d6fbb623de53f6f364d352309d4094dd4194040a10fd51833e418d49", size = 82454, upload-time = "2026-05-07T12:08:28.36Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b3/81/4da04ced5a082363ecfa159c010d200ecbd959ae410c10c0264a38cac0f5/markdown_it_py-4.2.0-py3-none-any.whl", hash = "sha256:9f7ebbcd14fe59494226453aed97c1070d83f8d24b6fc3a3bcf9a38092641c4a", size = 91687, upload-time = "2026-05-07T12:08:27.182Z" }, +] + +[[package]] +name = "mdurl" +version = "0.1.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d6/54/cfe61301667036ec958cb99bd3efefba235e65cdeb9c84d24a8293ba1d90/mdurl-0.1.2.tar.gz", hash = "sha256:bb413d29f5eea38f31dd4754dd7377d4465116fb207585f97bf925588687c1ba", size = 8729, upload-time = "2022-08-14T12:40:10.846Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b3/38/89ba8ad64ae25be8de66a6d463314cf1eb366222074cfda9ee839c56a4b4/mdurl-0.1.2-py3-none-any.whl", hash = "sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8", size = 9979, upload-time = "2022-08-14T12:40:09.779Z" }, +] + +[[package]] +name = "pygments" +version = "2.20.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/c3/b2/bc9c9196916376152d655522fdcebac55e66de6603a76a02bca1b6414f6c/pygments-2.20.0.tar.gz", hash = "sha256:6757cd03768053ff99f3039c1a36d6c0aa0b263438fcab17520b30a303a82b5f", size = 4955991, upload-time = "2026-03-29T13:29:33.898Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/f4/7e/a72dd26f3b0f4f2bf1dd8923c85f7ceb43172af56d63c7383eb62b332364/pygments-2.20.0-py3-none-any.whl", hash = "sha256:81a9e26dd42fd28a23a2d169d86d7ac03b46e2f8b59ed4698fb4785f946d0176", size = 1231151, upload-time = "2026-03-29T13:29:30.038Z" }, +] + +[[package]] +name = "rich" +version = "14.3.4" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "markdown-it-py" }, + { name = "pygments" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/e9/67/cae617f1351490c25a4b8ac3b8b63a4dda609295d8222bad12242dfdc629/rich-14.3.4.tar.gz", hash = "sha256:817e02727f2b25b40ef56f5aa2217f400c8489f79ca8f46ea2b70dd5e14558a9", size = 230524, upload-time = "2026-04-11T02:57:45.419Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b3/76/6d163cfac87b632216f71879e6b2cf17163f773ff59c00b5ff4900a80fa3/rich-14.3.4-py3-none-any.whl", hash = "sha256:07e7adb4690f68864777b1450859253bed81a99a31ac321ac1817b2313558952", size = 310480, upload-time = "2026-04-11T02:57:47.484Z" }, +] + [[package]] name = "typing-extensions" version = "4.16.0" diff --git a/extras/vault-sync/README.md b/extras/vault-sync/README.md index 0d5da42f..d23d834c 100644 --- a/extras/vault-sync/README.md +++ b/extras/vault-sync/README.md @@ -30,9 +30,13 @@ On Arch/CachyOS: sudo pacman -S obsidian syncthing ``` -The Linux tray also shows local ScreenPipe frame/audio counts and storage use, and +Both platforms use the same desktop entry point and share their state, logging, and +vault-sync orchestration; only the native menu UI is platform-specific. Both provide +**View Logs** for recent desktop activity. The Linux tray also shows local ScreenPipe frame/audio counts and storage use, and provides start, stop, and restart controls for `screenpipe.service` and -`chronicle-screenpipe.service`. +`chronicle-screenpipe.service`. Its top-level Settings menu can enable or disable audio +and screen capture, while the ScreenPipe menu can pause capture for 5, 15, or 30 +minutes, or 1, 2, or 8 hours. You also need the Chronicle **server** side running with vault sync enabled — see [Server setup](#server-setup-once) below. @@ -47,8 +51,11 @@ cp .env.template .env ./start.sh # run in the foreground (a ◈ icon appears in the menu bar) ``` -The Mac needs only three things: `BACKEND_URL`, `AUTH_USERNAME` (your Chronicle email), -and `AUTH_PASSWORD`. The pairing broker hands back everything else (server device id, +Vault sync first reads the repository root `.env`, then the optional +`extras/vault-sync/.env` as an override. You can use the root `ADMIN_EMAIL` and +`ADMIN_PASSWORD`, or set `AUTH_USERNAME` and `AUTH_PASSWORD` specifically for the +desktop sync app. It also needs `BACKEND_URL` unless backend discovery works on the +machine. The pairing broker hands back everything else (server device id, sync address) — you never set `VAULT_SYNC_*` on the Mac; those are server-only. > **macOS + Tailscale: set `BACKEND_URL` explicitly.** Auto-discovery (minidisc) needs diff --git a/extras/vault-sync/desktop_core.py b/extras/vault-sync/desktop_core.py new file mode 100644 index 00000000..8cd7883f --- /dev/null +++ b/extras/vault-sync/desktop_core.py @@ -0,0 +1,177 @@ +"""Platform-independent state, logging, and vault-sync orchestration.""" + +import logging +import threading +from collections import deque +from dataclasses import dataclass, field +from typing import Optional + +import httpx + +from syncthing_manager import SyncthingManager +from vault_core import VaultSyncConfig, broker_pair, get_jwt_token, save_vault_dir + +logger = logging.getLogger(__name__) + + +class MemoryLogHandler(logging.Handler): + """Keep recent application log lines for display by either desktop UI.""" + + def __init__(self, capacity: int = 500) -> None: + super().__init__() + self.lines: deque[str] = deque(maxlen=capacity) + + def emit(self, record: logging.LogRecord) -> None: + try: + self.lines.append(self.format(record)) + except Exception: + self.handleError(record) + + +log_buffer = MemoryLogHandler() + + +def configure_logging() -> None: + """Configure the shared desktop log buffer without adding it twice.""" + log_format = "%(asctime)s %(levelname)s %(name)s: %(message)s" + logging.basicConfig(format=log_format, level=logging.INFO) + log_buffer.setFormatter(logging.Formatter(log_format)) + root = logging.getLogger() + if log_buffer not in root.handlers: + root.addHandler(log_buffer) + logging.getLogger("httpx").setLevel(logging.WARNING) + + +@dataclass +class SharedState: + """Thread-safe state shared between a platform UI and its worker thread.""" + + _lock: threading.Lock = field(default_factory=threading.Lock, repr=False) + status: str = "idle" + error: Optional[str] = None + connected: bool = False + completion: Optional[float] = None + folder_error: Optional[str] = None + folder_id: Optional[str] = None + vault_dir: str = "" + + def snapshot(self) -> dict: + with self._lock: + return { + key: value for key, value in vars(self).items() if key != "_lock" + } + + def update(self, **values) -> None: + with self._lock: + for key, value in values.items(): + setattr(self, key, value) + + +class VaultSyncManager: + """Coordinate local Syncthing and the Chronicle pairing handshake.""" + + def __init__(self, state: SharedState) -> None: + self.state = state + self.config = VaultSyncConfig.from_env() + self.syncthing = SyncthingManager() + self.state.update(vault_dir=self.config.local_vault_dir) + self._lock = threading.Lock() + self._logged_errors: set[str] = set() + + def pair_async(self) -> None: + threading.Thread(target=self._pair, daemon=True).start() + + def _pair(self) -> None: + if not self._lock.acquire(blocking=False): + logger.info("Pair already in progress") + return + try: + cfg = self.config + if not cfg.auth_username or not cfg.auth_password: + error = ( + "Chronicle login is not configured. Set AUTH_USERNAME and " + "AUTH_PASSWORD (or ADMIN_EMAIL and ADMIN_PASSWORD) in " + f"{VaultSyncConfig.root_env_file()} or " + f"{VaultSyncConfig.local_env_file()}, then restart the desktop service." + ) + logger.error(error) + self.state.update( + status="error", error=error + ) + return + self.state.update(status="starting", error=None) + self.syncthing.start() + local_id = self.syncthing.device_id() + self.state.update(status="pairing") + token = get_jwt_token(cfg.auth_username, cfg.auth_password, cfg.backend_url) + if not token: + error = ( + f"Chronicle login failed for {cfg.auth_username} at {cfg.backend_url}. " + "Check the credentials in the configured .env file." + ) + logger.error(error) + self.state.update(status="error", error=error) + return + info = broker_pair(cfg.backend_url, token, local_id, cfg.device_name) + sync_address = info.get("sync_address") or "" + self.syncthing.ensure_server_device( + info["server_device_id"], + "Chronicle Server", + [sync_address] if sync_address else ["dynamic"], + ) + self.syncthing.ensure_folder( + folder_id=info["folder_id"], + path=cfg.local_vault_dir, + label=info.get("folder_label", "Chronicle Vault"), + server_device_id=info["server_device_id"], + self_device_id=local_id, + ) + self.state.update(status="syncing", folder_id=info["folder_id"], error=None) + logger.info("Paired. Folder %s -> %s", info["folder_id"], cfg.local_vault_dir) + except httpx.HTTPStatusError as error: + detail = error.response.text[:200] + self.state.update(status="error", error=f"Pair failed: {detail}") + logger.error("Pair failed: %s", detail) + except Exception as error: # noqa: BLE001 - the desktop UI must surface it + self.state.update(status="error", error=str(error)) + logger.exception("Pair error") + finally: + self._lock.release() + + def set_vault_dir(self, path: str) -> None: + save_vault_dir(path) + self.config.local_vault_dir = path + self.state.update(vault_dir=path) + logger.info("Vault folder set to %s — re-pairing", path) + self.pair_async() + + def refresh_status(self) -> None: + if not self.syncthing.is_running(): + return + snap = self.state.snapshot() + completion = None + folder_error = None + if snap["folder_id"]: + folder = self.syncthing.folder_status(snap["folder_id"]) + completion = folder["completion"] + folder_error = folder["error"] + + detailed = self.syncthing.collect_errors() + for message in detailed: + if message not in self._logged_errors: + logger.error("Syncthing: %s", message) + self._logged_errors = set(detailed) + if detailed: + folder_error = ( + detailed[0] + if len(detailed) == 1 + else f"{detailed[0]} (+{len(detailed) - 1} more)" + ) + self.state.update( + connected=self.syncthing.connection_count() > 0, + completion=completion, + folder_error=folder_error, + ) + + def shutdown(self) -> None: + self.syncthing.stop() diff --git a/extras/vault-sync/menu_linux.py b/extras/vault-sync/menu_linux.py index 7050efc3..73f07fdd 100644 --- a/extras/vault-sync/menu_linux.py +++ b/extras/vault-sync/menu_linux.py @@ -1,123 +1,37 @@ """KDE/Linux system tray for Chronicle capture and vault sync.""" +import json import logging +import shlex import sqlite3 import subprocess import sys -import threading -from dataclasses import dataclass, field from pathlib import Path -from typing import Optional from urllib.parse import quote -import httpx from dotenv import load_dotenv -from PySide6.QtCore import QTimer -from PySide6.QtGui import QAction, QDesktopServices, QIcon -from PySide6.QtWidgets import QApplication, QFileDialog, QMenu, QSystemTrayIcon -from PySide6.QtCore import QUrl +from PySide6.QtCore import QTimer, QUrl +from PySide6.QtGui import QAction, QActionGroup, QDesktopServices, QIcon +from PySide6.QtWidgets import ( + QApplication, + QDialog, + QDialogButtonBox, + QFileDialog, + QLabel, + QMenu, + QPlainTextEdit, + QSystemTrayIcon, + QVBoxLayout, +) -from syncthing_manager import SyncthingManager -from vault_core import VaultSyncConfig, broker_pair, get_jwt_token, save_vault_dir +from desktop_core import SharedState, VaultSyncManager, configure_logging, log_buffer logger = logging.getLogger(__name__) SCREENPIPE_DB = Path.home() / ".screenpipe/db.sqlite" +SCREENPIPE_UNIT = Path.home() / ".config/systemd/user/screenpipe.service" load_dotenv(Path(__file__).resolve().parent / ".env") -@dataclass -class SharedState: - _lock: threading.Lock = field(default_factory=threading.Lock, repr=False) - status: str = "idle" - error: Optional[str] = None - connected: bool = False - completion: Optional[float] = None - folder_error: Optional[str] = None - folder_id: Optional[str] = None - vault_dir: str = "" - - def snapshot(self) -> dict: - with self._lock: - return { - key: value for key, value in vars(self).items() if key != "_lock" - } - - def update(self, **values) -> None: - with self._lock: - for key, value in values.items(): - setattr(self, key, value) - - -class VaultSyncManager: - def __init__(self, state: SharedState) -> None: - self.state = state - self.config = VaultSyncConfig.from_env() - self.syncthing = SyncthingManager() - self.state.update(vault_dir=self.config.local_vault_dir) - self._lock = threading.Lock() - - def pair_async(self) -> None: - threading.Thread(target=self._pair, daemon=True).start() - - def _pair(self) -> None: - if not self._lock.acquire(blocking=False): - return - try: - cfg = self.config - if not cfg.auth_username or not cfg.auth_password: - self.state.update(status="error", error="set Chronicle login in .env") - return - self.state.update(status="starting", error=None) - self.syncthing.start() - self.state.update(status="pairing") - token = get_jwt_token(cfg.auth_username, cfg.auth_password, cfg.backend_url) - if not token: - self.state.update(status="error", error="backend authentication failed") - return - info = broker_pair( - cfg.backend_url, token, self.syncthing.device_id(), cfg.device_name - ) - self.syncthing.ensure_server_device( - info["server_device_id"], - "Chronicle Server", - [info["sync_address"]] if info.get("sync_address") else ["dynamic"], - ) - self.syncthing.ensure_folder( - info["folder_id"], cfg.local_vault_dir, - info.get("folder_label", "Chronicle Vault"), - info["server_device_id"], self.syncthing.device_id(), - ) - self.state.update(status="syncing", folder_id=info["folder_id"], error=None) - except (OSError, httpx.HTTPError, RuntimeError) as error: - logger.exception("Vault pairing failed") - self.state.update(status="error", error=str(error)) - finally: - self._lock.release() - - def set_vault_dir(self, path: str) -> None: - save_vault_dir(path) - self.config.local_vault_dir = path - self.state.update(vault_dir=path) - self.pair_async() - - def refresh_status(self) -> None: - if not self.syncthing.is_running(): - return - snap = self.state.snapshot() - status = ( - self.syncthing.folder_status(snap["folder_id"]) - if snap["folder_id"] else {} - ) - self.state.update( - connected=self.syncthing.connection_count() > 0, - completion=status.get("completion"), - folder_error=status.get("error"), - ) - - def shutdown(self) -> None: - self.syncthing.stop() - - def _unit_state(name: str) -> str: result = subprocess.run( ["systemctl", "--user", "is-active", name], capture_output=True, text=True @@ -150,12 +64,98 @@ def _screenpipe_stats() -> str: return f"Stats unavailable: {error}" +def _capture_settings(path: Path = SCREENPIPE_UNIT) -> tuple[str, bool]: + """Return the audio mode and whether screen capture is enabled.""" + text = path.read_text(encoding="utf-8") + exec_start = next(line for line in text.splitlines() if line.startswith("ExecStart=")) + args = shlex.split(exec_start.removeprefix("ExecStart=")) + if "--disable-audio" in args: + audio_mode = "off" + else: + devices = _argument_values(args, "--audio-device") + follows_defaults = _argument_value(args, "--use-system-default-audio") + if follows_defaults == "true" or (follows_defaults is None and not devices): + return "both", "--disable-vision" not in args + has_input = any(device.lower().endswith("(input)") for device in devices) + has_output = any(device.lower().endswith("(output)") for device in devices) + audio_mode = "both" if has_input and has_output else "mic" if has_input else "system" + return audio_mode, "--disable-vision" not in args + + +def _argument_value(args: list[str], option: str) -> str | None: + values = _argument_values(args, option) + return values[-1] if values else None + + +def _argument_values(args: list[str], option: str) -> list[str]: + return [args[index + 1] for index, value in enumerate(args[:-1]) if value == option] + + +def _without_options(args: list[str], options: set[str]) -> list[str]: + result = [] + skip = False + for value in args: + if skip: + skip = False + continue + if value in options: + skip = True + continue + result.append(value) + return result + + +def _audio_devices() -> list[str]: + result = subprocess.run( + ["screenpipe", "audio", "list", "--output", "json"], + capture_output=True, + text=True, + check=True, + ) + return [entry["name"] for entry in json.loads(result.stdout)["data"]] + + +def _save_capture_settings( + audio_mode: str, + screen_enabled: bool, + path: Path = SCREENPIPE_UNIT, + audio_devices: list[str] | None = None, +) -> None: + """Persist independent audio-source and screen settings in ScreenPipe's unit.""" + if audio_mode not in {"off", "system", "mic", "both"}: + raise ValueError(f"unsupported audio mode: {audio_mode}") + text = path.read_text(encoding="utf-8") + lines = text.splitlines() + index = next(i for i, line in enumerate(lines) if line.startswith("ExecStart=")) + args = shlex.split(lines[index].removeprefix("ExecStart=")) + args = _without_options(args, {"--audio-device", "--use-system-default-audio"}) + args = [arg for arg in args if arg not in {"--disable-audio", "--disable-vision"}] + if audio_mode == "off": + args.append("--disable-audio") + elif audio_mode == "both": + args.extend(["--use-system-default-audio", "true"]) + else: + suffix = f"({ 'output' if audio_mode == 'system' else 'input' })" + matching = [name for name in (audio_devices or _audio_devices()) if name.lower().endswith(suffix)] + if not matching: + raise ValueError(f"no {audio_mode} audio device is available") + args.extend(["--use-system-default-audio", "false", "--audio-device", matching[0]]) + if not screen_enabled: + args.append("--disable-vision") + lines[index] = f"ExecStart={shlex.join(args)}" + path.write_text("\n".join(lines) + ("\n" if text.endswith("\n") else ""), encoding="utf-8") + + class ChronicleTray(QSystemTrayIcon): def __init__(self, state: SharedState, manager: VaultSyncManager) -> None: icon = QIcon.fromTheme("view-calendar-timeline", QIcon.fromTheme("folder-sync")) super().__init__(icon) self.state = state self.manager = manager + self.service_actions: dict[str, dict[str, QAction]] = {} + self.pause_timer = QTimer(self) + self.pause_timer.setSingleShot(True) + self.pause_timer.timeout.connect(lambda: self.service("start", "screenpipe.service")) menu = QMenu() self.capture_status = menu.addAction("ScreenPipe: checking…") self.collector_status = menu.addAction("Chronicle collector: checking…") @@ -163,8 +163,29 @@ def __init__(self, state: SharedState, manager: VaultSyncManager) -> None: for item in (self.capture_status, self.collector_status, self.stats): item.setEnabled(False) menu.addSeparator() - self._service_actions(menu, "ScreenPipe", "screenpipe.service") + self._service_actions(menu, "ScreenPipe", "screenpipe.service", capture=True) self._service_actions(menu, "Collector", "chronicle-screenpipe.service") + settings_menu = menu.addMenu("Settings") + audio_menu = settings_menu.addMenu("Audio capture") + self.audio_group = QActionGroup(audio_menu) + self.audio_group.setExclusive(True) + self.audio_actions: dict[str, QAction] = {} + for mode, title in ( + ("off", "Off"), + ("system", "System audio"), + ("mic", "Microphone"), + ("both", "System audio + microphone"), + ): + action = audio_menu.addAction(title) + action.setCheckable(True) + action.triggered.connect( + lambda _checked=False, selected=mode: self.save_capture_settings(selected) + ) + self.audio_group.addAction(action) + self.audio_actions[mode] = action + self.screen_capture = settings_menu.addAction("Screen capture") + self.screen_capture.setCheckable(True) + self.screen_capture.triggered.connect(self.save_capture_settings) menu.addSeparator() self.sync_status = menu.addAction("Vault sync: starting…") self.sync_status.setEnabled(False) @@ -176,6 +197,7 @@ def __init__(self, state: SharedState, manager: VaultSyncManager) -> None: "Open Chronicle", lambda: QDesktopServices.openUrl(QUrl(manager.config.backend_url)), ) + menu.addAction("View Logs", self.view_logs) menu.addAction("Quit tray", QApplication.quit) self.setContextMenu(menu) self.setToolTip("Chronicle") @@ -184,21 +206,82 @@ def __init__(self, state: SharedState, manager: VaultSyncManager) -> None: self.timer.start(5000) self.refresh() - def _service_actions(self, menu: QMenu, label: str, unit: str) -> None: + def _service_actions( + self, menu: QMenu, label: str, unit: str, capture: bool = False + ) -> None: submenu = menu.addMenu(label) + actions = {} for title, verb in (("Start", "start"), ("Stop", "stop"), ("Restart", "restart")): action = QAction(title, submenu) action.triggered.connect(lambda _checked=False, v=verb, u=unit: self.service(v, u)) submenu.addAction(action) + actions[verb] = action + self.service_actions[unit] = actions + if capture: + self.pause_menu = submenu.addMenu("Pause for") + for title, minutes in ( + ("5 minutes", 5), + ("15 minutes", 15), + ("30 minutes", 30), + ("1 hour", 60), + ("2 hours", 120), + ("8 hours", 480), + ): + self.pause_menu.addAction( + title, lambda _checked=False, m=minutes: self.pause_capture(m) + ) def service(self, verb: str, unit: str) -> None: + if unit == "screenpipe.service" and verb in {"start", "restart"}: + self.pause_timer.stop() subprocess.run(["systemctl", "--user", verb, unit], check=False) QTimer.singleShot(500, self.refresh) + def pause_capture(self, minutes: int) -> None: + self.service("stop", "screenpipe.service") + self.pause_timer.start(minutes * 60 * 1000) + + def save_capture_settings(self, audio_mode: str | None = None) -> None: + try: + was_active = _unit_state("screenpipe.service") == "active" + _save_capture_settings( + audio_mode or _capture_settings()[0], self.screen_capture.isChecked() + ) + subprocess.run(["systemctl", "--user", "daemon-reload"], check=True) + if was_active: + self.service("restart", "screenpipe.service") + except (OSError, StopIteration, ValueError, subprocess.CalledProcessError) as error: + logger.exception("Could not save ScreenPipe settings") + self.showMessage("ScreenPipe settings", str(error), QSystemTrayIcon.Warning) + self.refresh_capture_settings() + + def refresh_capture_settings(self) -> None: + try: + audio_mode, screen_enabled = _capture_settings() + self.audio_actions[audio_mode].setChecked(True) + self.screen_capture.setChecked(screen_enabled) + for action in self.audio_actions.values(): + action.setEnabled(True) + self.screen_capture.setEnabled(True) + except (OSError, StopIteration, ValueError): + logger.exception("Could not read ScreenPipe settings") + for action in self.audio_actions.values(): + action.setEnabled(False) + self.screen_capture.setEnabled(False) + def refresh(self) -> None: self.manager.refresh_status() capture = _unit_state("screenpipe.service") collector = _unit_state("chronicle-screenpipe.service") + for unit, state in ( + ("screenpipe.service", capture), + ("chronicle-screenpipe.service", collector), + ): + active = state == "active" + self.service_actions[unit]["start"].setEnabled(not active) + self.service_actions[unit]["stop"].setEnabled(active) + self.pause_menu.setEnabled(capture == "active") + self.refresh_capture_settings() self.capture_status.setText(f"ScreenPipe: {capture}") self.collector_status.setText(f"Chronicle collector: {collector}") self.stats.setText(_screenpipe_stats()) @@ -223,9 +306,27 @@ def choose_folder(self) -> None: if chosen: self.manager.set_vault_dir(chosen) + def view_logs(self) -> None: + dialog = QDialog() + dialog.setWindowTitle("Chronicle Desktop — Logs") + dialog.resize(760, 440) + layout = QVBoxLayout(dialog) + lines = list(log_buffer.lines) + layout.addWidget(QLabel(f"Last {len(lines)} application log line(s)")) + viewer = QPlainTextEdit() + viewer.setReadOnly(True) + viewer.setLineWrapMode(QPlainTextEdit.NoWrap) + viewer.setPlainText("\n".join(lines) or "(no logs yet)") + viewer.verticalScrollBar().setValue(viewer.verticalScrollBar().maximum()) + layout.addWidget(viewer) + buttons = QDialogButtonBox(QDialogButtonBox.Close) + buttons.rejected.connect(dialog.reject) + layout.addWidget(buttons) + dialog.exec() + def main() -> None: - logging.basicConfig(level=logging.INFO) + configure_logging() app = QApplication(sys.argv) app.setQuitOnLastWindowClosed(False) if not QSystemTrayIcon.isSystemTrayAvailable(): diff --git a/extras/vault-sync/menu_vault.py b/extras/vault-sync/menu_vault.py index eb4b2786..9b9b92ce 100644 --- a/extras/vault-sync/menu_vault.py +++ b/extras/vault-sync/menu_vault.py @@ -7,41 +7,19 @@ import logging import subprocess -import threading -from collections import deque -from dataclasses import dataclass, field from pathlib import Path from typing import Optional from urllib.parse import quote -import httpx import rumps from dotenv import load_dotenv -from syncthing_manager import SyncthingManager -from vault_core import VaultSyncConfig, broker_pair, get_jwt_token, save_vault_dir +from desktop_core import SharedState, VaultSyncManager, configure_logging, log_buffer logger = logging.getLogger(__name__) load_dotenv() -class MemoryLogHandler(logging.Handler): - """Keep recent log lines in memory for display in the menu bar UI.""" - - def __init__(self, capacity: int = 500) -> None: - super().__init__() - self.lines: deque = deque(maxlen=capacity) - - def emit(self, record: logging.LogRecord) -> None: - try: - self.lines.append(self.format(record)) - except Exception: - self.handleError(record) - - -log_buffer = MemoryLogHandler() - - def _show_logs_dialog(title: str, lines) -> None: """Show log lines in a scrollable modal dialog.""" # Lazy import: macOS-only (AppKit/Foundation, not available cross-platform) @@ -104,151 +82,6 @@ def _choose_directory(default_path: str) -> Optional[str]: return None -# --- shared state ------------------------------------------------------------ - - -@dataclass -class SharedState: - """Thread-safe state shared between the rumps UI and the worker thread.""" - - _lock: threading.Lock = field(default_factory=threading.Lock, repr=False) - status: str = "idle" # idle | starting | pairing | syncing | error - error: Optional[str] = None - connected: bool = False - completion: Optional[float] = None - folder_error: Optional[str] = None # folder-level error from Syncthing - folder_id: Optional[str] = None - vault_dir: str = "" - - def snapshot(self) -> dict: - with self._lock: - return { - "status": self.status, - "error": self.error, - "connected": self.connected, - "completion": self.completion, - "folder_error": self.folder_error, - "folder_id": self.folder_id, - "vault_dir": self.vault_dir, - } - - def update(self, **kwargs) -> None: - with self._lock: - for k, v in kwargs.items(): - setattr(self, k, v) - - -# --- sync manager ------------------------------------------------------------ - - -class VaultSyncManager: - """Coordinates the local Syncthing and the backend pairing handshake.""" - - def __init__(self, state: SharedState) -> None: - self.state = state - self.config = VaultSyncConfig.from_env() - self.syncthing = SyncthingManager() - self.state.update(vault_dir=self.config.local_vault_dir) - self._lock = threading.Lock() - self._logged_errors: set[str] = set() - - def pair_async(self) -> None: - """Run the (blocking) pair flow on a background thread.""" - threading.Thread(target=self._pair, daemon=True).start() - - def _pair(self) -> None: - if not self._lock.acquire(blocking=False): - logger.info("Pair already in progress") - return - try: - cfg = self.config - if not cfg.auth_username or not cfg.auth_password: - self.state.update( - status="error", - error="Set AUTH_USERNAME/AUTH_PASSWORD in .env", - ) - return - - self.state.update(status="starting", error=None) - self.syncthing.start() - local_id = self.syncthing.device_id() - - self.state.update(status="pairing") - token = get_jwt_token(cfg.auth_username, cfg.auth_password, cfg.backend_url) - if not token: - self.state.update(status="error", error="Backend auth failed") - return - - info = broker_pair(cfg.backend_url, token, local_id, cfg.device_name) - - sync_address = info.get("sync_address") or "" - addresses = [sync_address] if sync_address else ["dynamic"] - self.syncthing.ensure_server_device( - info["server_device_id"], "Chronicle Server", addresses - ) - self.syncthing.ensure_folder( - folder_id=info["folder_id"], - path=cfg.local_vault_dir, - label=info.get("folder_label", "Chronicle Vault"), - server_device_id=info["server_device_id"], - self_device_id=local_id, - ) - self.state.update(status="syncing", folder_id=info["folder_id"], error=None) - logger.info( - "Paired. Folder %s -> %s", info["folder_id"], cfg.local_vault_dir - ) - except httpx.HTTPStatusError as e: - detail = e.response.text[:200] if e.response is not None else str(e) - self.state.update(status="error", error=f"Pair failed: {detail}") - logger.error("Pair failed: %s", detail) - except Exception as e: # noqa: BLE001 - surface any failure in the UI - self.state.update(status="error", error=str(e)) - logger.exception("Pair error") - finally: - self._lock.release() - - def set_vault_dir(self, path: str) -> None: - save_vault_dir(path) - self.config.local_vault_dir = path - self.state.update(vault_dir=path) - logger.info("Vault folder set to %s — re-pairing", path) - self.pair_async() - - def refresh_status(self) -> None: - if not self.syncthing.is_running(): - return - connected = self.syncthing.connection_count() > 0 - completion = None - folder_error = None - snap = self.state.snapshot() - if snap["folder_id"]: - fstatus = self.syncthing.folder_status(snap["folder_id"]) - completion = fstatus["completion"] - folder_error = fstatus["error"] - - # Pull Syncthing's own detailed errors into the log buffer so "View Logs" - # shows the real cause (e.g. a case collision), not just a count. Log each - # message once; re-log if it clears and recurs. - detailed = self.syncthing.collect_errors() - for msg in detailed: - if msg not in self._logged_errors: - logger.error("Syncthing: %s", msg) - self._logged_errors = set(detailed) - if detailed: - folder_error = ( - detailed[0] - if len(detailed) == 1 - else f"{detailed[0]} (+{len(detailed) - 1} more)" - ) - - self.state.update( - connected=connected, completion=completion, folder_error=folder_error - ) - - def shutdown(self) -> None: - self.syncthing.stop() - - # --- rumps menu bar app ------------------------------------------------------- @@ -354,11 +187,7 @@ def main() -> None: NSApplication.sharedApplication().setActivationPolicy_(1) # menu bar only - log_format = "%(asctime)s %(levelname)s %(name)s: %(message)s" - logging.basicConfig(format=log_format, level=logging.INFO) - log_buffer.setFormatter(logging.Formatter(log_format)) - logging.getLogger().addHandler(log_buffer) - logging.getLogger("httpx").setLevel(logging.WARNING) + configure_logging() state = SharedState() manager = VaultSyncManager(state) diff --git a/extras/vault-sync/service.py b/extras/vault-sync/service.py index 0d358465..7eacf96b 100644 --- a/extras/vault-sync/service.py +++ b/extras/vault-sync/service.py @@ -16,6 +16,19 @@ APP_BUNDLE = Path.home() / "Applications" / "Chronicle Vault Sync.app" PROJECT_DIR = Path(__file__).resolve().parent +ROOT_ENV_FILE = PROJECT_DIR.parents[1] / ".env" +LOCAL_ENV_FILE = PROJECT_DIR / ".env" + + +def _dotenv_environment() -> dict[str, str]: + """Load repo-wide settings, with vault-sync-specific values taking precedence.""" + env: dict[str, str] = {} + for env_file in (ROOT_ENV_FILE, LOCAL_ENV_FILE): + if env_file.exists(): + for key, value in dotenv_values(env_file).items(): + if value is not None: + env[key] = value + return env def _find_uv() -> str: @@ -80,12 +93,7 @@ def _remove_app_bundle() -> None: def _build_plist() -> dict: uv = _find_uv() - env = {} - env_file = PROJECT_DIR / ".env" - if env_file.exists(): - for key, value in dotenv_values(env_file).items(): - if value is not None: - env[key] = value + env = _dotenv_environment() # Ensure Homebrew bin is on PATH so the syncthing binary is found under launchd. env["PATH"] = "/opt/homebrew/bin:/usr/local/bin:" + os.environ.get("PATH", "") @@ -243,14 +251,19 @@ def _linux_install() -> None: "After=graphical-session.target network-online.target\n\n" "[Service]\nType=simple\n" f"WorkingDirectory={PROJECT_DIR}\n" + f"EnvironmentFile=-{ROOT_ENV_FILE}\n" + f"EnvironmentFile=-{LOCAL_ENV_FILE}\n" f"ExecStart={uv} run --project {PROJECT_DIR} python {PROJECT_DIR / 'main.py'} menu\n" "Restart=on-failure\nRestartSec=5\n\n" "[Install]\nWantedBy=graphical-session.target\n" ) subprocess.run(["systemctl", "--user", "daemon-reload"], check=True) subprocess.run( - ["systemctl", "--user", "enable", "--now", unit.name], check=True + ["systemctl", "--user", "enable", unit.name], check=True ) + # ``enable --now`` leaves an already-running unit untouched, so changes to + # EnvironmentFile would not take effect until the next login. + subprocess.run(["systemctl", "--user", "restart", unit.name], check=True) print(f"Installed and started {unit.name}") diff --git a/extras/vault-sync/start.sh b/extras/vault-sync/start.sh index 36969692..0c52b4c6 100755 --- a/extras/vault-sync/start.sh +++ b/extras/vault-sync/start.sh @@ -1,4 +1,7 @@ #!/bin/bash # Launch the Chronicle Vault Sync menu bar app (or pass a subcommand: install, logs, ...). -set -a && source .env 2>/dev/null; set +a +set -a +source ../../.env 2>/dev/null +source .env 2>/dev/null +set +a exec uv run python main.py "$@" diff --git a/extras/vault-sync/tests/test_menu_linux.py b/extras/vault-sync/tests/test_menu_linux.py new file mode 100644 index 00000000..1d01deaa --- /dev/null +++ b/extras/vault-sync/tests/test_menu_linux.py @@ -0,0 +1,72 @@ +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from menu_linux import _capture_settings, _save_capture_settings + + +def _unit(tmp_path: Path, flags: str = "") -> Path: + path = tmp_path / "screenpipe.service" + path.write_text( + "[Service]\nExecStart=/usr/bin/screenpipe record --api-auth true" + f" {flags}\nRestart=on-failure\n", + encoding="utf-8", + ) + return path + + +def test_capture_settings_default_to_audio_and_screen_enabled(tmp_path): + path = _unit(tmp_path) + + assert _capture_settings(path) == ("both", True) + + +def test_save_capture_settings_updates_only_capture_flags(tmp_path): + path = _unit(tmp_path, "--disable-audio") + + _save_capture_settings( + audio_mode="system", + screen_enabled=False, + path=path, + audio_devices=["Speakers (output)"], + ) + + assert _capture_settings(path) == ("system", False) + text = path.read_text(encoding="utf-8") + assert "--api-auth true" in text + assert "--disable-audio" not in text + assert text.count("--disable-vision") == 1 + + +def test_save_capture_settings_is_idempotent(tmp_path): + path = _unit(tmp_path, "--disable-audio --disable-vision") + + _save_capture_settings(audio_mode="off", screen_enabled=False, path=path) + + text = path.read_text(encoding="utf-8") + assert text.count("--disable-audio") == 1 + assert text.count("--disable-vision") == 1 + + +def test_save_capture_settings_supports_both_default_sources(tmp_path): + path = _unit(tmp_path, "--disable-audio") + + _save_capture_settings(audio_mode="both", screen_enabled=True, path=path) + + assert _capture_settings(path) == ("both", True) + assert "--use-system-default-audio true" in path.read_text(encoding="utf-8") + + +def test_save_capture_settings_supports_microphone_only(tmp_path): + path = _unit(tmp_path) + + _save_capture_settings( + audio_mode="mic", + screen_enabled=True, + path=path, + audio_devices=["Built-in Mic (input)", "Speakers (output)"], + ) + + assert _capture_settings(path) == ("mic", True) + assert "--audio-device 'Built-in Mic (input)'" in path.read_text(encoding="utf-8") diff --git a/extras/vault-sync/vault_core.py b/extras/vault-sync/vault_core.py index c1296f07..c3b93435 100644 --- a/extras/vault-sync/vault_core.py +++ b/extras/vault-sync/vault_core.py @@ -55,6 +55,14 @@ class VaultSyncConfig: local_vault_dir: str device_name: str + @staticmethod + def root_env_file() -> Path: + return _REPO_ROOT / ".env" + + @staticmethod + def local_env_file() -> Path: + return Path(__file__).resolve().parent / ".env" + @classmethod def from_env(cls) -> "VaultSyncConfig": # Lazy import: sys.path-dependent (repo-root discovery.py, inserted above) From ab4db63578cf2a659765cc092fd43d06c05a4432 Mon Sep 17 00:00:00 2001 From: Ankush <43288948+AnkushMalaker@users.noreply.github.com> Date: Wed, 22 Jul 2026 22:04:48 +0530 Subject: [PATCH 2/2] feat: separate screen audio capture sources --- .../services/device_audio_ingest.py | 16 ++++- .../tests/test_device_audio_ingest.py | 16 ++++- docs/screenpipe.md | 6 ++ extras/screenpipe-collector/README.md | 4 ++ .../tests/test_collector.py | 23 +++++++ extras/vault-sync/README.md | 3 +- extras/vault-sync/menu_linux.py | 60 ++++++++++++++++++- extras/vault-sync/tests/test_menu_linux.py | 18 +++++- 8 files changed, 138 insertions(+), 8 deletions(-) diff --git a/backends/advanced/src/advanced_omi_backend/services/device_audio_ingest.py b/backends/advanced/src/advanced_omi_backend/services/device_audio_ingest.py index fd512a23..2d3efbbc 100644 --- a/backends/advanced/src/advanced_omi_backend/services/device_audio_ingest.py +++ b/backends/advanced/src/advanced_omi_backend/services/device_audio_ingest.py @@ -48,7 +48,18 @@ def group_audio_sessions(items: list[DeviceInputItem]) -> list[list[DeviceInputI return sessions -async def _mix_session(items: list[DeviceInputItem], workspace: Path, output: Path) -> None: +def audio_stream_key(item: DeviceInputItem) -> tuple[str, str, str]: + """Keep microphone and system output in independent processing streams.""" + return ( + item.user_id, + item.source_id, + str(item.metadata.get("direction", "unknown")), + ) + + +async def _mix_session( + items: list[DeviceInputItem], workspace: Path, output: Path +) -> None: start = min(_as_utc(item.captured_at) for item in items) command = ["ffmpeg", "-hide_banner", "-loglevel", "error", "-y"] valid = [item for item in items if item.media_data] @@ -109,8 +120,7 @@ async def process_device_audio() -> dict[str, Any]: ) by_source: dict[tuple[str, str, str], list[DeviceInputItem]] = {} for item in pending: - direction = str(item.metadata.get("direction", "unknown")) - by_source.setdefault((item.user_id, item.source_id, direction), []).append(item) + by_source.setdefault(audio_stream_key(item), []).append(item) processed = 0 for (user_id, source_id, direction), source_items in by_source.items(): try: diff --git a/backends/advanced/tests/test_device_audio_ingest.py b/backends/advanced/tests/test_device_audio_ingest.py index 24e4e597..3222b502 100644 --- a/backends/advanced/tests/test_device_audio_ingest.py +++ b/backends/advanced/tests/test_device_audio_ingest.py @@ -1,7 +1,10 @@ from datetime import datetime, timedelta, timezone from types import SimpleNamespace -from advanced_omi_backend.services.device_audio_ingest import group_audio_sessions +from advanced_omi_backend.services.device_audio_ingest import ( + audio_stream_key, + group_audio_sessions, +) def item(identifier: str, start: datetime, duration: float = 30): @@ -12,6 +15,17 @@ def item(identifier: str, start: datetime, duration: float = 30): ) +def test_audio_stream_key_separates_microphone_from_system_output(): + microphone = SimpleNamespace( + user_id="user", source_id="rainbow", metadata={"direction": "input"} + ) + system = SimpleNamespace( + user_id="user", source_id="rainbow", metadata={"direction": "output"} + ) + + assert audio_stream_key(microphone) != audio_stream_key(system) + + def test_audio_chunks_group_across_input_and_output_devices(): start = datetime(2026, 7, 22, tzinfo=timezone.utc) sessions = group_audio_sessions( diff --git a/docs/screenpipe.md b/docs/screenpipe.md index fe9893eb..f32222e8 100644 --- a/docs/screenpipe.md +++ b/docs/screenpipe.md @@ -33,6 +33,12 @@ an explicit `--audio-device "... (output)"` or `--audio-device "... (input)"` wi `--use-system-default-audio false`. The collector preserves the source direction when it sends audio to Chronicle. +Local capture and Chronicle forwarding are independent policies. The collector's +`forward_audio` setting accepts `none`, `output`, `input`, or `both`; excluded chunks +remain in ScreenPipe's local store and are checkpointed without upload. Chronicle also +processes input and output as separate sessions so microphone and system audio are not +mixed together before transcription. + The setup wizard's capture-node option delegates to `extras/screenpipe-collector/init.py`. Pairing and standalone commands are documented in the [companion README](../extras/screenpipe-collector/README.md). diff --git a/extras/screenpipe-collector/README.md b/extras/screenpipe-collector/README.md index f0d61bb4..2c6b2666 100644 --- a/extras/screenpipe-collector/README.md +++ b/extras/screenpipe-collector/README.md @@ -41,6 +41,10 @@ Chronicle. See the full [capture-node architecture](../../docs/screenpipe.md). exact platform device names with `screenpipe audio list --output json`. Use `--disable-audio` only for screen-only capture. + Pair with `--forward-audio none|output|input|both` to independently control which + locally recorded sources are uploaded. The guided setup asks for both the local + capture mode and forwarding mode. + 4. Run the companion: ```bash diff --git a/extras/screenpipe-collector/tests/test_collector.py b/extras/screenpipe-collector/tests/test_collector.py index cb6bbb00..07f590b9 100644 --- a/extras/screenpipe-collector/tests/test_collector.py +++ b/extras/screenpipe-collector/tests/test_collector.py @@ -4,6 +4,7 @@ from chronicle_screenpipe.collector import ( Checkpoints, Collector, + Config, activity_is_salient, audio_duration, build_activity_sessions, @@ -140,3 +141,25 @@ def test_collect_audio_accepts_screenpipe_startup_schema(): db.execute("CREATE TABLE audio_chunks (placeholder TEXT)") collector = object.__new__(Collector) assert collector.collect_audio(db) == 0 + + +def test_collect_audio_checkpoints_sources_excluded_from_forwarding(tmp_path: Path): + db = sqlite3.connect(":memory:") + db.row_factory = sqlite3.Row + db.execute("CREATE TABLE audio_chunks (id INTEGER, file_path TEXT, timestamp TEXT)") + db.execute( + "INSERT INTO audio_chunks VALUES (1, ?, ?)", + (str(tmp_path / "Microphone (input)_1.wav"), "2026-07-22T10:00:00Z"), + ) + collector = object.__new__(Collector) + collector.config = Config( + backend_url="http://chronicle", + source_id="rainbow", + token="token", + screenpipe_dir=tmp_path, + forward_audio="output", + ) + collector.checkpoints = Checkpoints(tmp_path / "state.json") + + assert collector.collect_audio(db) == 0 + assert collector.checkpoints.get("audio") == 1 diff --git a/extras/vault-sync/README.md b/extras/vault-sync/README.md index d23d834c..3dfaba27 100644 --- a/extras/vault-sync/README.md +++ b/extras/vault-sync/README.md @@ -36,7 +36,8 @@ vault-sync orchestration; only the native menu UI is platform-specific. Both pro provides start, stop, and restart controls for `screenpipe.service` and `chronicle-screenpipe.service`. Its top-level Settings menu can enable or disable audio and screen capture, while the ScreenPipe menu can pause capture for 5, 15, or 30 -minutes, or 1, 2, or 8 hours. +minutes, or 1, 2, or 8 hours. Audio capture and audio forwarding are separate menus: +each can independently select off/none, system audio, microphone, or both. You also need the Chronicle **server** side running with vault sync enabled — see [Server setup](#server-setup-once) below. diff --git a/extras/vault-sync/menu_linux.py b/extras/vault-sync/menu_linux.py index 73f07fdd..89cf4acb 100644 --- a/extras/vault-sync/menu_linux.py +++ b/extras/vault-sync/menu_linux.py @@ -29,6 +29,7 @@ logger = logging.getLogger(__name__) SCREENPIPE_DB = Path.home() / ".screenpipe/db.sqlite" SCREENPIPE_UNIT = Path.home() / ".config/systemd/user/screenpipe.service" +COLLECTOR_CONFIG = Path.home() / ".config/chronicle-screenpipe/config.json" load_dotenv(Path(__file__).resolve().parent / ".env") @@ -115,6 +116,21 @@ def _audio_devices() -> list[str]: return [entry["name"] for entry in json.loads(result.stdout)["data"]] +def _forward_audio_setting(path: Path = COLLECTOR_CONFIG) -> str: + value = json.loads(path.read_text(encoding="utf-8"))["forward_audio"] + if value not in {"none", "output", "input", "both"}: + raise ValueError(f"unsupported forwarding mode: {value}") + return value + + +def _save_forward_audio_setting(mode: str, path: Path = COLLECTOR_CONFIG) -> None: + if mode not in {"none", "output", "input", "both"}: + raise ValueError(f"unsupported forwarding mode: {mode}") + config = json.loads(path.read_text(encoding="utf-8")) + config["forward_audio"] = mode + path.write_text(json.dumps(config, indent=2) + "\n", encoding="utf-8") + + def _save_capture_settings( audio_mode: str, screen_enabled: bool, @@ -135,8 +151,12 @@ def _save_capture_settings( elif audio_mode == "both": args.extend(["--use-system-default-audio", "true"]) else: - suffix = f"({ 'output' if audio_mode == 'system' else 'input' })" - matching = [name for name in (audio_devices or _audio_devices()) if name.lower().endswith(suffix)] + suffix = "(output)" if audio_mode == "system" else "(input)" + matching = [ + name + for name in (audio_devices or _audio_devices()) + if name.lower().endswith(suffix) + ] if not matching: raise ValueError(f"no {audio_mode} audio device is available") args.extend(["--use-system-default-audio", "false", "--audio-device", matching[0]]) @@ -186,6 +206,23 @@ def __init__(self, state: SharedState, manager: VaultSyncManager) -> None: self.screen_capture = settings_menu.addAction("Screen capture") self.screen_capture.setCheckable(True) self.screen_capture.triggered.connect(self.save_capture_settings) + forwarding_menu = settings_menu.addMenu("Send audio to Chronicle") + self.forwarding_group = QActionGroup(forwarding_menu) + self.forwarding_group.setExclusive(True) + self.forwarding_actions: dict[str, QAction] = {} + for mode, title in ( + ("none", "None"), + ("output", "System audio"), + ("input", "Microphone"), + ("both", "System audio + microphone"), + ): + action = forwarding_menu.addAction(title) + action.setCheckable(True) + action.triggered.connect( + lambda _checked=False, selected=mode: self.save_forwarding(selected) + ) + self.forwarding_group.addAction(action) + self.forwarding_actions[mode] = action menu.addSeparator() self.sync_status = menu.addAction("Vault sync: starting…") self.sync_status.setEnabled(False) @@ -268,6 +305,25 @@ def refresh_capture_settings(self) -> None: for action in self.audio_actions.values(): action.setEnabled(False) self.screen_capture.setEnabled(False) + try: + forwarding = _forward_audio_setting() + self.forwarding_actions[forwarding].setChecked(True) + for action in self.forwarding_actions.values(): + action.setEnabled(True) + except (OSError, KeyError, ValueError, json.JSONDecodeError): + logger.exception("Could not read collector audio forwarding settings") + for action in self.forwarding_actions.values(): + action.setEnabled(False) + + def save_forwarding(self, mode: str) -> None: + try: + _save_forward_audio_setting(mode) + if _unit_state("chronicle-screenpipe.service") == "active": + self.service("restart", "chronicle-screenpipe.service") + except (OSError, ValueError, json.JSONDecodeError) as error: + logger.exception("Could not save collector audio forwarding settings") + self.showMessage("Chronicle collector", str(error), QSystemTrayIcon.Warning) + self.refresh_capture_settings() def refresh(self) -> None: self.manager.refresh_status() diff --git a/extras/vault-sync/tests/test_menu_linux.py b/extras/vault-sync/tests/test_menu_linux.py index 1d01deaa..4b03cfc8 100644 --- a/extras/vault-sync/tests/test_menu_linux.py +++ b/extras/vault-sync/tests/test_menu_linux.py @@ -1,9 +1,15 @@ +import json import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) -from menu_linux import _capture_settings, _save_capture_settings +from menu_linux import ( + _capture_settings, + _forward_audio_setting, + _save_capture_settings, + _save_forward_audio_setting, +) def _unit(tmp_path: Path, flags: str = "") -> Path: @@ -70,3 +76,13 @@ def test_save_capture_settings_supports_microphone_only(tmp_path): assert _capture_settings(path) == ("mic", True) assert "--audio-device 'Built-in Mic (input)'" in path.read_text(encoding="utf-8") + + +def test_audio_forwarding_setting_is_independent_from_capture_unit(tmp_path): + path = tmp_path / "config.json" + path.write_text('{"source_id": "rainbow", "forward_audio": "both"}') + + _save_forward_audio_setting("output", path) + + assert _forward_audio_setting(path) == "output" + assert json.loads(path.read_text())["source_id"] == "rainbow"