From 36e2b650b7a91bd3f9e10e4851030416789df20c Mon Sep 17 00:00:00 2001 From: Philipp Date: Tue, 29 Sep 2026 10:58:36 +0200 Subject: [PATCH 1/2] feat(curation): group sessions by their topic page Sessions from container folders (home, code, emails, work, ...) mixed unrelated work into pages like `topics/emails`. - New `session_topics` table (SQLite 0004, PostgreSQL 0007) records the page a session belongs to, or NULL when it holds nothing durable. - `session_targets` resolves every session: a routed page first; sessions in container folders wait for routing; others go to the covering `topics/` page of their folder, else `topics/`. - The backlog and the curator's evidence pack group by that page, so two folders covered by one page yield one backlog item. - `curation.generic_workspaces` (env WIKIBRICKS_CURATION_GENERIC_WORKSPACES) lists the container folder names. Implemented by GLM 5.3 Flash; PostgreSQL tests rerun in review. Co-authored-by: Isaac --- CHANGELOG.md | 4 + src/wikibricks/config/__init__.py | 27 ++ src/wikibricks/config/defaults.yml | 12 + src/wikibricks/curation/backlog.py | 164 +++++++++-- .../sql/migrations/0007_session_topics.sql | 6 + .../sql/sqlite/0004_session_topics.sql | 6 + src/wikibricks/storage/store.py | 4 +- src/wikibricks_curator/curator.py | 9 +- src/wikibricks_curator/evidence.py | 24 +- tests/test_curation_backlog.py | 68 +++-- tests/test_local_curator.py | 28 +- tests/test_migration_immutability.py | 6 + tests/test_postgres_store.py | 2 +- tests/test_topic_routing.py | 256 ++++++++++++++++++ 14 files changed, 543 insertions(+), 73 deletions(-) create mode 100644 src/wikibricks/sql/migrations/0007_session_topics.sql create mode 100644 src/wikibricks/sql/sqlite/0004_session_topics.sql create mode 100644 tests/test_topic_routing.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 03c66cb..d994b7d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,9 @@ # Changelog +## Unreleased + +- Group curation sessions by routed topic pages instead of workspace folder names. + ## 0.13.0 - 2026-09-29 - Added optional `wikibricks-curator propose` for local backlog curation. It diff --git a/src/wikibricks/config/__init__.py b/src/wikibricks/config/__init__.py index 9180e00..b9956b4 100644 --- a/src/wikibricks/config/__init__.py +++ b/src/wikibricks/config/__init__.py @@ -16,6 +16,7 @@ "database": {"path": None, "url": None}, "search": {"default_results": None, "maximum_results": None}, "maintenance": {"prune_archived_sessions_after_days": None}, + "curation": {"generic_workspaces": None}, "automation": { "enabled": None, "poll_seconds": None, @@ -41,6 +42,7 @@ class WikiBricksConfig: search_default_results: int search_maximum_results: int prune_archived_sessions_after_days: int | None + curation_generic_workspaces: tuple[str, ...] automation_enabled: bool automation_poll_seconds: int automation_local_maintenance_hours: int @@ -115,6 +117,21 @@ def _string(value: Any, path: str, *, optional: bool = False) -> str | None: return value.strip() +def _string_list(value: Any, path: str) -> list[str]: + if not isinstance(value, list) or not all(isinstance(item, str) for item in value): + raise ValueError(f"{path} must be a list of strings") + if any(not item.strip() for item in value): + raise ValueError(f"{path} must not contain empty strings") + return [item.strip() for item in value] + + +def _comma_separated_strings(value: str) -> list[str]: + values = [item.strip() for item in value.split(",")] + if any(not item for item in values): + raise ValueError("must be a comma-separated list of non-empty strings") + return values + + def _environment_overlay(environ: Mapping[str, str]) -> dict[str, Any]: result: dict[str, Any] = {} mappings = { @@ -127,6 +144,11 @@ def _environment_overlay(environ: Mapping[str, str]) -> dict[str, Any]: "prune_archived_sessions_after_days", int, ), + "WIKIBRICKS_CURATION_GENERIC_WORKSPACES": ( + "curation", + "generic_workspaces", + _comma_separated_strings, + ), "WIKIBRICKS_SYNC_BATCH_SIZE": ("sync", "batch_size", int), "WIKIBRICKS_SYNC_APPLY_POLICY": ("sync", "apply_policy", str), "WIKIBRICKS_AUTOMATION_ENABLED": ("automation", "enabled", str), @@ -215,6 +237,10 @@ def load_config( minimum=1, maximum=36500, ) + generic_workspaces = _string_list( + value["curation"]["generic_workspaces"], + "curation.generic_workspaces", + ) automation_enabled = _boolean( value["automation"]["enabled"], "automation.enabled", @@ -260,6 +286,7 @@ def load_config( search_default_results=default_results, search_maximum_results=maximum_results, prune_archived_sessions_after_days=retention, + curation_generic_workspaces=tuple(generic_workspaces), automation_enabled=automation_enabled, automation_poll_seconds=poll_seconds, automation_local_maintenance_hours=local_maintenance_hours, diff --git a/src/wikibricks/config/defaults.yml b/src/wikibricks/config/defaults.yml index 0809dcd..1ce7ee1 100644 --- a/src/wikibricks/config/defaults.yml +++ b/src/wikibricks/config/defaults.yml @@ -7,6 +7,18 @@ search: maximum_results: 20 maintenance: prune_archived_sessions_after_days: null +curation: + generic_workspaces: + - code + - emails + - work + - projects + - repos + - src + - documents + - desktop + - downloads + - tmp automation: enabled: true poll_seconds: 300 diff --git a/src/wikibricks/curation/backlog.py b/src/wikibricks/curation/backlog.py index e044064..1d62e3f 100644 --- a/src/wikibricks/curation/backlog.py +++ b/src/wikibricks/curation/backlog.py @@ -5,11 +5,23 @@ from pathlib import Path from typing import Any +from wikibricks.config import load_config + def _normalize(value: str) -> str: return value.strip().lower().replace(" ", "-").replace("_", "-") +def _normalized_name(value: str) -> str: + return _normalize(Path(value).name) + + +def _unrouted_workspace(workspace: str | None, *, generic: set[str], home: Path) -> bool: + if workspace is None or not workspace or Path(workspace) == home: + return True + return _normalized_name(workspace) in generic + + def _timestamp(value: str | datetime) -> datetime: if isinstance(value, datetime): parsed = value @@ -21,56 +33,70 @@ def _timestamp(value: str | datetime) -> datetime: def curation_backlog( - sessions: Iterable[tuple[str | None, str | datetime]], + sessions: Iterable[tuple[str, str | None, str | datetime]], pages: Iterable[tuple[str, str, str | datetime]], *, now: datetime, days: int = 7, limit: int = 5, - home: Path | None = None, + targets: dict[str, str | None], ) -> list[dict[str, Any]]: """Return recent sessions that are newer than their covering pages.""" now = _timestamp(now) - home = home or Path.home() cutoff = now - timedelta(days=days) + page_updates = { + path: _timestamp(updated_at) for path, _, updated_at in pages if not path.startswith("_meta/") + } + page_titles = {path: title for path, title, _ in pages} projects: dict[str, dict[str, Any]] = {} - for workspace, updated_at in sessions: - if workspace is None or not workspace or Path(workspace) == home: + for session_id, workspace, updated_at in sessions: + target = targets.get(session_id) + if target is None: continue timestamp = _timestamp(updated_at) if timestamp < cutoff or timestamp > now: continue - project = _normalize(Path(workspace).name) + project = Path(target).name item = projects.setdefault( - project, + target, { "project": project, + "living_page": target, "workspace": workspace, + "workspaces": set(), "session_timestamps": [], }, ) + if workspace is not None and workspace: + item["workspaces"].add(workspace) + if ( + item["workspace"] is None + or not item["workspace"] + or workspace < item["workspace"] + ): + item["workspace"] = workspace item["session_timestamps"].append(timestamp) - covering: dict[str, list[tuple[str, datetime]]] = {} - for path, title, updated_at in pages: - if path.startswith("_meta/"): - continue - normalized_path = _normalize(path) - normalized_title = _normalize(title) - timestamp = _timestamp(updated_at) - for project, item in projects.items(): - if project in normalized_path or project in normalized_title: - covering.setdefault(project, []).append((path, timestamp)) - result: list[dict[str, Any]] = [] - for project, item in projects.items(): - project_pages = sorted(covering.get(project, [])) + for target, item in projects.items(): + project = item["project"] + living_page = item["living_page"] + last_page_update = page_updates.get(living_page) + project_pages: list[str] = [] + if last_page_update is not None: + project_pages.append(living_page) + project_pages.extend( + path + for path in sorted(page_updates) + if path != living_page + and path.startswith("topics/") + and ( + project in _normalize(path) + or project in _normalize(page_titles.get(path, "")) + ) + ) last_session_at = max(item["session_timestamps"]) - if project_pages: - last_page_update = max(timestamp for _, timestamp in project_pages) - else: - last_page_update = None new_sessions = sum( last_page_update is None or timestamp > last_page_update for timestamp in item["session_timestamps"] @@ -81,8 +107,10 @@ def curation_backlog( result.append( { **item, + "workspaces": sorted(item["workspaces"]), + "workspace": item["workspace"], "new_sessions": new_sessions, - "pages": [path for path, _ in project_pages[:3]], + "pages": project_pages[:3], "last_page_update": ( last_page_update.isoformat() if last_page_update else None ), @@ -95,16 +123,100 @@ def curation_backlog( )[:limit] +def session_targets( + conn: Any, + *, + generic: Iterable[str] | None = None, + home: Path | None = None, +) -> dict[str, str | None]: + """Route every session to the durable page that covers its work.""" + home = home or Path.home() + normalized_generic = { + _normalize(value) for value in ( + load_config().curation_generic_workspaces + if generic is None + else generic + ) + } + session_rows = conn.execute( + "SELECT s.session_id, s.workspace, t.session_id, t.page_path FROM sessions s " + "LEFT JOIN session_topics t ON t.session_id = s.session_id" + ).fetchall() + page_rows = conn.execute( + "SELECT p.path, v.title FROM pages p " + "JOIN page_versions v ON v.version_id = p.current_version_id " + "WHERE p.status = 'active' AND p.path LIKE 'topics/%' ORDER BY p.path" + ).fetchall() + pages = [(row[0], row[1]) for row in page_rows] + targets: dict[str, str | None] = {} + for session_id, workspace, topic_session_id, routed_page in session_rows: + if topic_session_id is not None: + targets[session_id] = routed_page + continue + if _unrouted_workspace(workspace, generic=normalized_generic, home=home): + targets[session_id] = None + continue + key = _normalized_name(workspace) + target = next( + ( + path + for path, title in pages + if key in _normalize(path) or key in _normalize(title) + ), + f"topics/{key}", + ) + targets[session_id] = target + return targets + + +def unrouted_sessions( + conn: Any, + *, + since: str | datetime, + generic: Iterable[str] | None = None, + home: Path | None = None, +) -> list[str]: + """Return container-folder sessions still waiting for topic routing.""" + home = home or Path.home() + normalized_generic = { + _normalize(value) for value in ( + load_config().curation_generic_workspaces + if generic is None + else generic + ) + } + cutoff = _timestamp(since) + rows = conn.execute( + "SELECT s.session_id, s.workspace, " + "COALESCE(s.source_updated_at, s.updated_at) FROM sessions s " + "LEFT JOIN session_topics t ON t.session_id = s.session_id " + "WHERE t.session_id IS NULL ORDER BY s.session_id" + ).fetchall() + return [ + session_id + for session_id, workspace, updated_at in rows + if _timestamp(updated_at) >= cutoff + and _unrouted_workspace(workspace, generic=normalized_generic, home=home) + ] + + def load_curation_backlog(conn: Any, **kwargs: Any) -> list[dict[str, Any]]: """Load sessions and active pages, then calculate the curation backlog.""" # Count sessions by when the work happened, not when a backfill imported them. sessions = conn.execute( - "SELECT workspace, COALESCE(source_updated_at, updated_at) FROM sessions" + "SELECT s.session_id, s.workspace, " + "COALESCE(s.source_updated_at, s.updated_at) FROM sessions s" ).fetchall() pages = conn.execute( "SELECT p.path, v.title, p.updated_at FROM pages p " "JOIN page_versions v ON v.version_id = p.current_version_id " "WHERE p.status = 'active'" ).fetchall() + home = kwargs.pop("home", None) + generic = kwargs.pop("generic", None) + if generic is None: + generic = load_config().curation_generic_workspaces + targets = session_targets(conn, generic=generic, home=home) kwargs.setdefault("now", datetime.now(timezone.utc)) + kwargs["targets"] = targets return curation_backlog(sessions, pages, **kwargs) diff --git a/src/wikibricks/sql/migrations/0007_session_topics.sql b/src/wikibricks/sql/migrations/0007_session_topics.sql new file mode 100644 index 0000000..80daa18 --- /dev/null +++ b/src/wikibricks/sql/migrations/0007_session_topics.sql @@ -0,0 +1,6 @@ +CREATE TABLE session_topics ( + session_id uuid PRIMARY KEY REFERENCES sessions(session_id) ON DELETE CASCADE, + page_path text, + origin text NOT NULL, + created_at timestamptz NOT NULL DEFAULT now() +); diff --git a/src/wikibricks/sql/sqlite/0004_session_topics.sql b/src/wikibricks/sql/sqlite/0004_session_topics.sql new file mode 100644 index 0000000..b11f0f9 --- /dev/null +++ b/src/wikibricks/sql/sqlite/0004_session_topics.sql @@ -0,0 +1,6 @@ +CREATE TABLE IF NOT EXISTS session_topics ( + session_id TEXT PRIMARY KEY REFERENCES sessions(session_id) ON DELETE CASCADE, + page_path TEXT, + origin TEXT NOT NULL, + created_at TEXT NOT NULL +); diff --git a/src/wikibricks/storage/store.py b/src/wikibricks/storage/store.py index 3a2109c..50987ca 100644 --- a/src/wikibricks/storage/store.py +++ b/src/wikibricks/storage/store.py @@ -76,8 +76,8 @@ def migrate(self) -> None: def clear_all(self) -> None: tables = ( - "remote_maintenance_runs, curation_conflicts, curation_receipts, " - "page_aliases, curation_patches, " + "session_topics, remote_maintenance_runs, curation_conflicts, " + "curation_receipts, page_aliases, curation_patches, " "curation_runs, sync_replicas, archive_batch_events, archive_events, " "archive_batches, curated_pages, archive_pages, sync_state, sync_outbox, " "session_search_chunks, session_event_versions, session_events, sessions, " diff --git a/src/wikibricks_curator/curator.py b/src/wikibricks_curator/curator.py index 2759177..1dc7922 100644 --- a/src/wikibricks_curator/curator.py +++ b/src/wikibricks_curator/curator.py @@ -25,13 +25,6 @@ Chat = Callable[[str, dict[str, Any], dict[str, Any]], dict[str, Any]] -def _living_page(item: dict[str, Any]) -> str: - return next( - (path for path in item.get("pages", []) if path.startswith("topics/")), - f"topics/{item['project']}", - ) - - def _proposal_result(raw: dict[str, Any]) -> list[dict[str, Any]]: return [ { @@ -149,7 +142,7 @@ def run_curator( for item in backlog: project = item["project"] - living_page = _living_page(item) + living_page = item["living_page"] result: dict[str, Any] = { "project": project, "living_page": living_page, diff --git a/src/wikibricks_curator/evidence.py b/src/wikibricks_curator/evidence.py index c2e3510..f407d05 100644 --- a/src/wikibricks_curator/evidence.py +++ b/src/wikibricks_curator/evidence.py @@ -8,13 +8,10 @@ from pathlib import Path from typing import Any +from wikibricks.curation.backlog import session_targets from wikibricks.storage.sqlite_store import SQLiteStore -def _normalize(value: str) -> str: - return Path(value).name.strip().lower().replace(" ", "-").replace("_", "-") - - def _database_path(conn: sqlite3.Connection) -> Path | None: row = conn.execute("PRAGMA database_list").fetchone() path = str(row[2]) if row else "" @@ -91,7 +88,7 @@ def _moment(value: Any) -> datetime: def _evidence( conn: sqlite3.Connection, *, - project: str, + living_page: str, cursor: str | None, last_page_update: str | None, max_events: int, @@ -100,13 +97,15 @@ def _evidence( ) -> tuple[list[dict[str, Any]], str]: # Compare parsed datetimes: stored ISO strings carry different UTC offsets. after = [_moment(value) for value in (cursor, last_page_update) if value] + targets = session_targets(conn) sessions = [ (_moment(row[2]), row[0]) for row in conn.execute( - "SELECT session_id, workspace, COALESCE(source_updated_at, updated_at) " - "FROM sessions WHERE workspace IS NOT NULL" + "SELECT session_id, workspace, " + "COALESCE(source_updated_at, updated_at) FROM sessions" ).fetchall() - if _normalize(row[1]) == project and all(_moment(row[2]) > bound for bound in after) + if targets.get(row[0]) == living_page + and all(_moment(row[2]) > bound for bound in after) ] if not sessions: return [], "" @@ -166,10 +165,7 @@ def build_request( """Build one bounded curator request for a backlog item.""" project = item["project"] requested_paths = list(dict.fromkeys(item["pages"])) - living_page = next( - (path for path in requested_paths if path.startswith("topics/")), - f"topics/{project}", - ) + living_page = item["living_page"] paths = set(requested_paths) | {living_page} pages = _active_pages( conn, @@ -179,7 +175,7 @@ def build_request( ) evidence, newest_evidence_at = _evidence( conn, - project=project, + living_page=living_page, cursor=cursor, last_page_update=item.get("last_page_update"), max_events=max_events, @@ -192,7 +188,7 @@ def build_request( page_ids = {page["evidence_id"] for page in pages} return { "request": { - "project": project, + "project": item["project"], "living_page": living_page, "current_pages": pages, "evidence": evidence, diff --git a/tests/test_curation_backlog.py b/tests/test_curation_backlog.py index a1ee986..79142a3 100644 --- a/tests/test_curation_backlog.py +++ b/tests/test_curation_backlog.py @@ -23,17 +23,24 @@ def test_curation_backlog_counts_recent_sessions_without_a_page(): from wikibricks.curation.backlog import curation_backlog sessions = [ - ("/Users/u/work/slide-hub", _timestamp(1, reference=NOW)), - ("/Users/u/work/slide-hub", _timestamp(2, reference=NOW)), - ("/Users/u/work/slide-hub", _timestamp(3, reference=NOW)), + ("one", "/Users/u/work/slide-hub", _timestamp(1, reference=NOW)), + ("two", "/Users/u/work/slide-hub", _timestamp(2, reference=NOW)), + ("three", "/Users/u/work/slide-hub", _timestamp(3, reference=NOW)), ] + targets = { + "one": "topics/slide-hub", + "two": "topics/slide-hub", + "three": "topics/slide-hub", + } - result = curation_backlog(sessions, [], now=NOW) + result = curation_backlog(sessions, [], now=NOW, targets=targets) assert result == [ { "project": "slide-hub", + "living_page": "topics/slide-hub", "workspace": "/Users/u/work/slide-hub", + "workspaces": ["/Users/u/work/slide-hub"], "new_sessions": 3, "last_session_at": _timestamp(1, reference=NOW), "pages": [], @@ -46,27 +53,35 @@ def test_curation_backlog_ignores_sessions_covered_by_a_newer_page(): from wikibricks.curation.backlog import curation_backlog sessions = [ - ("/Users/u/work/agent-compliance-cockpit", _timestamp(3, reference=NOW)), - ("/Users/u/work/agent-compliance-cockpit", _timestamp(4, reference=NOW)), - ("/Users/u/work/agent-compliance-cockpit", _timestamp(5, reference=NOW)), + ("one", "/Users/u/work/agent-compliance-cockpit", _timestamp(3, reference=NOW)), + ("two", "/Users/u/work/agent-compliance-cockpit", _timestamp(4, reference=NOW)), + ("three", "/Users/u/work/agent-compliance-cockpit", _timestamp(5, reference=NOW)), ] pages = [ ("topics/agent-atlas", "Agent Atlas (agent-compliance-cockpit)", _timestamp(2, reference=NOW)) ] - assert curation_backlog(sessions, pages, now=NOW) == [] + assert curation_backlog(sessions, pages, now=NOW, targets={ + "one": "topics/agent-atlas", + "two": "topics/agent-atlas", + "three": "topics/agent-atlas", + }) == [] def test_curation_backlog_counts_only_sessions_newer_than_the_page(): from wikibricks.curation.backlog import curation_backlog sessions = [ - ("/Users/u/work/slide-hub", _timestamp(1, reference=NOW)), - ("/Users/u/work/slide-hub", _timestamp(2, reference=NOW)), - ("/Users/u/work/slide-hub", _timestamp(5, reference=NOW)), + ("one", "/Users/u/work/slide-hub", _timestamp(1, reference=NOW)), + ("two", "/Users/u/work/slide-hub", _timestamp(2, reference=NOW)), + ("three", "/Users/u/work/slide-hub", _timestamp(5, reference=NOW)), ] pages = [("topics/slide-hub", "Slide Hub", _timestamp(3, reference=NOW))] - result = curation_backlog(sessions, pages, now=NOW) + result = curation_backlog(sessions, pages, now=NOW, targets={ + "one": "topics/slide-hub", + "two": "topics/slide-hub", + "three": "topics/slide-hub", + }) assert result[0]["new_sessions"] == 2 assert result[0]["pages"] == ["topics/slide-hub"] @@ -77,21 +92,32 @@ def test_curation_backlog_filters_expired_or_invalid_workspaces_and_orders_resul from wikibricks.curation.backlog import curation_backlog sessions = [ - ("/Users/u/work/slide-hub", _timestamp(24 * 8, reference=NOW)), - ("/Users/u/work/slide-hub", _timestamp(1, reference=NOW)), - (str(Path.home()), _timestamp(1, reference=NOW)), - ("", _timestamp(1, reference=NOW)), - (None, _timestamp(1, reference=NOW)), - ("/Users/u/work/zeta", _timestamp(2, reference=NOW)), - ("/Users/u/work/zeta", _timestamp(3, reference=NOW)), + ("old-slide", "/Users/u/work/slide-hub", _timestamp(24 * 8, reference=NOW)), + ("slide", "/Users/u/work/slide-hub", _timestamp(1, reference=NOW)), + ("home", str(Path.home()), _timestamp(1, reference=NOW)), + ("empty", "", _timestamp(1, reference=NOW)), + ("missing", None, _timestamp(1, reference=NOW)), + ("zeta-one", "/Users/u/work/zeta", _timestamp(2, reference=NOW)), + ("zeta-two", "/Users/u/work/zeta", _timestamp(3, reference=NOW)), ] + targets = { + "old-slide": "topics/slide-hub", + "slide": "topics/slide-hub", + "home": None, + "empty": None, + "missing": None, + "zeta-one": "topics/zeta", + "zeta-two": "topics/zeta", + } assert curation_backlog( - sessions, [], now=NOW, days=7, home=Path.home() + sessions, [], now=NOW, days=7, targets=targets ) == [ { "project": "zeta", + "living_page": "topics/zeta", "workspace": "/Users/u/work/zeta", + "workspaces": ["/Users/u/work/zeta"], "new_sessions": 2, "last_session_at": _timestamp(2, reference=NOW), "pages": [], @@ -99,7 +125,9 @@ def test_curation_backlog_filters_expired_or_invalid_workspaces_and_orders_resul }, { "project": "slide-hub", + "living_page": "topics/slide-hub", "workspace": "/Users/u/work/slide-hub", + "workspaces": ["/Users/u/work/slide-hub"], "new_sessions": 1, "last_session_at": _timestamp(1, reference=NOW), "pages": [], diff --git a/tests/test_local_curator.py b/tests/test_local_curator.py index f6d6aac..1d00ca3 100644 --- a/tests/test_local_curator.py +++ b/tests/test_local_curator.py @@ -94,10 +94,28 @@ def _populate(tmp_path: Path) -> tuple[SQLiteStore, dict]: "beta", "/Users/u/work/beta", NEW_SESSION, - [SessionEvent("user-beta", "user", "beta user")], + [SessionEvent("user-beta", "user", "beta user")], ) + with store.connection(write=True) as conn: + session_ids = { + row["external_id"]: row["session_id"] + for row in conn.execute( + "SELECT external_id, session_id FROM sessions WHERE external_id IN " + "('old-alpha', 'middle-alpha', 'new-alpha')" + ).fetchall() + } + conn.executemany( + "INSERT INTO session_topics(session_id, page_path, origin, created_at) " + "VALUES (?, 'topics/alpha', 'manual', ?)", + [ + (session_ids["old-alpha"], OLD_SESSION), + (session_ids["middle-alpha"], MIDDLE_SESSION), + (session_ids["new-alpha"], NEW_SESSION), + ], + ) item = { "project": "alpha-one", + "living_page": "topics/alpha", "workspace": "/Users/u/work/Alpha One", "new_sessions": 2, "last_session_at": NEW_SESSION, @@ -212,6 +230,7 @@ def test_build_request_defaults_living_page_for_a_project_without_coverage(tmp_p conn, { "project": "beta", + "living_page": "topics/beta", "workspace": "/Users/u/work/beta", "new_sessions": 1, "last_session_at": NEW_SESSION, @@ -431,7 +450,12 @@ def test_build_request_compares_timestamps_across_utc_offsets(tmp_path: Path): [SessionEvent("0", "user", "before the page update")]) _ingest(store, "after", "/Users/u/work/offsets", "2026-01-04T11:30:00+00:00", [SessionEvent("0", "user", "after the page update")]) - item = {"project": "offsets", "pages": [], "last_page_update": "2026-01-04T11:00:00+00:00"} + item = { + "project": "offsets", + "living_page": "topics/offsets", + "pages": [], + "last_page_update": "2026-01-04T11:00:00+00:00", + } with store.connection() as conn: built = build_request(conn, item) diff --git a/tests/test_migration_immutability.py b/tests/test_migration_immutability.py index 73a8613..1b2dbcb 100644 --- a/tests/test_migration_immutability.py +++ b/tests/test_migration_immutability.py @@ -20,11 +20,17 @@ "ba5ad411cff8fe7572900dcd1493efca4442e22818e4ef80a4192f46345cff5c" ), f"{LOCAL_MIGRATIONS}/0006_add_link.sql": "90de2f20cb319896227329a4de24f57cdac50f2c752ca1188781952bf4a5d4d8", + f"{LOCAL_MIGRATIONS}/0007_session_topics.sql": ( + "b24e80325ed2b46b2a472b790da84755da3dc0a2feb4f5feb4d836139fb4e6f6" + ), f"{SQLITE_MIGRATIONS}/0001_core.sql": "622f8257f4a8dd9d5db3419ae30f3484f77566094dd416f33f13974d3ca18e8e", f"{SQLITE_MIGRATIONS}/0002_sync.sql": "70e60dd7f65b11df52feb20e1a411cd88485a1d2a45de6daae74cb6781eb17cc", f"{SQLITE_MIGRATIONS}/0003_backfill_late_tables.sql": ( "cb5b62c81aa4c8d7c8aa22e11c558c13305ef16806a258567ea0c98cf4c55fbf" ), + f"{SQLITE_MIGRATIONS}/0004_session_topics.sql": ( + "8fad04bb0f94813aea4817acb82d10f465cc8b35a481121a1c9e9de912f8df6e" + ), f"{REMOTE_MIGRATIONS}/0001_lakebase_search.sql": "b9b2da1ebcf902ed6a10e48dd7bdb4f3d3a9adcb28752bf8dc2e4a1120b17144", } diff --git a/tests/test_postgres_store.py b/tests/test_postgres_store.py index 354341a..2ff0b68 100644 --- a/tests/test_postgres_store.py +++ b/tests/test_postgres_store.py @@ -43,7 +43,7 @@ def test_migrations_are_repeatable_and_create_required_indexes(postgres_url: str assert {"page_search_chunks_vector_idx", "session_search_chunks_vector_idx"} <= indexes assert {"pages_path_trgm_idx", "page_versions_title_trgm_idx"} <= indexes assert {"pages_active_path_idx", "curation_conflicts_pending_idx"} <= indexes - assert migration_count == 6 + assert migration_count == 7 def test_store_exposes_focused_repositories(store: PostgresStore): diff --git a/tests/test_topic_routing.py b/tests/test_topic_routing.py new file mode 100644 index 0000000..8edaad3 --- /dev/null +++ b/tests/test_topic_routing.py @@ -0,0 +1,256 @@ +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pytest + +from wikibricks.config import load_config +from wikibricks.curation.backlog import curation_backlog, session_targets, unrouted_sessions +from wikibricks.models import SessionEvent, SessionRecord +from wikibricks.storage.sqlite_store import SQLiteStore +from wikibricks_curator.evidence import build_request + +NOW = datetime(2026, 9, 29, 12, 0, tzinfo=timezone.utc) +HOME = Path("/Users/u") + + +def _timestamp(offset_hours: int) -> str: + return (NOW - timedelta(hours=offset_hours)).isoformat() + + +def _store(tmp_path: Path) -> SQLiteStore: + store = SQLiteStore(tmp_path / "wikibricks.db") + store.migrate() + return store + + +def _ingest( + store: SQLiteStore, + external_id: str, + workspace: str, + updated_at: str = _timestamp(1), +) -> str: + store.ingest_session( + SessionRecord( + harness="test-harness", + external_id=external_id, + user_id="user", + workspace=workspace, + updated_at=updated_at, + events=[SessionEvent("0", "user", f"{external_id} content")], + ) + ) + with store.connection(write=True) as conn: + return conn.execute( + "SELECT session_id FROM sessions " + "WHERE harness = 'test-harness' AND external_id = ?", + (external_id,), + ).fetchone()[0] + + +def test_session_targets_route_by_rows_and_pages(tmp_path: Path): + store = _store(tmp_path) + store.write_page( + "topics/agent-atlas", + "Agent Atlas (agent-compliance-cockpit)", + {"summary": "atlas", "body": "page"}, + ) + routed = _ingest(store, "routed", "/Users/u/work/covered") + routed_none = _ingest(store, "routed-none", "/Users/u/work/covered") + generic = _ingest(store, "generic", str(HOME / "code")) + home = _ingest(store, "home", str(HOME)) + covered = _ingest(store, "covered", str(HOME / "agent-compliance-cockpit")) + uncovered = _ingest(store, "uncovered", str(HOME / "uncovered-project")) + + with store.connection(write=True) as conn: + conn.executemany( + "INSERT INTO session_topics(session_id, page_path, origin, created_at) " + "VALUES (?, ?, 'manual', ?)", + [ + (routed, "topics/agent-atlas", _timestamp(2)), + (routed_none, None, _timestamp(2)), + ], + ) + targets = session_targets(conn, home=HOME) + + assert targets[routed] == "topics/agent-atlas" + assert targets[routed_none] is None + assert targets[generic] is None + assert targets[home] is None + assert targets[covered] == "topics/agent-atlas" + assert targets[uncovered] == "topics/uncovered-project" + + +def test_session_targets_honours_a_generic_override(tmp_path: Path): + store = _store(tmp_path) + overridden = _ingest(store, "overridden", str(HOME / "custom")) + project = _ingest(store, "project", str(HOME / "custom" / "project")) + + with store.connection(write=True) as conn: + targets = session_targets(conn, generic=("custom",), home=HOME) + + assert targets[overridden] is None + assert targets[project] == "topics/project" + + +def test_unrouted_sessions_returns_only_recent_container_folder_sessions(tmp_path: Path): + store = _store(tmp_path) + generic = _ingest(store, "generic", str(HOME / "code"), _timestamp(1)) + home = _ingest(store, "home", str(HOME), _timestamp(2)) + old = _ingest(store, "old", str(HOME / "code"), _timestamp(10)) + routed = _ingest(store, "routed", str(HOME / "code"), _timestamp(3)) + project = _ingest(store, "project", str(HOME / "project"), _timestamp(1)) + with store.connection(write=True) as conn: + conn.execute( + "INSERT INTO session_topics(session_id, page_path, origin, created_at) " + "VALUES (?, 'topics/agent-atlas', 'manual', ?)", + (routed, _timestamp(4)), + ) + + with store.connection() as conn: + result = unrouted_sessions(conn, since=_timestamp(5), home=HOME) + + assert set(result) == {generic, home} + assert old not in result + assert routed not in result + assert project not in result + + +def test_curation_backlog_groups_by_routed_pages(tmp_path: Path): + store = _store(tmp_path) + store.write_page( + "topics/agent-atlas", + "Agent Atlas (agent-compliance-cockpit)", + {"summary": "atlas", "body": "page"}, + ) + with store.connection(write=True) as conn: + conn.execute( + "UPDATE pages SET updated_at = ? WHERE path = 'topics/agent-atlas'", + (_timestamp(3),), + ) + project_one = _ingest(store, "one", str(HOME / "agent-compliance-cockpit")) + project_two = _ingest(store, "two", str(HOME / "agent-compliance-cockpit")) + routed = _ingest(store, "routed", str(HOME / "code")) + unrouted = _ingest(store, "unrouted", str(HOME / "code")) + routed_none = _ingest(store, "none", str(HOME / "agent-compliance-cockpit")) + with store.connection(write=True) as conn: + conn.executemany( + "INSERT INTO session_topics(session_id, page_path, origin, created_at) " + "VALUES (?, ?, 'manual', ?)", + [ + (routed, "topics/agent-atlas", _timestamp(2)), + (routed_none, None, _timestamp(2)), + ], + ) + targets = session_targets(conn, home=HOME) + + sessions = [ + (project_one, str(HOME / "agent-compliance-cockpit"), _timestamp(1)), + (project_two, str(HOME / "agent-compliance-cockpit"), _timestamp(1)), + (routed, str(HOME / "code"), _timestamp(1)), + (unrouted, str(HOME / "code"), _timestamp(1)), + (routed_none, str(HOME / "agent-compliance-cockpit"), _timestamp(1)), + ] + pages = [ + ( + "topics/agent-atlas", + "Agent Atlas (agent-compliance-cockpit)", + _timestamp(3), + ) + ] + result = curation_backlog(sessions, pages, now=NOW, targets=targets) + + assert targets[unrouted] is None + assert len(result) == 1 + item = result[0] + assert item["project"] == "agent-atlas" + assert item["living_page"] == "topics/agent-atlas" + assert item["workspaces"] == [ + str(HOME / "agent-compliance-cockpit"), + str(HOME / "code"), + ] + assert item["workspace"] == str(HOME / "agent-compliance-cockpit") + assert item["new_sessions"] == 3 + assert item["pages"] == ["topics/agent-atlas"] + + +def test_build_request_selects_routed_sessions(tmp_path: Path): + store = _store(tmp_path) + store.write_page("topics/agent-atlas", "Agent Atlas", {"summary": "a", "body": "b"}) + routed = _ingest(store, "routed", str(HOME / "code")) + unrouted = _ingest(store, "unrouted", str(HOME / "code")) + with store.connection(write=True) as conn: + conn.execute( + "INSERT INTO session_topics(session_id, page_path, origin, created_at) " + "VALUES (?, 'topics/agent-atlas', 'manual', ?)", + (routed, _timestamp(2)), + ) + item = { + "project": "agent-atlas", + "living_page": "topics/agent-atlas", + "new_sessions": 1, + "last_session_at": _timestamp(1), + "pages": ["topics/agent-atlas"], + "last_page_update": None, + } + + with store.connection(write=True) as conn: + result = build_request(conn, item, related=0) + + texts = [entry["text"] for entry in result["request"]["evidence"]] + assert texts == ["routed content"] + assert result["request"]["living_page"] == "topics/agent-atlas" + with store.connection() as conn: + assert session_targets(conn, home=HOME)[unrouted] is None + + +def test_config_defaults_environment_and_validation(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + defaults = load_config(home=tmp_path / "home", environ={}) + assert defaults.curation_generic_workspaces == ( + "code", + "emails", + "work", + "projects", + "repos", + "src", + "documents", + "desktop", + "downloads", + "tmp", + ) + + monkeypatch.setenv("WIKIBRICKS_CURATION_GENERIC_WORKSPACES", "code, Custom Workspace") + overridden = load_config(home=tmp_path / "home") + assert overridden.curation_generic_workspaces == ("code", "Custom Workspace") + + with pytest.raises(ValueError, match="WIKIBRICKS_CURATION_GENERIC_WORKSPACES"): + load_config( + home=tmp_path / "home", + environ={"WIKIBRICKS_CURATION_GENERIC_WORKSPACES": "code,"}, + ) + + +def test_migrations_apply_and_delete_session_topics_with_sessions(tmp_path: Path): + store = _store(tmp_path) + session_id = _ingest(store, "routed", str(HOME / "code")) + with store.connection(write=True) as conn: + conn.execute( + "INSERT INTO session_topics(session_id, page_path, origin, created_at) " + "VALUES (?, 'topics/agent-atlas', 'manual', ?)", + (session_id, _timestamp(2)), + ) + count = conn.execute("SELECT count(*) FROM session_topics").fetchone()[0] + store.migrate() + with store.connection(write=True) as conn: + conn.execute("DROP TABLE session_topics") + conn.execute( + "DELETE FROM schema_migrations WHERE name = '0004_session_topics.sql'" + ) + store.migrate() + with store.connection() as conn: + assert count == 1 + conn.execute("DELETE FROM sessions WHERE session_id = ?", (session_id,)) + remaining = conn.execute("SELECT count(*) FROM session_topics").fetchone()[0] + + assert remaining == 0 From 1b5209f689415ee2516ebb70491114f5856ca499 Mon Sep 17 00:00:00 2001 From: Philipp Date: Tue, 29 Sep 2026 11:14:34 +0200 Subject: [PATCH 2/2] feat(curator): route container-folder sessions to their topic Before building the backlog, the nightly curator sends unrouted sessions from container folders (home, code, emails, ...) and the existing topic pages to the model in one call. Each session goes to an existing page, a new `topics/` named after the subject, or null for throwaway work; the answer is stored in `session_topics`, so each session is routed once. - Routes are validated: known session, candidate or well-formed new path, never a container-folder name. Invalid routes are retried next night. - Pages named after a container folder (an old `topics/emails`) are not offered as candidates. - A session routed after its page's last update counts as new evidence for that page, so its content reaches the page. - One retry; a routing failure is reported and never blocks curation. Acceptance with GLM 5.3 Flash on a copy of the live store: 6 of 6 sessions routed by content (e.g. a Slide Hub MCP session from the emails folder to `topics/slide-hub`, two interview scripts to `topics/interviewing`), 0 invalid, then 5 of 5 pages applied. Implemented by GLM 5.3 Flash; candidate filtering and the null rule tightened in review after the first real run discarded 4 of 6 sessions. Co-authored-by: Isaac --- CHANGELOG.md | 2 + README.md | 4 + src/wikibricks/curation/backlog.py | 13 +- src/wikibricks_curator/curator.py | 10 +- src/wikibricks_curator/evidence.py | 17 +- src/wikibricks_curator/resources/router.md | 17 + .../resources/router.schema.json | 18 ++ src/wikibricks_curator/router.py | 190 +++++++++++ tests/test_topic_routing.py | 302 +++++++++++++++++- 9 files changed, 563 insertions(+), 10 deletions(-) create mode 100644 src/wikibricks_curator/resources/router.md create mode 100644 src/wikibricks_curator/resources/router.schema.json create mode 100644 src/wikibricks_curator/router.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d994b7d..41ac86a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,8 @@ ## Unreleased - Group curation sessions by routed topic pages instead of workspace folder names. + The nightly curator asks the model which topic each container-folder session is + about before building the backlog. ## 0.13.0 - 2026-09-29 diff --git a/README.md b/README.md index d7d42a8..e890d88 100644 --- a/README.md +++ b/README.md @@ -227,6 +227,10 @@ wikibricks curate --prune-archived-sessions-after-days 90 ## Nightly curation (optional) +Sessions captured under container folders such as `~/code` do not identify their +topic. Before building the backlog, the curator asks the model to route those +sessions to an existing or new page, subject to `curation.generic_workspaces`. + `wikibricks-curator propose` asks one Databricks model for updates to the top local backlog projects. It writes those proposals as a curation run, applies low-risk groups with the safe policy, and leaves high-risk or conflicting diff --git a/src/wikibricks/curation/backlog.py b/src/wikibricks/curation/backlog.py index 1d62e3f..7dbd3fb 100644 --- a/src/wikibricks/curation/backlog.py +++ b/src/wikibricks/curation/backlog.py @@ -204,9 +204,18 @@ def load_curation_backlog(conn: Any, **kwargs: Any) -> list[dict[str, Any]]: """Load sessions and active pages, then calculate the curation backlog.""" # Count sessions by when the work happened, not when a backfill imported them. sessions = conn.execute( - "SELECT s.session_id, s.workspace, " - "COALESCE(s.source_updated_at, s.updated_at) FROM sessions s" + "SELECT s.session_id, s.workspace, COALESCE(s.source_updated_at, s.updated_at), " + "t.created_at " + "FROM sessions s LEFT JOIN session_topics t ON t.session_id = s.session_id" ).fetchall() + sessions = [ + ( + row[0], + row[1], + _timestamp(row[2]) if row[3] is None else max(_timestamp(row[2]), _timestamp(row[3])), + ) + for row in sessions + ] pages = conn.execute( "SELECT p.path, v.title, p.updated_at FROM pages p " "JOIN page_versions v ON v.version_id = p.current_version_id " diff --git a/src/wikibricks_curator/curator.py b/src/wikibricks_curator/curator.py index 1dc7922..e27f1d1 100644 --- a/src/wikibricks_curator/curator.py +++ b/src/wikibricks_curator/curator.py @@ -4,6 +4,7 @@ import hashlib from dataclasses import replace +from datetime import datetime, timedelta, timezone from importlib.resources import files from pathlib import Path from typing import Any, Callable @@ -19,6 +20,7 @@ from wikibricks.curation.backlog import load_curation_backlog from wikibricks.storage.sqlite_store import SQLiteStore from wikibricks_curator.evidence import build_request +from wikibricks_curator.router import route_sessions from wikibricks_remote.proposals import _PROPOSAL_FIELDS, build_patches from wikibricks_remote.resources import load_policy, load_prompt, load_schema @@ -134,6 +136,12 @@ def run_curator( ) schema = load_schema() policy = replace(load_policy(), allowed_operations=("create_page", "update_page", "add_link")) + routing = route_sessions( + store, + chat, + since=datetime.now(timezone.utc) - timedelta(days=7), + dry_run=dry_run, + ) with store.connection() as conn: backlog = load_curation_backlog(conn, limit=projects) replica_id = get_or_create_replica_id(store) @@ -230,4 +238,4 @@ def run_curator( ) results.append(result) - return {"projects": results, "errors": errors} + return {"projects": results, "errors": errors, "routing": routing} diff --git a/src/wikibricks_curator/evidence.py b/src/wikibricks_curator/evidence.py index f407d05..e0635eb 100644 --- a/src/wikibricks_curator/evidence.py +++ b/src/wikibricks_curator/evidence.py @@ -98,14 +98,19 @@ def _evidence( # Compare parsed datetimes: stored ISO strings carry different UTC offsets. after = [_moment(value) for value in (cursor, last_page_update) if value] targets = session_targets(conn) + session_rows = conn.execute( + "SELECT s.session_id, s.workspace, COALESCE(s.source_updated_at, s.updated_at), " + "t.created_at " + "FROM sessions s LEFT JOIN session_topics t ON t.session_id = s.session_id" + ).fetchall() sessions = [ - (_moment(row[2]), row[0]) - for row in conn.execute( - "SELECT session_id, workspace, " - "COALESCE(source_updated_at, updated_at) FROM sessions" - ).fetchall() + ( + _moment(row[2]) if row[3] is None else max(_moment(row[2]), _moment(row[3])), + row[0], + ) + for row in session_rows if targets.get(row[0]) == living_page - and all(_moment(row[2]) > bound for bound in after) + and all(_moment(row[2] if row[3] is None else max(_moment(row[2]), _moment(row[3]))) > bound for bound in after) ] if not sessions: return [], "" diff --git a/src/wikibricks_curator/resources/router.md b/src/wikibricks_curator/resources/router.md new file mode 100644 index 0000000..d99b598 --- /dev/null +++ b/src/wikibricks_curator/resources/router.md @@ -0,0 +1,17 @@ +# Route sessions to wiki topics + +Each session in `sessions` ran in a container folder such as the home directory, `code`, +or `emails`. The folder says nothing about the subject. Assign each session to the wiki +page it is about. + +For each session return one route with `session_id`, `page_path`, and a short `reason`: + +1. Use a `candidates` path when the session's main subject is that page's topic. +2. Use a new path `topics/` when the session is about a durable subject that no + candidate covers. The slug names the subject (a customer, product, project, or + concept) in lowercase letters, digits, and hyphens. Never name it after the folder. +3. Use `null` only when the session holds nothing worth remembering: a one-off factual + question, a test of a tool, or small talk. A work product (script, guide, deck, email, + analysis) or a solved problem is durable: route it with rule 1 or 2. + +Return `{"routes": [...]}` with one entry per session. diff --git a/src/wikibricks_curator/resources/router.schema.json b/src/wikibricks_curator/resources/router.schema.json new file mode 100644 index 0000000..ad393e5 --- /dev/null +++ b/src/wikibricks_curator/resources/router.schema.json @@ -0,0 +1,18 @@ +{ + "type": "object", + "properties": { + "routes": { + "type": "array", + "items": { + "type": "object", + "properties": { + "session_id": {"type": "string"}, + "page_path": {"type": ["string", "null"]}, + "reason": {"type": "string"} + }, + "required": ["session_id", "page_path", "reason"] + } + } + }, + "required": ["routes"] +} diff --git a/src/wikibricks_curator/router.py b/src/wikibricks_curator/router.py new file mode 100644 index 0000000..c5b013e --- /dev/null +++ b/src/wikibricks_curator/router.py @@ -0,0 +1,190 @@ +"""Route container-folder sessions to topic pages with one model call.""" + +from __future__ import annotations + +import json +import re +from collections.abc import Callable +from datetime import datetime, timezone +from importlib.resources import files +from typing import Any + +from wikibricks.config import load_config +from wikibricks.curation.backlog import unrouted_sessions +from wikibricks.storage.sqlite_store import SQLiteStore + +Chat = Callable[[str, dict[str, Any], dict[str, Any]], dict[str, Any]] +_NEW_TOPIC = re.compile(r"^topics/[a-z0-9]+(-[a-z0-9]+)*$") +_ROUTER_PROMPT = files("wikibricks_curator").joinpath("resources", "router.md") +_ROUTER_SCHEMA = files("wikibricks_curator").joinpath("resources", "router.schema.json") + + +def _normalize(value: str) -> str: + return value.strip().lower().replace(" ", "-").replace("_", "-") + + +def _sessions(conn: Any, ids: list[str], max_chars: int) -> list[dict[str, str]]: + if not ids: + return [] + placeholders = ",".join("?" for _ in ids) + rows = conn.execute( + "SELECT s.session_id, s.workspace, s.title FROM sessions s " + f"WHERE s.session_id IN ({placeholders})", + ids, + ).fetchall() + event_rows = conn.execute( + "SELECT e.session_id, v.content FROM session_events e " + "JOIN session_event_versions v ON v.version_id = e.current_version_id " + "WHERE e.active = 1 AND v.kind = 'user' " + f"AND e.session_id IN ({placeholders}) ORDER BY e.session_id, e.position", + ids, + ).fetchall() + texts: dict[str, list[str]] = {} + for session_id, content in event_rows: + texts.setdefault(session_id, []).append(content) + return [ + { + "session_id": row["session_id"], + "title": row["title"], + "workspace": row["workspace"], + "text": "\n".join(texts.get(row["session_id"], []))[:max_chars], + } + for row in sorted(rows, key=lambda row: ids.index(row["session_id"])) + ] + + +def _candidates(conn: Any, generic: set[str]) -> list[dict[str, str]]: + # Pages named after a container folder (e.g. an old `topics/emails`) are not topics. + rows = conn.execute( + "SELECT p.path, v.title, v.content FROM pages p " + "JOIN page_versions v ON v.version_id = p.current_version_id " + "WHERE p.status = 'active' " + "AND (p.path LIKE 'topics/%' OR p.path LIKE 'projects/%') ORDER BY p.path" + ).fetchall() + return [ + { + "path": row["path"], + "title": row["title"], + "summary": str(json.loads(row["content"]).get("summary", ""))[:200], + } + for row in rows + if _normalize(row["path"].rsplit("/", 1)[-1]) not in generic | {"home"} + ] + + +def _new_path(path: str, *, candidates: set[str], generic: set[str]) -> bool: + if not isinstance(path, str) or path in candidates: + return False + last_segment = path.rsplit("/", 1)[-1] + return _NEW_TOPIC.fullmatch(path) is not None and _normalize(last_segment) not in generic | {"home"} + + +def _valid_path(path: Any, *, candidates: set[str], generic: set[str]) -> bool: + if not isinstance(path, str): + return False + return path in candidates or _new_path(path, candidates=candidates, generic=generic) + + +def _schema() -> dict[str, Any]: + return json.loads(_ROUTER_SCHEMA.read_text(encoding="utf-8")) + + +def route_sessions( + store: SQLiteStore, + chat: Chat, + *, + since: str | datetime, + limit: int = 20, + max_chars: int = 600, + dry_run: bool = False, +) -> dict[str, Any]: + """Ask the model for topic routes and store the accepted answers.""" + generic = {_normalize(value) for value in load_config().curation_generic_workspaces} + with store.connection() as conn: + ids = unrouted_sessions(conn, since=since)[:limit] + if not ids: + return { + "sessions": 0, + "routed": 0, + "to_none": 0, + "new_topics": [], + "invalid": 0, + "error": None, + } + sessions = _sessions(conn, ids, max_chars) + candidates = _candidates(conn, generic) + candidate_paths = {candidate["path"] for candidate in candidates} + prompt = _ROUTER_PROMPT.read_text(encoding="utf-8") + schema = _schema() + empty = {session["session_id"] for session in sessions if not session["text"]} + model_ids = [session_id for session_id in ids if session_id not in empty] + model_id_set = set(model_ids) + request = { + "sessions": [session for session in sessions if session["session_id"] in model_id_set], + "candidates": candidates, + } + raw: Any = None + error: str | None = None + for _attempt in (1, 2): + try: + raw = chat(prompt, request, schema) if model_ids else {"routes": []} + error = None + break + except Exception as exc: + error = str(exc)[:300] + if error is not None: + return { + "sessions": len(ids), + "routed": 0, + "to_none": 0, + "new_topics": [], + "invalid": 0, + "error": error, + } + + routes = raw.get("routes") if isinstance(raw, dict) else None + invalid_routes = not isinstance(routes, list) + routes = routes if isinstance(routes, list) else [] + accepted: dict[str, str | None] = {} + invalid = 0 + for route in routes: + session_id = route.get("session_id") if isinstance(route, dict) else None + page_path = route.get("page_path") if isinstance(route, dict) else None + if ( + not isinstance(session_id, str) + or session_id not in model_id_set + or session_id in accepted + or not ( + page_path is None + or _valid_path(page_path, candidates=candidate_paths, generic=generic) + ) + ): + invalid += 1 + continue + accepted[session_id] = page_path + for session_id in empty: + accepted[session_id] = None + if invalid_routes: + invalid += len(model_ids) + new_topics = sorted( + {path for path in accepted.values() if path is not None and path not in candidate_paths} + ) + if not dry_run: + timestamp = datetime.now(timezone.utc).isoformat() + with store.connection(write=True) as conn: + conn.executemany( + "INSERT OR IGNORE INTO session_topics" + "(session_id, page_path, origin, created_at) VALUES (?, ?, 'local-curator-router', ?)", + [ + (session_id, page_path, timestamp) + for session_id, page_path in accepted.items() + ], + ) + return { + "sessions": len(ids), + "routed": len(accepted), + "to_none": sum(path is None for path in accepted.values()), + "new_topics": new_topics, + "invalid": invalid, + "error": None, + } diff --git a/tests/test_topic_routing.py b/tests/test_topic_routing.py index 8edaad3..3bf325d 100644 --- a/tests/test_topic_routing.py +++ b/tests/test_topic_routing.py @@ -2,14 +2,22 @@ from datetime import datetime, timedelta, timezone from pathlib import Path +from typing import Any import pytest from wikibricks.config import load_config -from wikibricks.curation.backlog import curation_backlog, session_targets, unrouted_sessions +from wikibricks.curation.backlog import ( + curation_backlog, + load_curation_backlog, + session_targets, + unrouted_sessions, +) from wikibricks.models import SessionEvent, SessionRecord from wikibricks.storage.sqlite_store import SQLiteStore +from wikibricks_curator.curator import run_curator from wikibricks_curator.evidence import build_request +from wikibricks_curator.router import route_sessions NOW = datetime(2026, 9, 29, 12, 0, tzinfo=timezone.utc) HOME = Path("/Users/u") @@ -254,3 +262,295 @@ def test_migrations_apply_and_delete_session_topics_with_sessions(tmp_path: Path remaining = conn.execute("SELECT count(*) FROM session_topics").fetchone()[0] assert remaining == 0 + + +def _topic_routes(store: SQLiteStore, external_ids: tuple[str, ...]) -> list[str]: + return [_ingest(store, external_id, str(HOME / "code")) for external_id in external_ids] + + +def test_route_sessions_writes_model_routes_and_builds_backlog(tmp_path: Path): + store = _store(tmp_path) + store.write_page( + "topics/agent-atlas", + "Agent Atlas", + {"summary": "Atlas summary", "body": "Atlas body"}, + ) + store.write_page("projects/unity", "Unity", {"summary": "Unity", "body": "project"}) + with store.connection(write=True) as conn: + conn.execute( + "UPDATE pages SET updated_at = ? WHERE path = 'topics/agent-atlas'", + (_timestamp(4),), + ) + atlas, unity, quick = _topic_routes(store, ("atlas", "unity", "quick")) + calls: list[tuple[str, dict[str, Any], dict[str, Any]]] = [] + + def chat(prompt: str, request: dict[str, Any], schema: dict[str, Any]) -> dict[str, Any]: + calls.append((prompt, request, schema)) + return { + "routes": [ + {"session_id": atlas, "page_path": "topics/agent-atlas", "reason": "atlas"}, + {"session_id": unity, "page_path": "topics/unity-gateway", "reason": "new"}, + {"session_id": quick, "page_path": None, "reason": "quick"}, + ] + } + + result = route_sessions(store, chat, since=_timestamp(3)) + + assert result == { + "sessions": 3, + "routed": 3, + "to_none": 1, + "new_topics": ["topics/unity-gateway"], + "invalid": 0, + "error": None, + } + assert calls[0][0].startswith("# Route sessions to wiki topics") + assert [item["path"] for item in calls[0][1]["candidates"]] == [ + "projects/unity", + "topics/agent-atlas", + ] + assert sorted(item["session_id"] for item in calls[0][1]["sessions"]) == sorted([atlas, unity, quick]) + assert calls[0][2]["properties"]["routes"]["items"]["properties"]["page_path"]["type"] == [ + "string", + "null", + ] + with store.connection() as conn: + rows = { + row[0]: (row[1], row[2]) + for row in conn.execute( + "SELECT session_id, page_path, origin FROM session_topics" + ) + } + targets = session_targets(conn, home=HOME) + assert rows == { + atlas: ("topics/agent-atlas", "local-curator-router"), + unity: ("topics/unity-gateway", "local-curator-router"), + quick: (None, "local-curator-router"), + } + assert targets == { + atlas: "topics/agent-atlas", + unity: "topics/unity-gateway", + quick: None, + } + with store.connection() as conn: + backlog = load_curation_backlog(conn, home=HOME, now=NOW) + assert {item["living_page"] for item in backlog} == { + "topics/unity-gateway", + "topics/agent-atlas", + } + assert all(item["new_sessions"] == 1 for item in backlog) + + +def test_route_sessions_skips_invalid_routes(tmp_path: Path): + store = _store(tmp_path) + store.write_page("topics/valid", "Valid", {"summary": "s", "body": "b"}) + (folder, email, home, project, bad_slug) = _topic_routes( + store, + ("folder", "email", "home", "project", "bad-slug"), + ) + + def chat(_prompt: str, _request: dict[str, Any], _schema: dict[str, Any]) -> dict[str, Any]: + return { + "routes": [ + {"session_id": "missing", "page_path": "topics/valid", "reason": "unknown"}, + {"session_id": folder, "page_path": "topics/code", "reason": "folder"}, + {"session_id": email, "page_path": "topics/emails", "reason": "folder"}, + {"session_id": home, "page_path": "topics/home", "reason": "folder"}, + {"session_id": project, "page_path": "projects/x", "reason": "not candidate"}, + {"session_id": bad_slug, "page_path": "Topics/Bad Slug", "reason": "bad"}, + ] + } + + with store.connection() as conn: + before = set(unrouted_sessions(conn, since=_timestamp(2), home=HOME)) + result = route_sessions(store, chat, since=_timestamp(2)) + with store.connection() as conn: + after = set(unrouted_sessions(conn, since=_timestamp(2), home=HOME)) + + assert result["sessions"] == 5 + assert result["routed"] == 0 + assert result["to_none"] == 0 + assert result["new_topics"] == [] + assert result["invalid"] == 6 + assert result["error"] is None + assert before == after + + +def test_route_sessions_routes_empty_session_without_chat(tmp_path: Path): + store = _store(tmp_path) + store.ingest_session( + SessionRecord( + harness="test-harness", + external_id="empty", + user_id="user", + workspace=str(HOME / "code"), + updated_at=_timestamp(1), + events=[], + ) + ) + with store.connection(write=True) as conn: + session_id = conn.execute("SELECT session_id FROM sessions").fetchone()[0] + + def chat(*_args: Any) -> dict[str, Any]: + raise AssertionError("chat should not be called") + + result = route_sessions(store, chat, since=_timestamp(2)) + + assert result == { + "sessions": 1, + "routed": 1, + "to_none": 1, + "new_topics": [], + "invalid": 0, + "error": None, + } + with store.connection() as conn: + row = conn.execute( + "SELECT page_path, origin FROM session_topics WHERE session_id = ?", + (session_id,), + ).fetchone() + assert tuple(row) == (None, "local-curator-router") + + +def test_route_sessions_retries_once_and_reports_failure(tmp_path: Path): + store = _store(tmp_path) + _topic_routes(store, ("one", "two")) + attempts = 0 + + def succeeding(_prompt: str, _request: dict[str, Any], _schema: dict[str, Any]) -> dict[str, Any]: + nonlocal attempts + attempts += 1 + assert attempts == 2 + return {"routes": []} + + result = route_sessions(store, succeeding, since=_timestamp(2)) + assert result["error"] is None + + def failing(_prompt: str, _request: dict[str, Any], _schema: dict[str, Any]) -> dict[str, Any]: + raise RuntimeError("model unavailable") + + failure = route_sessions(store, failing, since=_timestamp(2)) + assert failure["error"] == "model unavailable" + with store.connection() as conn: + assert conn.execute( + "SELECT count(*) FROM session_topics WHERE origin = 'local-curator-router'" + ).fetchone()[0] == 0 + + +def test_route_sessions_dry_run_writes_nothing(tmp_path: Path): + store = _store(tmp_path) + (session_id,) = _topic_routes(store, ("dry",)) + + def chat(_prompt: str, _request: dict[str, Any], _schema: dict[str, Any]) -> dict[str, Any]: + return {"routes": [{"session_id": session_id, "page_path": "topics/dry", "reason": "ok"}]} + + result = route_sessions(store, chat, since=_timestamp(2), dry_run=True) + + assert result["routed"] == 1 + assert result["new_topics"] == ["topics/dry"] + with store.connection() as conn: + assert conn.execute("SELECT count(*) FROM session_topics").fetchone()[0] == 0 + + +def test_routed_time_counts_as_new_evidence(tmp_path: Path): + store = _store(tmp_path) + store.write_page("topics/agent-atlas", "Agent Atlas", {"summary": "a", "body": "b"}) + with store.connection(write=True) as conn: + conn.execute( + "UPDATE pages SET updated_at = ? WHERE path = 'topics/agent-atlas'", + (NOW.isoformat(),), + ) + session_id = _ingest( + store, + "routed-at", + str(HOME / "code"), + (NOW - timedelta(hours=1)).isoformat(), + ) + with store.connection(write=True) as conn: + conn.execute( + "INSERT INTO session_topics(session_id, page_path, origin, created_at) " + "VALUES (?, 'topics/agent-atlas', 'local-curator-router', ?)", + (session_id, (NOW + timedelta(hours=1)).isoformat()), + ) + targets = session_targets(conn, home=HOME) + assert targets[session_id] == "topics/agent-atlas" + with store.connection() as conn: + result = load_curation_backlog( + conn, + now=NOW + timedelta(hours=2), + home=HOME, + ) + assert result[0]["new_sessions"] == 1 + + item = { + "project": "agent-atlas", + "living_page": "topics/agent-atlas", + "new_sessions": 1, + "last_session_at": (NOW + timedelta(hours=1)).isoformat(), + "pages": ["topics/agent-atlas"], + "last_page_update": NOW.isoformat(), + } + with store.connection(write=True) as conn: + built = build_request(conn, item) + assert built is not None + assert built["request"]["evidence"][0]["text"] == "routed-at content" + assert built["newest_evidence_at"] == (NOW + timedelta(hours=1)).isoformat() + + +def test_run_curator_routes_before_backlog_and_hides_routing_errors(tmp_path: Path): + store = _store(tmp_path) + store.write_page("topics/agent-atlas", "Agent Atlas", {"summary": "a", "body": "b"}) + with store.connection(write=True) as conn: + conn.execute( + "UPDATE pages SET updated_at = ? WHERE path = 'topics/agent-atlas'", + (_timestamp(5),), + ) + routed = _ingest(store, "routed", str(HOME / "code"), _timestamp(4)) + _ingest(store, "router", str(HOME / "code")) + with store.connection(write=True) as conn: + conn.execute( + "INSERT INTO session_topics(session_id, page_path, origin, created_at) " + "VALUES (?, 'topics/agent-atlas', 'manual', ?)", + (routed, _timestamp(3)), + ) + calls = 0 + + def chat(prompt: str, request: dict[str, Any], schema: dict[str, Any]) -> dict[str, Any]: + nonlocal calls + calls += 1 + if prompt.startswith("# Route sessions"): + raise RuntimeError("router temporarily unavailable") + return {"proposals": []} + + result = run_curator(tmp_path / "wikibricks.db", chat=chat) + + assert calls == 3 + assert result["errors"] == 0 + assert result["routing"]["error"] == "router temporarily unavailable" + assert result["routing"]["sessions"] == 1 + assert result["routing"]["invalid"] == 0 + assert result["projects"][0]["status"] == "no_changes" + with store.connection() as conn: + assert conn.execute( + "SELECT count(*) FROM session_topics WHERE origin = 'local-curator-router'" + ).fetchone()[0] == 0 + + +def test_pages_named_after_a_container_folder_are_not_candidates(tmp_path: Path): + # The live store already has a `topics/emails` page from before routing existed. + store = _store(tmp_path) + store.write_page("topics/emails", "Emails", {"summary": "mixed", "body": "Unrelated work."}) + store.write_page("topics/agent-atlas", "Agent Atlas", {"summary": "s", "body": "Page."}) + email = _ingest(store, "email-session", "/Users/u/work/50-communications/emails", _timestamp(1)) + seen: dict[str, Any] = {} + + def chat(_prompt: str, request: dict[str, Any], _schema: dict[str, Any]) -> dict[str, Any]: + seen["candidates"] = [candidate["path"] for candidate in request["candidates"]] + return {"routes": [{"session_id": email, "page_path": "topics/emails", "reason": "folder"}]} + + result = route_sessions(store, chat, since=NOW - timedelta(days=7)) + + assert seen["candidates"] == ["topics/agent-atlas"] + assert result["invalid"] == 1 + with store.connection() as conn: + assert email in unrouted_sessions(conn, since=NOW - timedelta(days=7))