From f121e58616b4b717723aaf4d3feccf0680d830eb Mon Sep 17 00:00:00 2001 From: Ankush <43288948+AnkushMalaker@users.noreply.github.com> Date: Wed, 22 Jul 2026 06:39:50 +0530 Subject: [PATCH 01/16] feat: guide ScreenPipe capture-node setup --- extras/screenpipe-collector/init.py | 114 ++++++++++++++++++ .../screenpipe-collector/tests/test_init.py | 19 +++ tests/unit/test_wizard_defaults.py | 9 ++ wizard.py | 55 +++++++-- 4 files changed, 188 insertions(+), 9 deletions(-) create mode 100644 extras/screenpipe-collector/init.py create mode 100644 extras/screenpipe-collector/tests/test_init.py diff --git a/extras/screenpipe-collector/init.py b/extras/screenpipe-collector/init.py new file mode 100644 index 000000000..8328b4e3a --- /dev/null +++ b/extras/screenpipe-collector/init.py @@ -0,0 +1,114 @@ +#!/usr/bin/env python3 +"""Guided host-native setup for a Chronicle ScreenPipe capture node.""" + +from __future__ import annotations + +import argparse +import os +import secrets +import shutil +import subprocess +from pathlib import Path + +from rich.console import Console +from rich.prompt import Confirm, Prompt + + +console = Console() +PROJECT = Path(__file__).resolve().parent +SYSTEMD_USER_DIR = Path.home() / ".config/systemd/user" + + +def screenpipe_command() -> str | None: + return shutil.which("screenpipe") + + +def write_screenpipe_unit(binary: str, api_key: str) -> 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", + "--disable-clipboard-capture", + "--prioritize-input-latency", + "--pause-on-drm-content", + "--disable-meeting-detector", + "--disable-telemetry", + "--video-quality", "balanced", + "--retention-days", "90", + "--retention-mode", "media", + "--api-auth", "true", + ] + 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" + "[Install]\nWantedBy=default.target\n", + encoding="utf-8", + ) + path.chmod(0o600) + return path + + +def run(*args: str) -> None: + subprocess.run(list(args), check=True) + + +def main() -> None: + parser = argparse.ArgumentParser(description="Configure a Chronicle capture node") + parser.add_argument("--backend") + args = parser.parse_args() + + console.print("\nπŸ–₯️ [bold cyan]Chronicle capture node[/bold cyan]") + binary = screenpipe_command() + if not binary: + console.print( + "[red]βœ— ScreenPipe is not installed.[/red] Install it independently, verify " + "[cyan]screenpipe record --help[/cyan], then rerun this wizard." + ) + raise SystemExit(1) + 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") + console.print( + "Open Chronicle β†’ Timeline β†’ Sources and create a pairing code. " + "The code expires after 10 minutes." + ) + code = Prompt.ask("Pairing code").strip() + if not code: + raise SystemExit("pairing code is required") + + api_key = secrets.token_urlsafe(32) + run( + "uv", "run", "--project", str(PROJECT), "chronicle-screenpipe", "pair", + "--backend", backend, + "--code", code, + "--screenpipe-dir", str(Path.home() / ".screenpipe"), + "--screenpipe-url", "http://127.0.0.1:3030", + "--screenpipe-token", api_key, + ) + write_screenpipe_unit(binary, api_key) + run("uv", "run", "--project", str(PROJECT), "chronicle-screenpipe", "install-service") + run("systemctl", "--user", "daemon-reload") + run("systemctl", "--user", "enable", "--now", "screenpipe.service") + run("systemctl", "--user", "restart", "chronicle-screenpipe.service") + + if Confirm.ask("Check service status now?", default=True): + run( + "systemctl", "--user", "--no-pager", "--full", "status", + "screenpipe.service", "chronicle-screenpipe.service", + ) + console.print( + "\n[green]βœ… Capture node connected.[/green] Activity should appear in " + "Chronicle Timeline after ScreenPipe records a stable app/window span." + ) + + +if __name__ == "__main__": + main() diff --git a/extras/screenpipe-collector/tests/test_init.py b/extras/screenpipe-collector/tests/test_init.py new file mode 100644 index 000000000..8ecafb45c --- /dev/null +++ b/extras/screenpipe-collector/tests/test_init.py @@ -0,0 +1,19 @@ +import importlib.util +from pathlib import Path + + +INIT_PATH = Path(__file__).parents[1] / "init.py" +SPEC = importlib.util.spec_from_file_location("screenpipe_collector_init", INIT_PATH) +MODULE = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(MODULE) + + +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") + 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 diff --git a/tests/unit/test_wizard_defaults.py b/tests/unit/test_wizard_defaults.py index fef0c0047..e349a3bde 100644 --- a/tests/unit/test_wizard_defaults.py +++ b/tests/unit/test_wizard_defaults.py @@ -41,6 +41,7 @@ def _load_wizard(): get_existing_stt_provider = _wizard.get_existing_stt_provider get_existing_stream_provider = _wizard.get_existing_stream_provider select_llm_provider = _wizard.select_llm_provider +select_setup_type = _wizard.select_setup_type # --------------------------------------------------------------------------- @@ -150,3 +151,11 @@ def test_select_llm_provider_none_config(): """Treats None config_yml as empty dict (defaults to openai).""" result = _select_llm_with_eof(None) assert result == "openai" + + +@pytest.mark.parametrize( + "choice, expected", [("1", "main"), ("2", "join"), ("3", "capture")] +) +def test_select_setup_type(choice, expected): + with patch.object(_wizard.Prompt, "ask", return_value=choice): + assert select_setup_type() == expected diff --git a/wizard.py b/wizard.py index 60a28ea32..406519ea1 100755 --- a/wizard.py +++ b/wizard.py @@ -1606,9 +1606,43 @@ def select_setup_type(): console.print( " advertises it to an existing backend on your Tailnet (no backend here)" ) + console.print( + " 3) Capture node β€” run ScreenPipe + the Chronicle companion (no containers)" + ) console.print() choice = Prompt.ask("Enter choice", default="1") - return "join" if choice.strip() == "2" else "main" + if choice.strip() == "2": + return "join" + if choice.strip() == "3": + return "capture" + return "main" + + +def setup_capture_node(): + """Delegate ScreenPipe capture-node setup to its separate companion.""" + init_script = Path("extras/screenpipe-collector/init.py") + if not init_script.exists(): + console.print(f"[red]βœ— Capture-node setup is missing: {init_script}[/red]") + return False + + backend_url = discovery.discover_service(discovery.CHRONICLE_BACKEND) + cmd = [ + "uv", + "run", + "--with-requirements", + "../../setup-requirements.txt", + "python", + "init.py", + ] + if backend_url: + console.print(f"[green]βœ…[/green] Found Chronicle at [cyan]{backend_url}[/cyan]") + cmd.extend(["--backend", backend_url]) + try: + subprocess.run(cmd, cwd=init_script.parent, check=True) + return True + except (OSError, subprocess.CalledProcessError) as exc: + console.print(f"[red]βœ— Capture-node setup failed: {exc}[/red]") + return False def join_cluster(): @@ -1848,11 +1882,6 @@ def main(): config_mgr = ConfigManager() config_mgr.ensure_config_yml() - # Container-engine prereq β€” everything below runs in containers, so bail early - # (with a clear reason) rather than failing deep in a build/start step. - if not check_container_engine(): - return - # Setup git hooks first setup_git_hooks() @@ -1862,9 +1891,17 @@ def main(): # Read existing config.yml once β€” used as defaults for ALL wizard questions below config_yml = config_mgr.get_full_config() - # Fork: a service-only node joining an existing cluster takes a separate, much - # shorter path (no backend / LLM / memory setup here) and returns. - if select_setup_type() == "join": + # Capture nodes are host-native and intentionally do not require containers. + setup_type = select_setup_type() + if setup_type == "capture": + setup_capture_node() + return + + # All hub and compute-node services below run in containers. + if not check_container_engine(): + return + + if setup_type == "join": join_cluster() return From 977bdfbb393fd3913926cd606d3a90906806f460 Mon Sep 17 00:00:00 2001 From: Ankush <43288948+AnkushMalaker@users.noreply.github.com> Date: Wed, 22 Jul 2026 06:46:30 +0530 Subject: [PATCH 02/16] fix: refresh capture timeline and tolerate recorder startup --- backends/advanced/webui/src/pages/Timeline.tsx | 6 +++++- .../chronicle_screenpipe/collector.py | 5 +++++ extras/screenpipe-collector/tests/test_collector.py | 10 +++++++++- 3 files changed, 19 insertions(+), 2 deletions(-) diff --git a/backends/advanced/webui/src/pages/Timeline.tsx b/backends/advanced/webui/src/pages/Timeline.tsx index c6131837b..730d770e2 100644 --- a/backends/advanced/webui/src/pages/Timeline.tsx +++ b/backends/advanced/webui/src/pages/Timeline.tsx @@ -19,7 +19,11 @@ function ItemIcon({ item }: { item: DeviceInputItem }) { export default function Timeline() { const [day, setDay] = useState(() => new Date().toISOString().slice(0, 10)) const [start, end] = useMemo(() => dayBounds(day), [day]) - const timeline = useQuery({ queryKey: ['device-timeline', day], queryFn: async () => (await deviceInputApi.getTimeline(start, end)).data.items }) + const timeline = useQuery({ + queryKey: ['device-timeline', day], + queryFn: async () => (await deviceInputApi.getTimeline(start, end)).data.items, + refetchInterval: 10_000, + }) const sources = useQuery({ queryKey: ['device-input-sources'], queryFn: async () => (await deviceInputApi.getSources()).data.sources, refetchInterval: 30_000 }) const pairing = useMutation({ mutationFn: async () => (await deviceInputApi.createPairingCode()).data }) diff --git a/extras/screenpipe-collector/chronicle_screenpipe/collector.py b/extras/screenpipe-collector/chronicle_screenpipe/collector.py index 2e8119424..8dd119741 100644 --- a/extras/screenpipe-collector/chronicle_screenpipe/collector.py +++ b/extras/screenpipe-collector/chronicle_screenpipe/collector.py @@ -190,6 +190,11 @@ def collect_audio(self, connection: sqlite3.Connection) -> int: columns = table_columns(connection, "audio_chunks") required = {"id", "file_path", "timestamp"} if not required <= columns: + # ScreenPipe may expose a migration placeholder briefly before the + # first audio segment initializes the final schema. + count = connection.execute("SELECT COUNT(*) FROM audio_chunks").fetchone()[0] + if count == 0: + return 0 raise RuntimeError(f"unsupported ScreenPipe audio_chunks schema; missing {sorted(required - columns)}") cursor = self.checkpoints.get("audio") rows = connection.execute( diff --git a/extras/screenpipe-collector/tests/test_collector.py b/extras/screenpipe-collector/tests/test_collector.py index a6be91dd1..2e313c886 100644 --- a/extras/screenpipe-collector/tests/test_collector.py +++ b/extras/screenpipe-collector/tests/test_collector.py @@ -1,7 +1,7 @@ import sqlite3 from pathlib import Path -from chronicle_screenpipe.collector import Checkpoints, audio_duration, build_activity_sessions, fold_activity_rows, infer_audio_direction +from chronicle_screenpipe.collector import Checkpoints, Collector, audio_duration, build_activity_sessions, fold_activity_rows, infer_audio_direction def test_activity_sessions_collapse_same_window(): @@ -54,3 +54,11 @@ def test_wav_duration_is_read_from_media(tmp_path: Path): audio.setframerate(16000) audio.writeframes(b"\0\0" * 8000) assert audio_duration(target) == 0.5 + + +def test_collect_audio_accepts_screenpipe_startup_schema(): + db = sqlite3.connect(":memory:") + db.row_factory = sqlite3.Row + db.execute("CREATE TABLE audio_chunks (placeholder TEXT)") + collector = object.__new__(Collector) + assert collector.collect_audio(db) == 0 From 9e3bc74c09ae51a88793ba5a64ae385a1f636e4d Mon Sep 17 00:00:00 2001 From: Ankush <43288948+AnkushMalaker@users.noreply.github.com> Date: Wed, 22 Jul 2026 06:51:30 +0530 Subject: [PATCH 03/16] fix: group Timeline audio chunks into sessions --- .../advanced/webui/src/pages/Timeline.tsx | 57 ++++++++++++++++++- 1 file changed, 54 insertions(+), 3 deletions(-) diff --git a/backends/advanced/webui/src/pages/Timeline.tsx b/backends/advanced/webui/src/pages/Timeline.tsx index 730d770e2..6ed5aff6a 100644 --- a/backends/advanced/webui/src/pages/Timeline.tsx +++ b/backends/advanced/webui/src/pages/Timeline.tsx @@ -16,6 +16,55 @@ function ItemIcon({ item }: { item: DeviceInputItem }) { return } +const AUDIO_SESSION_GAP_MS = 90_000 +const AUDIO_SESSION_MAX_MS = 30 * 60_000 + +export function groupTimelineAudio(items: DeviceInputItem[]): DeviceInputItem[] { + const visible = items.filter(item => item.kind !== 'audio') + const bySource = new Map() + for (const item of items) { + if (item.kind !== 'audio') continue + bySource.set(item.source_id, [...(bySource.get(item.source_id) || []), item]) + } + + for (const [sourceId, sourceItems] of bySource) { + const ordered = sourceItems.sort((a, b) => Date.parse(a.captured_at) - Date.parse(b.captured_at)) + let session: DeviceInputItem | null = null + for (const item of ordered) { + const itemStart = Date.parse(item.captured_at) + const itemEnd = Date.parse(item.ended_at || item.captured_at) + const sessionStart = session ? Date.parse(session.captured_at) : 0 + const sessionEnd = session ? Date.parse(session.ended_at || session.captured_at) : 0 + if (!session || itemStart - sessionEnd > AUDIO_SESSION_GAP_MS || itemStart - sessionStart >= AUDIO_SESSION_MAX_MS) { + session = { + ...item, + id: `audio-session:${sourceId}:${item.id}`, + metadata: { + chunk_count: 1, + directions: item.metadata.direction ? [item.metadata.direction] : [], + }, + } + visible.push(session) + continue + } + session.ended_at = new Date(Math.max(sessionEnd, itemEnd)).toISOString() + session.metadata.chunk_count += 1 + const direction = item.metadata.direction + if (direction && !session.metadata.directions.includes(direction)) { + session.metadata.directions.push(direction) + } + } + } + return visible.sort((a, b) => Date.parse(a.captured_at) - Date.parse(b.captured_at)) +} + +function formatDuration(item: DeviceInputItem) { + const seconds = Math.max(0, Math.round((Date.parse(item.ended_at || item.captured_at) - Date.parse(item.captured_at)) / 1000)) + if (seconds < 60) return `${seconds}s` + const minutes = Math.floor(seconds / 60) + return `${minutes}m ${seconds % 60}s` +} + export default function Timeline() { const [day, setDay] = useState(() => new Date().toISOString().slice(0, 10)) const [start, end] = useMemo(() => dayBounds(day), [day]) @@ -26,6 +75,7 @@ export default function Timeline() { }) const sources = useQuery({ queryKey: ['device-input-sources'], queryFn: async () => (await deviceInputApi.getSources()).data.sources, refetchInterval: 30_000 }) const pairing = useMutation({ mutationFn: async () => (await deviceInputApi.createPairingCode()).data }) + const visibleItems = useMemo(() => groupTimelineAudio(timeline.data || []), [timeline.data]) return (
@@ -62,15 +112,16 @@ export default function Timeline() {

Activity

{timeline.isFetching && }
- {(timeline.data || []).map(item => ( + {visibleItems.map(item => (
{new Date(item.captured_at).toLocaleTimeString()} {item.ended_at && `– ${new Date(item.ended_at).toLocaleTimeString()}`}
-

{item.metadata.app_name || item.metadata.window_name || item.metadata.text || item.kind}

+

{item.kind === 'audio' ? 'Audio capture' : item.metadata.app_name || item.metadata.window_name || item.metadata.text || item.kind}

+ {item.kind === 'audio' &&

{formatDuration(item)} Β· {item.metadata.chunk_count} chunks{item.metadata.directions?.length ? ` Β· ${item.metadata.directions.join(' + ')}` : ''}

} {item.metadata.window_name && item.metadata.window_name !== item.metadata.app_name &&

{item.metadata.window_name}

}
))} - {!timeline.isLoading && !timeline.data?.length &&
Nothing captured for this day.
} + {!timeline.isLoading && !visibleItems.length &&
Nothing captured for this day.
}
From 09cff63aeb5bac9f58497ea62af71cc0267c00f2 Mon Sep 17 00:00:00 2001 From: Ankush Malaker <43288948+AnkushMalaker@users.noreply.github.com> Date: Wed, 22 Jul 2026 01:40:01 +0000 Subject: [PATCH 04/16] =?UTF-8?q?fix:=20unblock=20CI=20=E2=80=94=20test=20?= =?UTF-8?q?env=20defaults=20+=20workers-test=20startup=20deadlock?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - tests/conftest.py: provide AUTH_SECRET_KEY/ADMIN_PASSWORD defaults so collection works without a local .env (CI has none; auth.py raises at import since the readiness/fleet work landed). - docker-compose-test.yml: drop workers-test dependency on backend health. /readiness now requires the worker-fleet heartbeat, so gating workers on backend health deadlocked startup (backend unhealthy -> workers never start). Matches prod compose: workers need mongo+redis. --- backends/advanced/docker-compose-test.yml | 7 +++++-- backends/advanced/tests/conftest.py | 18 ++++++++++++++++++ 2 files changed, 23 insertions(+), 2 deletions(-) create mode 100644 backends/advanced/tests/conftest.py diff --git a/backends/advanced/docker-compose-test.yml b/backends/advanced/docker-compose-test.yml index 6bf75b87b..3668ab7fe 100644 --- a/backends/advanced/docker-compose-test.yml +++ b/backends/advanced/docker-compose-test.yml @@ -384,8 +384,11 @@ services: - LANGFUSE_PUBLIC_KEY=pk-lf-test-public-key - LANGFUSE_SECRET_KEY=sk-lf-test-secret-key depends_on: - chronicle-backend-test: - condition: service_healthy + # NOTE: no dependency on chronicle-backend-test. The backend's /readiness + # healthcheck requires the worker fleet heartbeat in Redis, so gating the + # workers on backend health deadlocks startup (backend waits for workers, + # workers wait for backend). Matches prod compose: workers only need + # mongo + redis. mongo-test: condition: service_healthy redis-test: diff --git a/backends/advanced/tests/conftest.py b/backends/advanced/tests/conftest.py new file mode 100644 index 000000000..6abb1cc0a --- /dev/null +++ b/backends/advanced/tests/conftest.py @@ -0,0 +1,18 @@ +"""Shared pytest fixtures and test environment defaults. + +Several backend modules (notably ``advanced_omi_backend.auth``) validate that +required secrets are configured at *import* time. In CI there is no ``.env`` +file, so importing the app during test collection would raise +``ValueError: is not set``. We provide deterministic test defaults here so +collection succeeds without depending on a developer's local ``.env``. + +``setdefault`` is used so a real environment (CI secrets or a local ``.env`` +already exported) always wins over these placeholders. +""" + +import os + +# Import-time required secrets (see advanced_omi_backend.auth). +os.environ.setdefault("AUTH_SECRET_KEY", "test-auth-secret-key") +os.environ.setdefault("ADMIN_PASSWORD", "test-admin-password") +os.environ.setdefault("ADMIN_EMAIL", "admin@example.com") From 107626ce1f189593f024f7a877091aa75288faf3 Mon Sep 17 00:00:00 2001 From: Ankush Malaker <43288948+AnkushMalaker@users.noreply.github.com> Date: Wed, 22 Jul 2026 01:41:16 +0000 Subject: [PATCH 05/16] fix: land SessionStore conversation-assignment API tested by #336 test_session_store.py was committed in #336 but the SessionStore implementation it exercises (set/get/clear/expire current conversation, conversation_create_lock) stayed in a local stash. Bring the implementation in so the committed tests pass. --- .../services/audio_stream/session_store.py | 188 ++++++++++++++++++ 1 file changed, 188 insertions(+) diff --git a/backends/advanced/src/advanced_omi_backend/services/audio_stream/session_store.py b/backends/advanced/src/advanced_omi_backend/services/audio_stream/session_store.py index ab7524058..31416c815 100644 --- a/backends/advanced/src/advanced_omi_backend/services/audio_stream/session_store.py +++ b/backends/advanced/src/advanced_omi_backend/services/audio_stream/session_store.py @@ -18,14 +18,24 @@ ``decode_responses`` setting. """ +import asyncio import json import logging import time +import uuid +from contextlib import asynccontextmanager from dataclasses import dataclass, field from datetime import datetime, timezone from enum import Enum from typing import AsyncIterator, Literal, Optional +from redis.exceptions import WatchError + +from advanced_omi_backend.redis_keys import ( + conversation_create_lock as conversation_create_lock_key, +) +from advanced_omi_backend.redis_keys import conversation_current + logger = logging.getLogger(__name__) # Key templates @@ -391,6 +401,184 @@ async def persist_session(self, session_id: str) -> None: async def delete(self, session_id: str) -> None: await self._redis.delete(self._key(session_id)) + # ------------------------------------------------ conversation assignment + async def set_current_conversation( + self, + session_id: str, + conversation_id: str, + *, + ttl: Optional[int] = 86400, + ) -> None: + """Assign the conversation that receives this session's persisted audio. + + ``ttl=None`` is reserved for an ``always_persist`` placeholder, whose + lifetime is tied to the session rather than a rotation timeout. + """ + key = conversation_current(session_id) + if ttl is None: + await self._redis.set(key, conversation_id) + else: + await self._redis.set(key, conversation_id, ex=ttl) + + async def get_current_conversation_id(self, session_id: str) -> Optional[str]: + """Return the decoded current-conversation assignment, if one exists.""" + return _to_str(await self._redis.get(conversation_current(session_id))) + + async def assign_current_conversation_if_active( + self, + session_id: str, + conversation_id: str, + *, + ttl: Optional[int] = 86400, + ) -> bool: + """Assign only when the session is active and has no current owner. + + The session status check and pointer creation share one Redis transaction, + closing the finalization race between a separate ``get_status`` and ``set``. + """ + session_key = self._key(session_id) + current_key = conversation_current(session_id) + + while True: + async with self._redis.pipeline(transaction=True) as pipe: + try: + await pipe.watch(session_key, current_key) + status = _to_str(await pipe.hget(session_key, "status")) + current_id = _to_str(await pipe.get(current_key)) + if status != SessionStatus.ACTIVE.value or current_id is not None: + await pipe.unwatch() + return False + + pipe.multi() + if ttl is None: + pipe.set(current_key, conversation_id) + else: + pipe.set(current_key, conversation_id, ex=ttl) + await pipe.execute() + return True + except WatchError: + continue + + async def replace_current_conversation_if_active( + self, + session_id: str, + expected_id: str, + replacement_id: str, + *, + ttl: Optional[int] = 86400, + ) -> bool: + """Atomically rotate an active session from one owner to its successor. + + There is no unassigned interval: an audio XADD watching the same pointer + observes either ``expected_id`` or ``replacement_id``. Finalization and a + competing rotation both make the compare-and-swap fail without mutation. + """ + session_key = self._key(session_id) + current_key = conversation_current(session_id) + + while True: + async with self._redis.pipeline(transaction=True) as pipe: + try: + await pipe.watch(session_key, current_key) + status = _to_str(await pipe.hget(session_key, "status")) + current_id = _to_str(await pipe.get(current_key)) + if ( + status != SessionStatus.ACTIVE.value + or current_id != expected_id + ): + await pipe.unwatch() + return False + + pipe.multi() + if ttl is None: + pipe.set(current_key, replacement_id) + else: + pipe.set(current_key, replacement_id, ex=ttl) + await pipe.execute() + return True + except WatchError: + continue + + async def clear_current_conversation( + self, + session_id: str, + *, + expected_id: Optional[str] = None, + ) -> bool: + """Clear the assignment, optionally only when it still has ``expected_id``. + + The compare-and-delete form prevents a late cleanup for conversation A + from deleting a successor assignment to conversation B. + """ + key = conversation_current(session_id) + if expected_id is None: + return bool(await self._redis.delete(key)) + + while True: + async with self._redis.pipeline(transaction=True) as pipe: + try: + await pipe.watch(key) + if _to_str(await pipe.get(key)) != expected_id: + await pipe.unwatch() + return False + pipe.multi() + pipe.delete(key) + result = await pipe.execute() + return bool(result[0]) + except WatchError: + continue + + async def expire_current_conversation(self, session_id: str, ttl: int) -> bool: + """Apply a cleanup TTL only when a conversation assignment exists.""" + return bool(await self._redis.expire(conversation_current(session_id), ttl)) + + @asynccontextmanager + async def conversation_create_lock( + self, + session_id: str, + *, + wait_timeout: float = 5.0, + lease_seconds: int = 30, + poll: float = 0.05, + ): + """Serialize get/create/assign sequences for one streaming session. + + Creation remains available if Redis locking is unhealthy: after + ``wait_timeout`` the caller runs unlocked and receives ``False``. This + avoids deadlocking audio ingestion while making the degraded mode visible + to callers and logs. + """ + key = conversation_create_lock_key(session_id) + token = uuid.uuid4().hex + deadline = time.monotonic() + wait_timeout + acquired = False + + while time.monotonic() < deadline: + acquired = bool( + await self._redis.set(key, token, nx=True, ex=lease_seconds) + ) + if acquired: + break + await asyncio.sleep(poll) + + try: + yield acquired + finally: + if acquired: + while True: + async with self._redis.pipeline(transaction=True) as pipe: + try: + await pipe.watch(key) + if _to_str(await pipe.get(key)) != token: + await pipe.unwatch() + break + pipe.multi() + pipe.delete(key) + await pipe.execute() + break + except WatchError: + continue + # ----------------------------------------------------------- field writes async def set_audio_format(self, session_id: str, audio_format: dict) -> None: await self._redis.hset( From 8c771390b426b968142bd9ed9567b4c67534ec37 Mon Sep 17 00:00:00 2001 From: Ankush <43288948+AnkushMalaker@users.noreply.github.com> Date: Wed, 22 Jul 2026 07:18:42 +0530 Subject: [PATCH 06/16] feat: add Linux Chronicle desktop tray --- extras/vault-sync/README.md | 22 ++- extras/vault-sync/main.py | 15 +- extras/vault-sync/menu_linux.py | 241 +++++++++++++++++++++++++ extras/vault-sync/pyproject.toml | 3 +- extras/vault-sync/service.py | 59 +++++- extras/vault-sync/syncthing_manager.py | 9 +- extras/vault-sync/vault_core.py | 3 + 7 files changed, 339 insertions(+), 13 deletions(-) create mode 100644 extras/vault-sync/menu_linux.py diff --git a/extras/vault-sync/README.md b/extras/vault-sync/README.md index 46610d823..0d5da42f5 100644 --- a/extras/vault-sync/README.md +++ b/extras/vault-sync/README.md @@ -1,6 +1,6 @@ -# Chronicle Vault Sync (macOS) +# Chronicle Desktop Tray and Vault Sync -A menu bar app that keeps your Chronicle Obsidian vault +A macOS menu bar / Linux system tray app that keeps your Chronicle Obsidian vault (`data/conversation_docs/{your_user}` on the server) synced to a folder on your Mac, so you can open it in **Obsidian** with full backlinks, graph view, etc. @@ -17,17 +17,27 @@ Server Syncthing ◀──── sync protocol :22000 (over Tailscale) ── └─────────────── Mac authenticates with its JWT ─────────── Obsidian ``` -## Prerequisites (Mac) +## Prerequisites ```bash brew install syncthing # the sync engine # uv (if you don't have it): curl -LsSf https://astral.sh/uv/install.sh | sh ``` +On Arch/CachyOS: + +```bash +sudo pacman -S obsidian syncthing +``` + +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`. + You also need the Chronicle **server** side running with vault sync enabled β€” see [Server setup](#server-setup-once) below. -## Setup (Mac) +## Setup ```bash cd extras/vault-sync @@ -70,6 +80,10 @@ into `~/ChronicleVault` (or `LOCAL_VAULT_DIR`). From the menu: ./start.sh uninstall ``` +On macOS this installs a launchd agent. On Linux it installs +`chronicle-desktop.service` as a systemd user service attached to the graphical +session. + ## Server setup (once) On the machine running the advanced backend: diff --git a/extras/vault-sync/main.py b/extras/vault-sync/main.py index 4afda1526..94df37fb8 100644 --- a/extras/vault-sync/main.py +++ b/extras/vault-sync/main.py @@ -1,6 +1,7 @@ -"""Chronicle Vault Sync β€” entry point with service-management subcommands.""" +"""Chronicle desktop tray β€” entry point with service-management subcommands.""" import argparse +import sys from service import install, kickstart, logs, status, uninstall @@ -11,8 +12,8 @@ def cli() -> None: parser = argparse.ArgumentParser(description="Chronicle Vault Sync") sub = parser.add_subparsers(dest="command") sub.add_parser("menu", help="Launch the menu bar app (default)") - sub.add_parser("install", help="Install as a macOS login item") - sub.add_parser("uninstall", help="Remove the macOS login item") + sub.add_parser("install", help="Install as a desktop login service") + sub.add_parser("uninstall", help="Remove the desktop login service") sub.add_parser("kickstart", help="Relaunch the menu bar app") sub.add_parser("status", help="Show service status") sub.add_parser("logs", help="Tail service logs") @@ -21,8 +22,12 @@ def cli() -> None: command = args.command or "menu" if command == "menu": - # Lazy import: macOS-only (rumps, darwin-only per pyproject.toml) - from menu_vault import main as menu_main + if sys.platform == "darwin": + from menu_vault import main as menu_main + elif sys.platform.startswith("linux"): + from menu_linux import main as menu_main + else: + raise SystemExit(f"unsupported desktop platform: {sys.platform}") menu_main() elif command == "install": diff --git a/extras/vault-sync/menu_linux.py b/extras/vault-sync/menu_linux.py new file mode 100644 index 000000000..7050efc3a --- /dev/null +++ b/extras/vault-sync/menu_linux.py @@ -0,0 +1,241 @@ +"""KDE/Linux system tray for Chronicle capture and vault sync.""" + +import logging +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 syncthing_manager import SyncthingManager +from vault_core import VaultSyncConfig, broker_pair, get_jwt_token, save_vault_dir + +logger = logging.getLogger(__name__) +SCREENPIPE_DB = Path.home() / ".screenpipe/db.sqlite" +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 + ) + return result.stdout.strip() or "unknown" + + +def _screenpipe_stats() -> str: + if not SCREENPIPE_DB.exists(): + return "No local database" + try: + uri = f"file:{SCREENPIPE_DB.resolve()}?mode=ro" + with sqlite3.connect(uri, uri=True, timeout=2) as db: + tables = { + row[0] for row in db.execute("SELECT name FROM sqlite_master WHERE type='table'") + } + frames = ( + db.execute("SELECT count(*) FROM frames").fetchone()[0] + if "frames" in tables + else 0 + ) + audio = ( + db.execute("SELECT count(*) FROM audio_chunks").fetchone()[0] + if "audio_chunks" in tables + else 0 + ) + size = sum(p.stat().st_size for p in SCREENPIPE_DB.parent.rglob("*") if p.is_file()) + return f"{frames:,} frames Β· {audio:,} audio chunks Β· {size / 1024**3:.1f} GiB" + except (OSError, sqlite3.Error) as error: + return f"Stats unavailable: {error}" + + +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 + menu = QMenu() + self.capture_status = menu.addAction("ScreenPipe: checking…") + self.collector_status = menu.addAction("Chronicle collector: checking…") + self.stats = menu.addAction("Stats: checking…") + 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, "Collector", "chronicle-screenpipe.service") + menu.addSeparator() + self.sync_status = menu.addAction("Vault sync: starting…") + self.sync_status.setEnabled(False) + menu.addAction("Open vault in Obsidian", self.open_obsidian) + menu.addAction("Choose vault folder…", self.choose_folder) + menu.addAction("Sync now / re-pair", manager.pair_async) + menu.addSeparator() + menu.addAction( + "Open Chronicle", + lambda: QDesktopServices.openUrl(QUrl(manager.config.backend_url)), + ) + menu.addAction("Quit tray", QApplication.quit) + self.setContextMenu(menu) + self.setToolTip("Chronicle") + self.timer = QTimer(self) + self.timer.timeout.connect(self.refresh) + self.timer.start(5000) + self.refresh() + + def _service_actions(self, menu: QMenu, label: str, unit: str) -> None: + submenu = menu.addMenu(label) + 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) + + def service(self, verb: str, unit: str) -> None: + subprocess.run(["systemctl", "--user", verb, unit], check=False) + QTimer.singleShot(500, self.refresh) + + def refresh(self) -> None: + self.manager.refresh_status() + capture = _unit_state("screenpipe.service") + collector = _unit_state("chronicle-screenpipe.service") + self.capture_status.setText(f"ScreenPipe: {capture}") + self.collector_status.setText(f"Chronicle collector: {collector}") + self.stats.setText(_screenpipe_stats()) + snap = self.state.snapshot() + if snap["status"] == "error": + sync = f"error β€” {snap['error']}" + elif snap["completion"] is not None: + sync = f"{snap['completion']:.0f}%" + else: + sync = snap["status"] + self.sync_status.setText(f"Vault sync: {sync}") + self.setToolTip(f"Chronicle\nScreenPipe: {capture}\nCollector: {collector}\nVault: {sync}") + + def open_obsidian(self) -> None: + vault = self.state.snapshot()["vault_dir"] + Path(vault).mkdir(parents=True, exist_ok=True) + QDesktopServices.openUrl(QUrl(f"obsidian://open?path={quote(vault)}")) + + def choose_folder(self) -> None: + current = self.state.snapshot()["vault_dir"] + chosen = QFileDialog.getExistingDirectory(None, "Choose Chronicle vault", current) + if chosen: + self.manager.set_vault_dir(chosen) + + +def main() -> None: + logging.basicConfig(level=logging.INFO) + app = QApplication(sys.argv) + app.setQuitOnLastWindowClosed(False) + if not QSystemTrayIcon.isSystemTrayAvailable(): + raise SystemExit("No system tray is available in this desktop session") + state = SharedState() + manager = VaultSyncManager(state) + manager.pair_async() + tray = ChronicleTray(state, manager) + tray.show() + try: + sys.exit(app.exec()) + finally: + manager.shutdown() diff --git a/extras/vault-sync/pyproject.toml b/extras/vault-sync/pyproject.toml index 2b1e41697..eded17581 100644 --- a/extras/vault-sync/pyproject.toml +++ b/extras/vault-sync/pyproject.toml @@ -1,7 +1,7 @@ [project] name = "chronicle-vault-sync" version = "0.1.0" -description = "macOS menu bar app that syncs your Chronicle Obsidian vault via Syncthing." +description = "Desktop tray app that syncs your Chronicle Obsidian vault via Syncthing." readme = "README.md" requires-python = ">=3.10" dependencies = [ @@ -9,5 +9,6 @@ dependencies = [ "python-dotenv>=1.0.0", "rumps>=0.4.0; sys_platform == 'darwin'", "pyobjc-framework-Cocoa>=10.0; sys_platform == 'darwin'", + "PySide6>=6.8.0; sys_platform == 'linux'", "minidisc-python>=0.1.0", ] diff --git a/extras/vault-sync/service.py b/extras/vault-sync/service.py index 2d37eb5aa..0d3584652 100644 --- a/extras/vault-sync/service.py +++ b/extras/vault-sync/service.py @@ -1,4 +1,4 @@ -"""launchd service management for Chronicle Vault Sync on macOS.""" +"""Desktop service management for Chronicle Vault Sync.""" import os import plistlib @@ -112,6 +112,9 @@ def _build_plist() -> dict: def install() -> None: + if sys.platform.startswith("linux"): + _linux_install() + return LOG_DIR.mkdir(parents=True, exist_ok=True) PLIST_PATH.parent.mkdir(parents=True, exist_ok=True) @@ -145,6 +148,9 @@ def install() -> None: def uninstall() -> None: + if sys.platform.startswith("linux"): + _linux_uninstall() + return if not PLIST_PATH.exists(): print(f"No plist found at {PLIST_PATH}") return @@ -165,6 +171,9 @@ def uninstall() -> None: def kickstart() -> None: + if sys.platform.startswith("linux"): + _linux_systemctl("restart") + return if not PLIST_PATH.exists(): print("Service not installed. Run './start.sh install' first.") return @@ -180,6 +189,9 @@ def kickstart() -> None: def status() -> None: + if sys.platform.startswith("linux"): + _linux_systemctl("status") + return if not PLIST_PATH.exists(): print(f"Service not installed (no plist at {PLIST_PATH})") return @@ -200,6 +212,11 @@ def status() -> None: def logs(follow: bool = True) -> None: + if sys.platform.startswith("linux"): + args = ["journalctl", "--user", "-u", "chronicle-desktop.service"] + args += ["-f"] if follow else ["-n", "100", "--no-pager"] + subprocess.run(args, check=False) + return if not LOG_FILE.exists(): print(f"No log file at {LOG_FILE}") return @@ -211,3 +228,43 @@ def logs(follow: bool = True) -> None: pass else: print(LOG_FILE.read_text()[-5000:]) + + +def _linux_unit_path() -> Path: + return Path.home() / ".config/systemd/user/chronicle-desktop.service" + + +def _linux_install() -> None: + uv = _find_uv() + unit = _linux_unit_path() + unit.parent.mkdir(parents=True, exist_ok=True) + unit.write_text( + "[Unit]\nDescription=Chronicle desktop tray\n" + "After=graphical-session.target network-online.target\n\n" + "[Service]\nType=simple\n" + f"WorkingDirectory={PROJECT_DIR}\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 + ) + print(f"Installed and started {unit.name}") + + +def _linux_uninstall() -> None: + unit = _linux_unit_path() + subprocess.run( + ["systemctl", "--user", "disable", "--now", unit.name], check=False + ) + unit.unlink(missing_ok=True) + subprocess.run(["systemctl", "--user", "daemon-reload"], check=True) + print(f"Removed {unit}") + + +def _linux_systemctl(action: str) -> None: + subprocess.run( + ["systemctl", "--user", action, "chronicle-desktop.service"], check=False + ) diff --git a/extras/vault-sync/syncthing_manager.py b/extras/vault-sync/syncthing_manager.py index 44bf0174c..ebbb0845a 100644 --- a/extras/vault-sync/syncthing_manager.py +++ b/extras/vault-sync/syncthing_manager.py @@ -11,6 +11,7 @@ import secrets import shutil import subprocess +import sys import time from pathlib import Path from typing import List, Optional @@ -21,6 +22,9 @@ APP_SUPPORT = ( Path.home() / "Library" / "Application Support" / "Chronicle" / "vault-sync" + if sys.platform == "darwin" + else Path(os.getenv("XDG_STATE_HOME", Path.home() / ".local/state")) + / "chronicle-vault-sync" ) SYNCTHING_HOME = APP_SUPPORT / "syncthing" # config, keys, index db APIKEY_FILE = APP_SUPPORT / "apikey" @@ -33,18 +37,19 @@ def _find_binary() -> str: - """Locate the syncthing binary, preferring PATH then common Homebrew locations.""" + """Locate the syncthing binary, preferring PATH then common install locations.""" exe = shutil.which("syncthing") if exe: return exe for candidate in ( Path("/opt/homebrew/bin/syncthing"), Path("/usr/local/bin/syncthing"), + Path("/usr/bin/syncthing"), ): if candidate.exists(): return str(candidate) raise FileNotFoundError( - "syncthing not found. Install it with: brew install syncthing" + "syncthing not found. Install it with your system package manager" ) diff --git a/extras/vault-sync/vault_core.py b/extras/vault-sync/vault_core.py index 24f2e9865..c1296f074 100644 --- a/extras/vault-sync/vault_core.py +++ b/extras/vault-sync/vault_core.py @@ -26,6 +26,9 @@ # Persisted local vault directory (set via the "Choose Vault Folder…" menu item). APP_SUPPORT = ( Path.home() / "Library" / "Application Support" / "Chronicle" / "vault-sync" + if sys.platform == "darwin" + else Path(os.getenv("XDG_STATE_HOME", Path.home() / ".local/state")) + / "chronicle-vault-sync" ) VAULT_DIR_FILE = APP_SUPPORT / "vault_dir.txt" From 328d3dda6b2409a0a305b00435912449317ce932 Mon Sep 17 00:00:00 2001 From: Ankush <43288948+AnkushMalaker@users.noreply.github.com> Date: Wed, 22 Jul 2026 07:27:25 +0530 Subject: [PATCH 07/16] feat: add ScreenPipe context thumbnails to Timeline --- .../routers/modules/device_input_routes.py | 100 +++++++++++- .../advanced/webui/src/pages/Timeline.tsx | 27 +++- backends/advanced/webui/src/services/api.ts | 6 + .../chronicle_screenpipe/collector.py | 153 ++++++++++++++---- .../tests/test_collector.py | 89 ++++++++-- 5 files changed, 327 insertions(+), 48 deletions(-) diff --git a/backends/advanced/src/advanced_omi_backend/routers/modules/device_input_routes.py b/backends/advanced/src/advanced_omi_backend/routers/modules/device_input_routes.py index 3c01891b2..10e36a035 100644 --- a/backends/advanced/src/advanced_omi_backend/routers/modules/device_input_routes.py +++ b/backends/advanced/src/advanced_omi_backend/routers/modules/device_input_routes.py @@ -32,7 +32,9 @@ utcnow, ) from advanced_omi_backend.models.user import User -from advanced_omi_backend.services.device_context import request_conversation_context_jobs +from advanced_omi_backend.services.device_context import ( + request_conversation_context_jobs, +) from advanced_omi_backend.services.memory.vault_manager import ConvDocVaultManager router = APIRouter(prefix="/device-input", tags=["device-input"]) @@ -180,6 +182,24 @@ async def ingest_activity( try: await item.insert() accepted += 1 + frame_id = incoming.metadata.get("representative_frame_id") + if ( + frame_id is not None + and "screen_context" in source.capabilities + and any( + incoming.metadata.get(key) + for key in ("app_name", "window_name", "text") + ) + ): + await DeviceInputJob( + user_id=source.user_id, + source_id=source.source_id, + kind="thumbnail", + start_at=incoming.captured_at, + end_at=incoming.ended_at, + purpose="timeline_thumbnail", + payload={"item_id": str(item.id), "frame_id": frame_id}, + ).insert() except DuplicateKeyError: duplicates += 1 existing = await DeviceInputItem.find_one( @@ -307,6 +327,37 @@ async def complete_job( return {"ok": True} +@router.post("/jobs/{job_id}/thumbnail") +async def complete_thumbnail_job( + job_id: str, + file: UploadFile = File(...), + source: CaptureSource = Depends(_device_source), +): + job = await DeviceInputJob.get(job_id) + if job is None or job.source_id != source.source_id or job.kind != "thumbnail": + raise HTTPException(status_code=404, detail="Thumbnail job not found") + item_id = job.payload.get("item_id") + item = await DeviceInputItem.get(item_id) if item_id else None + if item is None or item.source_id != source.source_id: + raise HTTPException(status_code=404, detail="Timeline item not found") + content_type = (file.content_type or "").split(";", 1)[0] + if not content_type.startswith("image/"): + raise HTTPException(status_code=415, detail="Thumbnail must be an image") + data = await file.read(_MAX_IMAGE_BYTES + 1) + if len(data) > _MAX_IMAGE_BYTES: + raise HTTPException(status_code=413, detail="Thumbnail exceeds the media limit") + item.media_data = data + item.media_filename = file.filename or "screenpipe-thumbnail.jpg" + item.media_content_type = content_type + item.content_hash = hashlib.sha256(data).hexdigest() + item.metadata = {**item.metadata, "thumbnail_available": True} + await item.save() + job.status = "complete" + job.completed_at = utcnow() + await job.save() + return {"ok": True, "item_id": str(item.id)} + + @router.get("/sources") async def list_sources(user: User = Depends(current_active_user)): rows = ( @@ -476,6 +527,16 @@ async def _immich_bytes(asset_id: str, endpoint: str) -> tuple[bytes, str]: @router.get("/items/{item_id}/thumbnail") async def context_thumbnail(item_id: str, user: User = Depends(current_active_user)): item = await _owned_item(item_id, user) + if ( + item.media_data + and item.media_content_type + and item.media_content_type.startswith("image/") + ): + return Response( + content=item.media_data, + media_type=item.media_content_type, + headers={"Cache-Control": "private, max-age=3600"}, + ) asset_id = item.metadata.get("asset_id") if item.kind != "immich_memory" or not asset_id: raise HTTPException( @@ -489,6 +550,43 @@ async def context_thumbnail(item_id: str, user: User = Depends(current_active_us ) +@router.post("/items/{item_id}/request-thumbnail") +async def request_item_thumbnail( + item_id: str, user: User = Depends(current_active_user) +): + item = await _owned_item(item_id, user) + if item.kind != "activity": + raise HTTPException(status_code=409, detail="Only activity items have frames") + if item.media_data: + return {"status": "complete"} + frame_id = ( + item.metadata.get("representative_frame_id") + or item.metadata.get("last_frame_id") + or item.metadata.get("first_frame_id") + ) + if frame_id is None: + raise HTTPException(status_code=409, detail="Activity has no source frame") + existing = await DeviceInputJob.find_one( + DeviceInputJob.source_id == item.source_id, + DeviceInputJob.kind == "thumbnail", + {"payload.item_id": item_id}, + {"status": {"$in": ["pending", "claimed"]}}, + ) + if existing: + return {"status": existing.status, "job_id": str(existing.id)} + job = DeviceInputJob( + user_id=item.user_id, + source_id=item.source_id, + kind="thumbnail", + start_at=item.captured_at, + end_at=item.ended_at, + purpose="timeline_thumbnail", + payload={"item_id": item_id, "frame_id": frame_id}, + ) + await job.insert() + return {"status": "pending", "job_id": str(job.id)} + + @router.post("/items/{item_id}/promote") async def promote_context_item(item_id: str, user: User = Depends(current_active_user)): item = await _owned_item(item_id, user) diff --git a/backends/advanced/webui/src/pages/Timeline.tsx b/backends/advanced/webui/src/pages/Timeline.tsx index 6ed5aff6a..201d1ca1e 100644 --- a/backends/advanced/webui/src/pages/Timeline.tsx +++ b/backends/advanced/webui/src/pages/Timeline.tsx @@ -1,4 +1,4 @@ -import { useMemo, useState } from 'react' +import { useEffect, useMemo, useState } from 'react' import { useMutation, useQuery } from '@tanstack/react-query' import { Activity, AppWindow, CalendarDays, Copy, Image, Link2, Monitor, RefreshCw } from 'lucide-react' import { deviceInputApi, DeviceInputItem } from '../services/api' @@ -16,6 +16,29 @@ function ItemIcon({ item }: { item: DeviceInputItem }) { return } +function TimelineThumbnail({ item }: { item: DeviceInputItem }) { + useQuery({ + queryKey: ['device-input-thumbnail-request', item.id], + queryFn: async () => (await deviceInputApi.requestThumbnail(item.id)).data, + enabled: item.kind === 'activity' && item.metadata.thumbnail_available !== true, + staleTime: Infinity, + retry: false, + }) + const thumbnail = useQuery({ + queryKey: ['device-input-thumbnail', item.id], + queryFn: async () => (await deviceInputApi.getThumbnail(item.id)).data, + enabled: item.metadata.thumbnail_available === true, + staleTime: Infinity, + }) + const url = useMemo( + () => thumbnail.data ? URL.createObjectURL(thumbnail.data) : null, + [thumbnail.data], + ) + useEffect(() => () => { if (url) URL.revokeObjectURL(url) }, [url]) + if (!url) return null + return Screen captured during this activity +} + const AUDIO_SESSION_GAP_MS = 90_000 const AUDIO_SESSION_MAX_MS = 30 * 60_000 @@ -119,6 +142,8 @@ export default function Timeline() {

{item.kind === 'audio' ? 'Audio capture' : item.metadata.app_name || item.metadata.window_name || item.metadata.text || item.kind}

{item.kind === 'audio' &&

{formatDuration(item)} Β· {item.metadata.chunk_count} chunks{item.metadata.directions?.length ? ` Β· ${item.metadata.directions.join(' + ')}` : ''}

} {item.metadata.window_name && item.metadata.window_name !== item.metadata.app_name &&

{item.metadata.window_name}

} + {item.kind === 'activity' && item.metadata.text &&

{item.metadata.text}

} + ))} {!timeline.isLoading && !visibleItems.length &&
Nothing captured for this day.
} diff --git a/backends/advanced/webui/src/services/api.ts b/backends/advanced/webui/src/services/api.ts index 39c5b5634..21f63cf9f 100644 --- a/backends/advanced/webui/src/services/api.ts +++ b/backends/advanced/webui/src/services/api.ts @@ -193,6 +193,12 @@ export const deviceInputApi = { api.get<{ items: DeviceInputItem[] }>('/api/device-input/timeline', { params: { start_at: startAt, end_at: endAt }, }), + getThumbnail: (itemId: string) => + api.get(`/api/device-input/items/${itemId}/thumbnail`, { + responseType: 'blob', + }), + requestThumbnail: (itemId: string) => + api.post(`/api/device-input/items/${itemId}/request-thumbnail`), getConversationContext: (conversationId: string) => api.get<{ items: DeviceInputItem[] }>(`/api/device-input/conversations/${conversationId}/context`), requestConversationContext: (conversationId: string) => diff --git a/extras/screenpipe-collector/chronicle_screenpipe/collector.py b/extras/screenpipe-collector/chronicle_screenpipe/collector.py index 8dd119741..8d8dc7b0a 100644 --- a/extras/screenpipe-collector/chronicle_screenpipe/collector.py +++ b/extras/screenpipe-collector/chronicle_screenpipe/collector.py @@ -95,7 +95,24 @@ def activity_key(row: sqlite3.Row) -> tuple[str, str, str]: return (row["app_name"] or "", row["window_name"] or "", row["browser_url"] or "") -def build_activity_sessions(rows: Iterable[sqlite3.Row], debounce_seconds: float = 10.0) -> list[dict[str, Any]]: +def text_excerpt(value: str | None, limit: int = 2000) -> str: + """Keep useful searchable context without mirroring ScreenPipe's full text.""" + return " ".join((value or "").split())[:limit] + + +def update_representative(current: dict[str, Any], row: sqlite3.Row) -> None: + text = text_excerpt(row["full_text"] if "full_text" in row.keys() else None) + if text: + current["text"] = text + current["text_source"] = ( + row["text_source"] if "text_source" in row.keys() else None + ) + current["representative_frame_id"] = row["id"] + + +def build_activity_sessions( + rows: Iterable[sqlite3.Row], debounce_seconds: float = 10.0 +) -> list[dict[str, Any]]: """Collapse frame headers into transitions; OCR and pixels are intentionally ignored.""" sessions: list[dict[str, Any]] = [] current: dict[str, Any] | None = None @@ -106,6 +123,7 @@ def build_activity_sessions(rows: Iterable[sqlite3.Row], debounce_seconds: float current["ended_at"] = captured current["last_frame_id"] = row["id"] current["frame_count"] += 1 + update_representative(current, row) continue if current is not None: sessions.append(current) @@ -121,13 +139,17 @@ def build_activity_sessions(rows: Iterable[sqlite3.Row], debounce_seconds: float "window_name": key[1], "browser_url": key[2], "capture_trigger": row["capture_trigger"] or "", + "representative_frame_id": row["id"], } + update_representative(current, row) if current is not None: sessions.append(current) return sessions -def fold_activity_rows(rows: Iterable[sqlite3.Row], current: dict[str, Any] | None) -> tuple[list[dict[str, Any]], dict[str, Any] | None]: +def fold_activity_rows( + rows: Iterable[sqlite3.Row], current: dict[str, Any] | None +) -> tuple[list[dict[str, Any]], dict[str, Any] | None]: """Extend the open activity across poll boundaries and return closed sessions.""" closed: list[dict[str, Any]] = [] for row in rows: @@ -137,6 +159,7 @@ def fold_activity_rows(rows: Iterable[sqlite3.Row], current: dict[str, Any] | No current["ended_at"] = captured current["last_frame_id"] = row["id"] current["frame_count"] += 1 + update_representative(current, row) continue if current is not None: closed.append(current) @@ -152,7 +175,9 @@ def fold_activity_rows(rows: Iterable[sqlite3.Row], current: dict[str, Any] | No "window_name": key[1], "browser_url": key[2], "capture_trigger": row["capture_trigger"] or "", + "representative_frame_id": row["id"], } + update_representative(current, row) return closed, current @@ -192,10 +217,14 @@ def collect_audio(self, connection: sqlite3.Connection) -> int: if not required <= columns: # ScreenPipe may expose a migration placeholder briefly before the # first audio segment initializes the final schema. - count = connection.execute("SELECT COUNT(*) FROM audio_chunks").fetchone()[0] + count = connection.execute("SELECT COUNT(*) FROM audio_chunks").fetchone()[ + 0 + ] if count == 0: return 0 - raise RuntimeError(f"unsupported ScreenPipe audio_chunks schema; missing {sorted(required - columns)}") + raise RuntimeError( + f"unsupported ScreenPipe audio_chunks schema; missing {sorted(required - columns)}" + ) cursor = self.checkpoints.get("audio") rows = connection.execute( "SELECT id, file_path, timestamp FROM audio_chunks WHERE id > ? AND timestamp IS NOT NULL ORDER BY id LIMIT 100", @@ -207,17 +236,31 @@ def collect_audio(self, connection: sqlite3.Connection) -> int: if not path.is_file(): captured = timestamp_seconds(iso_timestamp(row["timestamp"])) if time.time() - captured < 120: - logger.warning("audio chunk %s is not available yet: %s", row["id"], path) + logger.warning( + "audio chunk %s is not available yet: %s", row["id"], path + ) break self.rejections_path.parent.mkdir(parents=True, exist_ok=True) with self.rejections_path.open("a", encoding="utf-8") as rejected: - rejected.write(json.dumps({"stream": "audio", "source_item_id": row["id"], "detail": "source media missing"}) + "\n") + rejected.write( + json.dumps( + { + "stream": "audio", + "source_item_id": row["id"], + "detail": "source media missing", + } + ) + + "\n" + ) self.checkpoints.set("audio", row["id"]) continue before = path.stat() time.sleep(0.05) after = path.stat() - if before.st_size != after.st_size or before.st_mtime_ns != after.st_mtime_ns: + if ( + before.st_size != after.st_size + or before.st_mtime_ns != after.st_mtime_ns + ): break digest = hashlib.sha256(path.read_bytes()).hexdigest() content_type = mimetypes.guess_type(path.name)[0] or "audio/wav" @@ -240,7 +283,17 @@ def collect_audio(self, connection: sqlite3.Connection) -> int: logger.error("audio chunk %s rejected: %s", row["id"], response.text) self.rejections_path.parent.mkdir(parents=True, exist_ok=True) with self.rejections_path.open("a", encoding="utf-8") as rejected: - rejected.write(json.dumps({"stream": "audio", "source_item_id": row["id"], "status": response.status_code, "detail": response.text[:1000]}) + "\n") + rejected.write( + json.dumps( + { + "stream": "audio", + "source_item_id": row["id"], + "status": response.status_code, + "detail": response.text[:1000], + } + ) + + "\n" + ) self.checkpoints.set("audio", row["id"]) sent += 1 return sent @@ -249,11 +302,14 @@ def collect_activity(self, connection: sqlite3.Connection) -> int: columns = table_columns(connection, "frames") required = {"id", "timestamp", "app_name", "window_name"} if not required <= columns: - raise RuntimeError(f"unsupported ScreenPipe frames schema; missing {sorted(required - columns)}") + raise RuntimeError( + f"unsupported ScreenPipe frames schema; missing {sorted(required - columns)}" + ) optional = lambda name: name if name in columns else f"NULL AS {name}" cursor = self.checkpoints.get("frames") rows = connection.execute( - f"SELECT id, timestamp, app_name, window_name, {optional('browser_url')}, {optional('capture_trigger')} " + f"SELECT id, timestamp, app_name, window_name, {optional('browser_url')}, " + f"{optional('capture_trigger')}, {optional('full_text')}, {optional('text_source')} " "FROM frames WHERE id > ? ORDER BY id LIMIT 1000", (cursor,), ).fetchall() @@ -266,7 +322,8 @@ def collect_activity(self, connection: sqlite3.Connection) -> int: sessions = [ session for session in ([*closed, current] if current else closed) - if timestamp_seconds(session["ended_at"]) - timestamp_seconds(session["captured_at"]) + if timestamp_seconds(session["ended_at"]) + - timestamp_seconds(session["captured_at"]) >= self.config.activity_debounce_seconds ] if not sessions: @@ -282,7 +339,11 @@ def collect_activity(self, connection: sqlite3.Connection) -> int: "source_item_id": session["source_item_id"], "captured_at": session["captured_at"], "ended_at": session["ended_at"], - "metadata": {k: v for k, v in session.items() if k not in {"key", "source_item_id", "captured_at", "ended_at"}}, + "metadata": { + k: v + for k, v in session.items() + if k not in {"key", "source_item_id", "captured_at", "ended_at"} + }, } for session in sessions ] @@ -303,18 +364,46 @@ def process_job(self) -> bool: if not job: return False try: + if job["kind"] == "thumbnail": + frame_id = job.get("payload", {}).get("frame_id") + if frame_id is None: + raise RuntimeError("thumbnail job is missing frame_id") + headers = ( + {"Authorization": f"Bearer {self.config.screenpipe_token}"} + if self.config.screenpipe_token + else None + ) + thumbnail = httpx.get( + f"{self.config.screenpipe_url.rstrip('/')}/frames/{frame_id}/thumbnail", + params={"width": 640, "quality": 75}, + headers=headers, + timeout=30, + ) + thumbnail.raise_for_status() + done = self.client.post( + f"/api/device-input/jobs/{job['id']}/thumbnail", + files={ + "file": ( + f"screenpipe-frame-{frame_id}.jpg", + thumbnail.content, + thumbnail.headers.get("content-type", "image/jpeg"), + ) + }, + ) + done.raise_for_status() + return True raw_items = [] offset = 0 page_size = 500 while True: params = { "content_type": "ocr", - "start_time": iso_timestamp(job["start_at"]) - if job.get("start_at") - else None, - "end_time": iso_timestamp(job["end_at"]) - if job.get("end_at") - else None, + "start_time": ( + iso_timestamp(job["start_at"]) if job.get("start_at") else None + ), + "end_time": ( + iso_timestamp(job["end_at"]) if job.get("end_at") else None + ), "limit": page_size, "offset": offset, } @@ -341,21 +430,25 @@ def process_job(self) -> bool: frame_id = content.get("frame_id") if frame_id is None: continue - items.append({ - "source_item_id": f"frame:{frame_id}", - "captured_at": content.get("timestamp"), - "metadata": { - "frame_id": frame_id, - "app_name": content.get("app_name"), - "window_name": content.get("window_name"), - "browser_url": content.get("browser_url"), - "text": content.get("text"), - }, - }) + items.append( + { + "source_item_id": f"frame:{frame_id}", + "captured_at": content.get("timestamp"), + "metadata": { + "frame_id": frame_id, + "app_name": content.get("app_name"), + "window_name": content.get("window_name"), + "browser_url": content.get("browser_url"), + "text": content.get("text"), + }, + } + ) result = {"success": True, "items": items} except Exception as exc: result = {"success": False, "items": [], "error": str(exc)} - done = self.client.post(f"/api/device-input/jobs/{job['id']}/complete", json=result) + done = self.client.post( + f"/api/device-input/jobs/{job['id']}/complete", json=result + ) done.raise_for_status() return True diff --git a/extras/screenpipe-collector/tests/test_collector.py b/extras/screenpipe-collector/tests/test_collector.py index 2e313c886..ec08b1503 100644 --- a/extras/screenpipe-collector/tests/test_collector.py +++ b/extras/screenpipe-collector/tests/test_collector.py @@ -1,20 +1,36 @@ import sqlite3 from pathlib import Path -from chronicle_screenpipe.collector import Checkpoints, Collector, audio_duration, build_activity_sessions, fold_activity_rows, infer_audio_direction +from chronicle_screenpipe.collector import ( + Checkpoints, + Collector, + audio_duration, + build_activity_sessions, + fold_activity_rows, + infer_audio_direction, + text_excerpt, +) def test_activity_sessions_collapse_same_window(): db = sqlite3.connect(":memory:") db.row_factory = sqlite3.Row - db.execute("CREATE TABLE frames (id INTEGER, timestamp TEXT, app_name TEXT, window_name TEXT, browser_url TEXT, capture_trigger TEXT)") - db.executemany("INSERT INTO frames VALUES (?, ?, ?, ?, ?, ?)", [ - (1, "2026-07-22T10:00:00", "Code", "chronicle", None, "AppSwitch"), - (2, "2026-07-22T10:00:05", "Code", "chronicle", None, "Keystroke"), - (3, "2026-07-22T10:01:00", "Game", "Game", None, "AppSwitch"), - ]) + db.execute( + "CREATE TABLE frames (id INTEGER, timestamp TEXT, app_name TEXT, window_name TEXT, browser_url TEXT, capture_trigger TEXT)" + ) + db.executemany( + "INSERT INTO frames VALUES (?, ?, ?, ?, ?, ?)", + [ + (1, "2026-07-22T10:00:00", "Code", "chronicle", None, "AppSwitch"), + (2, "2026-07-22T10:00:05", "Code", "chronicle", None, "Keystroke"), + (3, "2026-07-22T10:01:00", "Game", "Game", None, "AppSwitch"), + ], + ) sessions = build_activity_sessions(db.execute("SELECT * FROM frames ORDER BY id")) - assert [(s["app_name"], s["frame_count"]) for s in sessions] == [("Code", 2), ("Game", 1)] + assert [(s["app_name"], s["frame_count"]) for s in sessions] == [ + ("Code", 2), + ("Game", 1), + ] def test_checkpoints_are_atomic(tmp_path: Path): @@ -31,22 +47,63 @@ def test_audio_direction_from_screenpipe_filename(): def test_activity_session_survives_poll_boundary(): db = sqlite3.connect(":memory:") db.row_factory = sqlite3.Row - db.execute("CREATE TABLE frames (id INTEGER, timestamp TEXT, app_name TEXT, window_name TEXT, browser_url TEXT, capture_trigger TEXT)") - db.executemany("INSERT INTO frames VALUES (?, ?, ?, ?, ?, ?)", [ - (1, "2026-07-22T10:00:00", "Game", "Game", None, "AppSwitch"), - (2, "2026-07-22T10:00:05", "Game", "Game", None, "VisualChange"), - (3, "2026-07-22T10:01:00", "Code", "chronicle", None, "AppSwitch"), - ]) - closed, current = fold_activity_rows(db.execute("SELECT * FROM frames WHERE id <= 1"), None) + db.execute( + "CREATE TABLE frames (id INTEGER, timestamp TEXT, app_name TEXT, window_name TEXT, browser_url TEXT, capture_trigger TEXT)" + ) + db.executemany( + "INSERT INTO frames VALUES (?, ?, ?, ?, ?, ?)", + [ + (1, "2026-07-22T10:00:00", "Game", "Game", None, "AppSwitch"), + (2, "2026-07-22T10:00:05", "Game", "Game", None, "VisualChange"), + (3, "2026-07-22T10:01:00", "Code", "chronicle", None, "AppSwitch"), + ], + ) + closed, current = fold_activity_rows( + db.execute("SELECT * FROM frames WHERE id <= 1"), None + ) assert closed == [] - closed, current = fold_activity_rows(db.execute("SELECT * FROM frames WHERE id > 1"), current) + closed, current = fold_activity_rows( + db.execute("SELECT * FROM frames WHERE id > 1"), current + ) assert closed[0]["source_item_id"] == "activity:1" assert closed[0]["frame_count"] == 2 assert current["app_name"] == "Code" +def test_activity_uses_bounded_text_and_its_frame_as_representative(): + db = sqlite3.connect(":memory:") + db.row_factory = sqlite3.Row + db.execute( + "CREATE TABLE frames (id INTEGER, timestamp TEXT, app_name TEXT, " + "window_name TEXT, browser_url TEXT, capture_trigger TEXT, " + "full_text TEXT, text_source TEXT)" + ) + db.executemany( + "INSERT INTO frames VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + [ + (1, "2026-07-22T10:00:00", "Game", "Game", None, "AppSwitch", None, None), + ( + 2, + "2026-07-22T10:00:05", + "Game", + "Game", + None, + "VisualChange", + " score\n 10 ", + "accessibility", + ), + ], + ) + session = build_activity_sessions(db.execute("SELECT * FROM frames"))[0] + assert session["text"] == "score 10" + assert session["text_source"] == "accessibility" + assert session["representative_frame_id"] == 2 + assert len(text_excerpt("x" * 3000)) == 2000 + + def test_wav_duration_is_read_from_media(tmp_path: Path): import wave + target = tmp_path / "sample.wav" with wave.open(str(target), "wb") as audio: audio.setnchannels(1) From e3514f9cf42bac5fbe9c0d6f4088c27c91fa02ad Mon Sep 17 00:00:00 2001 From: Ankush Malaker <43288948+AnkushMalaker@users.noreply.github.com> Date: Wed, 22 Jul 2026 02:15:35 +0000 Subject: [PATCH 08/16] ci: install libopus0 for advanced backend unit tests services/device_audio.py imports opuslib at module level, which needs the native Opus library; the GitHub runner doesn't ship it. Was masked until now by the earlier AUTH_SECRET_KEY collection error. --- .github/workflows/python-tests.yml | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/.github/workflows/python-tests.yml b/.github/workflows/python-tests.yml index ad82e7de1..9adc5d808 100644 --- a/.github/workflows/python-tests.yml +++ b/.github/workflows/python-tests.yml @@ -90,6 +90,11 @@ jobs: with: version: "latest" + - name: Install system dependencies + # libopus0: native library behind opuslib (services/device_audio.py), + # imported transitively during test collection. + run: sudo apt-get update && sudo apt-get install -y --no-install-recommends libopus0 + - name: Install test dependencies run: uv sync --locked --group test From f7d102c2c95224413f2eb3fcc0bfb6e79f665245 Mon Sep 17 00:00:00 2001 From: Ankush <43288948+AnkushMalaker@users.noreply.github.com> Date: Wed, 22 Jul 2026 07:49:55 +0530 Subject: [PATCH 09/16] fix: preserve meaningful ScreenPipe transitions --- .../advanced/webui/src/pages/Timeline.tsx | 2 +- .../chronicle_screenpipe/collector.py | 18 +++++++++++++--- .../tests/test_collector.py | 21 +++++++++++++++++++ 3 files changed, 37 insertions(+), 4 deletions(-) diff --git a/backends/advanced/webui/src/pages/Timeline.tsx b/backends/advanced/webui/src/pages/Timeline.tsx index 201d1ca1e..b4508fb3e 100644 --- a/backends/advanced/webui/src/pages/Timeline.tsx +++ b/backends/advanced/webui/src/pages/Timeline.tsx @@ -139,7 +139,7 @@ export default function Timeline() {
{new Date(item.captured_at).toLocaleTimeString()} {item.ended_at && `– ${new Date(item.ended_at).toLocaleTimeString()}`}
-

{item.kind === 'audio' ? 'Audio capture' : item.metadata.app_name || item.metadata.window_name || item.metadata.text || item.kind}

+

{item.kind === 'audio' ? 'Audio capture' : item.metadata.app_name || item.metadata.window_name || (item.kind === 'activity' ? 'Screen change' : item.kind)}

{item.kind === 'audio' &&

{formatDuration(item)} Β· {item.metadata.chunk_count} chunks{item.metadata.directions?.length ? ` Β· ${item.metadata.directions.join(' + ')}` : ''}

} {item.metadata.window_name && item.metadata.window_name !== item.metadata.app_name &&

{item.metadata.window_name}

} {item.kind === 'activity' && item.metadata.text &&

{item.metadata.text}

} diff --git a/extras/screenpipe-collector/chronicle_screenpipe/collector.py b/extras/screenpipe-collector/chronicle_screenpipe/collector.py index 8d8dc7b0a..766e33f77 100644 --- a/extras/screenpipe-collector/chronicle_screenpipe/collector.py +++ b/extras/screenpipe-collector/chronicle_screenpipe/collector.py @@ -110,6 +110,18 @@ def update_representative(current: dict[str, Any], row: sqlite3.Row) -> None: current["representative_frame_id"] = row["id"] +def activity_is_salient( + session: dict[str, Any], unknown_min_seconds: float = 10.0 +) -> bool: + """Keep named/textual changes immediately; gate unattributed visual noise.""" + if any(session.get(key) for key in ("app_name", "window_name", "text")): + return True + duration = timestamp_seconds(session["ended_at"]) - timestamp_seconds( + session["captured_at"] + ) + return session.get("frame_count", 0) >= 2 and duration >= unknown_min_seconds + + def build_activity_sessions( rows: Iterable[sqlite3.Row], debounce_seconds: float = 10.0 ) -> list[dict[str, Any]]: @@ -322,9 +334,9 @@ def collect_activity(self, connection: sqlite3.Connection) -> int: sessions = [ session for session in ([*closed, current] if current else closed) - if timestamp_seconds(session["ended_at"]) - - timestamp_seconds(session["captured_at"]) - >= self.config.activity_debounce_seconds + if activity_is_salient( + session, unknown_min_seconds=self.config.activity_debounce_seconds + ) ] if not sessions: self.activity_path.parent.mkdir(parents=True, exist_ok=True) diff --git a/extras/screenpipe-collector/tests/test_collector.py b/extras/screenpipe-collector/tests/test_collector.py index ec08b1503..cb6bbb005 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, + activity_is_salient, audio_duration, build_activity_sessions, fold_activity_rows, @@ -101,6 +102,26 @@ def test_activity_uses_bounded_text_and_its_frame_as_representative(): assert len(text_excerpt("x" * 3000)) == 2000 +def test_named_or_textual_changes_are_not_lost_to_time_debounce(): + base = { + "captured_at": "2026-07-22T10:00:00Z", + "ended_at": "2026-07-22T10:00:00Z", + "frame_count": 1, + "app_name": "", + "window_name": "", + } + assert activity_is_salient({**base, "window_name": "A short-lived tab"}) + assert activity_is_salient({**base, "text": "terminal output"}) + assert not activity_is_salient(base) + assert activity_is_salient( + { + **base, + "ended_at": "2026-07-22T10:00:12Z", + "frame_count": 2, + } + ) + + def test_wav_duration_is_read_from_media(tmp_path: Path): import wave From 0c177a33795cc6f79ad9ff489fabedb466913829 Mon Sep 17 00:00:00 2001 From: Ankush <43288948+AnkushMalaker@users.noreply.github.com> Date: Wed, 22 Jul 2026 08:15:31 +0530 Subject: [PATCH 10/16] fix: normalize ScreenPipe audio timestamps --- .../services/device_audio_ingest.py | 34 +++++++++++++------ .../tests/test_device_audio_ingest.py | 14 ++++++++ 2 files changed, 38 insertions(+), 10 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 601f87252..f794b2167 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 @@ -2,7 +2,7 @@ import asyncio import tempfile -from datetime import timedelta +from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any @@ -21,18 +21,26 @@ _MAX_SESSION = timedelta(minutes=30) +def _as_utc(value: datetime) -> datetime: + """Normalize Mongo's naΓ―ve UTC datetimes before ordering or arithmetic.""" + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + def group_audio_sessions(items: list[DeviceInputItem]) -> list[list[DeviceInputItem]]: sessions: list[list[DeviceInputItem]] = [] - for item in sorted(items, key=lambda row: row.captured_at): + for item in sorted(items, key=lambda row: _as_utc(row.captured_at)): if not sessions: sessions.append([item]) continue previous = sessions[-1][-1] - previous_end = previous.ended_at or previous.captured_at - session_start = sessions[-1][0].captured_at + previous_end = _as_utc(previous.ended_at or previous.captured_at) + session_start = _as_utc(sessions[-1][0].captured_at) + captured_at = _as_utc(item.captured_at) if ( - item.captured_at - previous_end > _SESSION_GAP - or item.captured_at - session_start >= _MAX_SESSION + captured_at - previous_end > _SESSION_GAP + or captured_at - session_start >= _MAX_SESSION ): sessions.append([item]) else: @@ -41,7 +49,7 @@ def group_audio_sessions(items: list[DeviceInputItem]) -> list[list[DeviceInputI async def _mix_session(items: list[DeviceInputItem], workspace: Path, output: Path) -> None: - start = min(item.captured_at for item in items) + 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] if not valid: @@ -54,7 +62,9 @@ async def _mix_session(items: list[DeviceInputItem], workspace: Path, output: Pa chains = [] labels = [] for index, item in enumerate(valid): - delay_ms = max(0, int((item.captured_at - start).total_seconds() * 1000)) + delay_ms = max( + 0, int((_as_utc(item.captured_at) - start).total_seconds() * 1000) + ) label = f"a{index}" chains.append( f"[{index}:a]aresample=16000,aformat=channel_layouts=mono,adelay={delay_ms}[{label}]" @@ -109,7 +119,9 @@ async def process_device_audio() -> dict[str, Any]: if user is None: continue for session in group_audio_sessions(source_items): - session_end = max((item.ended_at or item.captured_at) for item in session) + session_end = max( + _as_utc(item.ended_at or item.captured_at) for item in session + ) if session_end > utcnow() - _CLOSE_DELAY: continue with tempfile.TemporaryDirectory( @@ -138,7 +150,9 @@ async def process_device_audio() -> dict[str, Any]: Conversation.conversation_id == conversation_id ) if conversation is not None: - conversation.created_at = min(item.captured_at for item in session) + 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_type = "screenpipe" await conversation.save() diff --git a/backends/advanced/tests/test_device_audio_ingest.py b/backends/advanced/tests/test_device_audio_ingest.py index 271511e8a..24e4e597c 100644 --- a/backends/advanced/tests/test_device_audio_ingest.py +++ b/backends/advanced/tests/test_device_audio_ingest.py @@ -44,3 +44,17 @@ def test_continuous_capture_is_bounded_into_processing_windows(): rows = [item(str(index), start + timedelta(minutes=index)) for index in range(32)] sessions = group_audio_sessions(rows) assert [len(session) for session in sessions] == [30, 2] + + +def test_mongo_naive_and_aware_timestamps_group_together(): + naive = datetime(2026, 7, 22, 10, 0) + aware = datetime(2026, 7, 22, 10, 0, 30, tzinfo=timezone.utc) + sessions = group_audio_sessions( + [ + item("mongo-naive", naive), + item("api-aware", aware), + ] + ) + assert [[row.source_item_id for row in session] for session in sessions] == [ + ["mongo-naive", "api-aware"] + ] From 20a69b6e726b8020f08c7040030ac84623f49470 Mon Sep 17 00:00:00 2001 From: Ankush Malaker <43288948+AnkushMalaker@users.noreply.github.com> Date: Wed, 22 Jul 2026 02:59:18 +0000 Subject: [PATCH 11/16] fix: ignore opuslib SyntaxWarning under pytest error filter opuslib 3.x contains `is not 0`, a SyntaxWarning on py3.12. With filterwarnings=error it becomes a SyntaxError on first import in a fresh venv (CI), killing collection. Cached .pyc masked it locally. --- backends/advanced/pyproject.toml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/backends/advanced/pyproject.toml b/backends/advanced/pyproject.toml index dec87b445..66f519b86 100644 --- a/backends/advanced/pyproject.toml +++ b/backends/advanced/pyproject.toml @@ -104,6 +104,9 @@ filterwarnings = [ "ignore::UserWarning", "ignore::DeprecationWarning", "ignore::PendingDeprecationWarning", + # opuslib 3.x uses `is not 0`, a SyntaxWarning on py3.12 that "error" would + # promote to SyntaxError on first (uncached) import. Third-party; not ours. + "ignore::SyntaxWarning:.*opuslib.*", ] [tool.coverage.run] From abb59fc5b7aab1c11f42a11f7b388378b27301b8 Mon Sep 17 00:00:00 2001 From: Ankush Malaker <43288948+AnkushMalaker@users.noreply.github.com> Date: Wed, 22 Jul 2026 03:05:26 +0000 Subject: [PATCH 12/16] fix: make memory.processed fire under the no-api mock config MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two layers were breaking the plugin-event tests: 1. mock-services.yml had no defaults.fallback_llm, so the memory agent's note-guarantee recovery pass (force_fallback=True) raised 'No fallback LLM is configured' and the memory job failed with nothing recorded. Add mock-llm-fallback (same mock server, distinct name β€” a fallback identical to defaults.llm is rejected as pointless). 2. chronicle.py treated a recovery-pass exception as fatal, skipping the source-preserving fallback note entirely. The note guarantee now degrades: log the recovery failure, write the fallback note, return its path β€” so memory.processed still fires and the vault keeps the conversation. Also lands the speaker-rename guard in _speaker_rename_guidance: conversation-scoped relabels between real named people no longer trigger a vault-wide rename_person (which merged the wrong person's history); only placeholder labels (Speaker N / Unknown Speaker N) may be globally renamed. --- .../services/memory/providers/chronicle.py | 89 ++++++++++++++----- tests/configs/mock-services.yml | 19 ++++ 2 files changed, 88 insertions(+), 20 deletions(-) diff --git a/backends/advanced/src/advanced_omi_backend/services/memory/providers/chronicle.py b/backends/advanced/src/advanced_omi_backend/services/memory/providers/chronicle.py index adfb5d433..43fd80dda 100644 --- a/backends/advanced/src/advanced_omi_backend/services/memory/providers/chronicle.py +++ b/backends/advanced/src/advanced_omi_backend/services/memory/providers/chronicle.py @@ -14,6 +14,7 @@ """ import logging +import re import time from datetime import datetime, timezone from pathlib import Path @@ -233,14 +234,38 @@ async def _run_agent_with_note_guarantee( "short or low-information transcript, summarize the exact utterance rather " "than leaving either section blank." ) - recovery = await agent_class(user_root, force_fallback=True).run( - transcript, - source_id, - date=trusted_date, - duration_minutes=source_duration_minutes, - title=source_title, - guidance=recovery_guidance, - ) + # The recovery pass is best-effort: it may fail outright (no + # defaults.fallback_llm configured, fallback unreachable, ...). The note + # guarantee must survive that β€” degrade to an empty recovery result and + # let the source-preserving fallback note below do its job, rather than + # failing the whole memory job with nothing recorded. + try: + recovery = await agent_class(user_root, force_fallback=True).run( + transcript, + source_id, + date=trusted_date, + duration_minutes=source_duration_minutes, + title=source_title, + guidance=recovery_guidance, + ) + except Exception as exc: # noqa: BLE001 - degrade, never lose the note + # Lazy import: circular dependency (agent β†’ memory_agent β†’ llm_client β†’ + # back into providers), same as _agent_class above. + from ..agent.memory_agent import MemoryAgentResult + + memory_logger.error( + "Memory-agent recovery pass failed for %s (%s); falling back to " + "the source-preserving conversation note", + source_id, + exc, + ) + recovery = MemoryAgentResult( + conversation_id=source_id, + rounds=0, + touched=[], + summary="", + errors=[f"recovery pass failed: {exc}"], + ) recovery.rounds += result.rounds recovery.tool_calls += result.tool_calls recovery.touched = list(dict.fromkeys((*result.touched, *recovery.touched))) @@ -440,27 +465,51 @@ async def _record_agent_touches( ), ) - @staticmethod - def _speaker_rename_guidance(transcript_diff: Optional[list]) -> str: + # Diarization placeholders ("Speaker 0", "Unknown Speaker 1") β€” the only labels a + # conversation-scoped diff may globally rename. A person note under a real name + # aggregates facts from many conversations, so renaming it from one conversation's + # relabel merges the wrong person's whole history (ankush.md -> roshan.md, 2026-07-17). + _PLACEHOLDER_SPEAKER_RE = re.compile( + r"^(unknown\s+)?speaker[\s_]*\d+$", re.IGNORECASE + ) + + @classmethod + def _speaker_rename_guidance(cls, transcript_diff: Optional[list]) -> str: """Turn a speaker diff into an instruction to rename the matching person notes.""" if not transcript_diff: return "" renames: dict[str, str] = {} + relabels: dict[str, str] = {} for ch in transcript_diff: if isinstance(ch, dict) and ch.get("type") == "speaker_change": old, new = ch.get("old_speaker"), ch.get("new_speaker") if old and new and old != new: - renames[old] = new - if not renames: + if cls._PLACEHOLDER_SPEAKER_RE.match(old.strip()): + renames[old] = new + else: + relabels[old] = new + parts: list[str] = [] + if renames: + pairs = "; ".join(f"'{o}' is now '{n}'" for o, n in renames.items()) + parts.append( + f"Placeholder speakers were identified: {pairs}. For each, if a " + "People/.md note exists, call rename_person(old, new) FIRST " + "β€” it renames the note and rewrites every [[wikilink]] across the vault β€” " + "then record the conversation and update the renamed person notes. Do not " + "leave notes under placeholder speaker labels." + ) + if relabels: + pairs = "; ".join(f"'{o}' is now '{n}'" for o, n in relabels.items()) + parts.append( + f"Attribution changed between named people IN THIS CONVERSATION ONLY: " + f"{pairs}. Fix the conversation note and move only facts sourced from " + "this conversation between the affected person notes. Do NOT call " + "rename_person for these β€” both notes describe real people whose facts " + "come from many other conversations." + ) + if not parts: return "" - pairs = "; ".join(f"'{o}' is now '{n}'" for o, n in renames.items()) - return ( - "This is a REPROCESS after speaker re-identification. Speaker labels changed: " - f"{pairs}. For each change, if a People/.md note exists, call " - "rename_person(old, new) FIRST β€” it renames the note and rewrites every " - "[[wikilink]] across the vault β€” then record the conversation and update the " - "renamed person notes. Do not leave notes under the old speaker labels." - ) + return "This is a REPROCESS after speaker re-identification. " + " ".join(parts) # ========================================================================= # SEARCH diff --git a/tests/configs/mock-services.yml b/tests/configs/mock-services.yml index c2e12934f..c670d5c1d 100644 --- a/tests/configs/mock-services.yml +++ b/tests/configs/mock-services.yml @@ -3,6 +3,12 @@ chat: memories and conversation history. defaults: embedding: mock-embed + # Without a fallback the memory agent's note-guarantee recovery pass + # (force_fallback=True) raises "No fallback LLM is configured" and the whole + # memory job fails before the source-preserving fallback note can be written. + # Must be a DIFFERENT model name than defaults.llm β€” a fallback identical to + # the primary is rejected as pointless (llm_client._get_fallback_model_def). + fallback_llm: mock-llm-fallback llm: mock-llm stt: mock-stt stt_stream: mock-stt-stream @@ -25,6 +31,19 @@ models: model_type: llm model_url: http://host.docker.internal:11435/v1 name: mock-llm +- api_family: openai + api_key: dummy-key-not-used + description: Same mock LLM server under a distinct name, so it passes the + fallback!=primary check and the recovery pass actually runs in tests + model_name: gpt-4o-mini + model_output: json + model_params: + max_tokens: 2000 + temperature: 0.2 + model_provider: openai + model_type: llm + model_url: http://host.docker.internal:11435/v1 + name: mock-llm-fallback - api_family: openai api_key: dummy-key-not-used description: Mock embedding server for testing (local) From 260453eb483a3474f9deefad00a12167e8ca3cbe Mon Sep 17 00:00:00 2001 From: Ankush Malaker <43288948+AnkushMalaker@users.noreply.github.com> Date: Wed, 22 Jul 2026 03:29:32 +0000 Subject: [PATCH 13/16] ci: real redis/mongo services for backend unit tests; upload robot service logs - test_vault_rename_person needs Redis (vault locks are Redis-backed and fail closed) and test_leading_silence_trim_db needs Mongo; both passed locally only because a dev stack was listening. Give the CI job real service containers on the ports the code defaults to. - robot-tests saved backend/worker logs to files but never uploaded them, leaving container-side failures undiagnosable. Upload them. --- .github/workflows/python-tests.yml | 12 ++++++++++++ .github/workflows/robot-tests.yml | 9 +++++++++ 2 files changed, 21 insertions(+) diff --git a/.github/workflows/python-tests.yml b/.github/workflows/python-tests.yml index 9adc5d808..155be63d1 100644 --- a/.github/workflows/python-tests.yml +++ b/.github/workflows/python-tests.yml @@ -75,6 +75,18 @@ jobs: defaults: run: working-directory: backends/advanced + services: + # Some backend tests hit real local services (vault locks are Redis-backed + # and fail closed; test_leading_silence_trim_db uses Mongo). Match the + # defaults the code assumes: redis://localhost:6379, mongodb://localhost:27018. + redis: + image: redis:7-alpine + ports: + - 6379:6379 + mongo: + image: mongo:8 + ports: + - 27018:27017 steps: - name: Checkout code diff --git a/.github/workflows/robot-tests.yml b/.github/workflows/robot-tests.yml index 1d3fffa79..9eb226923 100644 --- a/.github/workflows/robot-tests.yml +++ b/.github/workflows/robot-tests.yml @@ -100,6 +100,15 @@ jobs: echo "βœ“ Logs saved to backends/advanced/logs/" ls -lh logs/ + - name: Upload service logs + if: always() + uses: actions/upload-artifact@v4 + with: + name: robot-service-logs + path: backends/advanced/logs/ + if-no-files-found: warn + retention-days: 14 + - name: Check if test results exist if: always() id: check_results From 85d561fad387257144230a41d0dafdcdd3ec8d0f Mon Sep 17 00:00:00 2001 From: Ankush Malaker <43288948+AnkushMalaker@users.noreply.github.com> Date: Wed, 22 Jul 2026 03:50:22 +0000 Subject: [PATCH 14/16] fix: resolve host.docker.internal in test containers on Docker/CI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every real HTTP caller in the test stack (memory-agent LLM, streaming STT consumer) addresses the mock services as host.docker.internal, but Docker on Linux does not resolve that name without an explicit host-gateway mapping β€” CI workers logged 'Name or service not known' for every LLM and streaming-STT call. This broke memory.processed events and streaming speech detection (open_conversation) in the no-api suite, while passing locally under podman, which resolves the name natively. Prod compose already carries the same extra_hosts. --- backends/advanced/docker-compose-test.yml | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/backends/advanced/docker-compose-test.yml b/backends/advanced/docker-compose-test.yml index 3668ab7fe..317a6995a 100644 --- a/backends/advanced/docker-compose-test.yml +++ b/backends/advanced/docker-compose-test.yml @@ -62,6 +62,11 @@ services: - LANGFUSE_BASE_URL=http://langfuse-web-test:3000 - LANGFUSE_PUBLIC_KEY=pk-lf-test-public-key - LANGFUSE_SECRET_KEY=sk-lf-test-secret-key + extra_hosts: + # Mock services (LLM/STT) publish ports on the host and are addressed as + # host.docker.internal. Docker on Linux (CI) does not resolve that name + # without an explicit host-gateway mapping; podman resolves it natively. + - "host.docker.internal:host-gateway" depends_on: mongo-test: condition: service_healthy @@ -383,6 +388,10 @@ services: - LANGFUSE_BASE_URL=http://langfuse-web-test:3000 - LANGFUSE_PUBLIC_KEY=pk-lf-test-public-key - LANGFUSE_SECRET_KEY=sk-lf-test-secret-key + extra_hosts: + # See chronicle-backend-test: required for Docker-on-Linux (CI) to reach + # the mock LLM/STT services published on the host. + - "host.docker.internal:host-gateway" depends_on: # NOTE: no dependency on chronicle-backend-test. The backend's /readiness # healthcheck requires the worker fleet heartbeat in Redis, so gating the From f51021133cf59ac3442e4d4f51147f03c89c0b79 Mon Sep 17 00:00:00 2001 From: Ankush Malaker <43288948+AnkushMalaker@users.noreply.github.com> Date: Wed, 22 Jul 2026 04:17:11 +0000 Subject: [PATCH 15/16] fix: runtime config overrides survive reloads save_config_section() patched the in-memory config cache once, but any subsequent reload_config() (frequent in normal operation) rebuilt the cache from disk and silently reverted the change whenever CONFIG_FILE points away from config.yml (test environments). The always-persist robot test toggles always_persist_enabled via /api/misc-settings and raced exactly this: on slow CI a reload landed between the toggle and the stream open, so a placeholder conversation appeared despite always_persist=false. Register overrides and re-apply them on every load instead. --- .../src/advanced_omi_backend/config_loader.py | 28 +++++++++++++++---- 1 file changed, 22 insertions(+), 6 deletions(-) diff --git a/backends/advanced/src/advanced_omi_backend/config_loader.py b/backends/advanced/src/advanced_omi_backend/config_loader.py index e91f71594..80fcd4fc4 100644 --- a/backends/advanced/src/advanced_omi_backend/config_loader.py +++ b/backends/advanced/src/advanced_omi_backend/config_loader.py @@ -16,6 +16,11 @@ # Global config cache _config_cache: Optional[DictConfig] = None +# Runtime overrides registered by save_config_section(), keyed by section path. +# Re-applied on every load_config() so they survive cache reloads even when +# CONFIG_FILE points away from config.yml (test environments). +_runtime_overrides: dict = {} + def get_config_dir() -> Path: """Get config directory path (single source of truth).""" @@ -117,6 +122,14 @@ def load_config(force_reload: bool = False) -> DictConfig: f"{[m.get('name') for m in extra_defaults]}" ) + # Re-apply runtime overrides saved via save_config_section(). When + # CONFIG_FILE points away from config.yml (test environments), the saved + # values exist only in memory β€” without this they silently revert on the + # next reload, so a runtime toggle only "took" until some unrelated code + # path refreshed the config. + for section_path, values in _runtime_overrides.items(): + OmegaConf.update(merged, section_path, values, merge=True) + # Cache result _config_cache = merged @@ -193,13 +206,16 @@ def save_config_section(section_path: str, values: dict) -> bool: # Save back to file OmegaConf.save(existing_config, config_path) - # Reload config from the primary config file (CONFIG_FILE env var) - merged = reload_config() + # Register a runtime override BEFORE reloading: when CONFIG_FILE points + # to a different file than config.yml (test configs), the value we just + # saved is not in the file load_config() reads, so it must be re-applied + # on every load β€” a one-shot in-memory patch would silently revert on + # the next reload_config() from any code path. + _runtime_overrides[section_path] = values - # Also apply the values to the in-memory cache directly. - # This is needed when CONFIG_FILE points to a different file than config.yml - # (e.g., test configs), so the saved values still take effect at runtime. - OmegaConf.update(merged, section_path, values, merge=True) + # Reload config from the primary config file (CONFIG_FILE env var); + # load_config() re-applies _runtime_overrides on top. + reload_config() logger.info(f"Saved config section '{section_path}' to {config_path}") return True From 662e3214c2ac9e2b106a673b77ea74e036b28441 Mon Sep 17 00:00:00 2001 From: Ankush Malaker <43288948+AnkushMalaker@users.noreply.github.com> Date: Wed, 22 Jul 2026 04:24:24 +0000 Subject: [PATCH 16/16] fix: first runtime settings save no longer fails without config.yml MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit save_config_section() passed a plain {} to OmegaConf.update() when config.yml didn't exist yet, raising 'Unexpected type: {}' β€” so the very first runtime settings save on a fresh install failed (and the API reported it only in the response body, still HTTP 200). --- .../advanced/src/advanced_omi_backend/config_loader.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/backends/advanced/src/advanced_omi_backend/config_loader.py b/backends/advanced/src/advanced_omi_backend/config_loader.py index 80fcd4fc4..df263d98b 100644 --- a/backends/advanced/src/advanced_omi_backend/config_loader.py +++ b/backends/advanced/src/advanced_omi_backend/config_loader.py @@ -195,8 +195,11 @@ def save_config_section(section_path: str, values: dict) -> bool: try: config_path = get_config_dir() / "config.yml" - # Load existing config - existing_config = {} + # Load existing config. Must be a DictConfig even when the file doesn't + # exist yet β€” OmegaConf.update() raises "Unexpected type" on a plain + # dict, which made the very first runtime settings save fail silently + # on installs without a config.yml. + existing_config = OmegaConf.create({}) if config_path.exists(): existing_config = OmegaConf.load(config_path)