diff --git a/CHANGELOG.md b/CHANGELOG.md index 7801f46..eda4cf3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,34 @@ commits and tag annotations of those releases. --- +## [Unreleased] + +### Added + +- **The Codex finalizer.** Codex threads now get the full pipeline — + embeddings, episodes, project inference — on their own, 30 minutes after + the rollout goes quiet. It is the Codex twin of Claude Code's SessionEnd + hook: Codex sends no session-end signal, so quiet stands in for it. Until + now an active thread that had been indexed with `--semantic` regressed to + exact-record-only the moment it grew, and nothing ever indexed it again. + `longhand codex-sync` runs the finalizer after its exact-record pass + (`--finalize-after SECONDS` to tune the bound, `--no-finalize` to opt out), + one thread per run so the every-minute poller stays bounded; `reconcile + --fix` clears the whole quiet backlog. The embedding model loads only on a + run that has a quiet thread to finalize. A finalized thread that resumes is + captured exact-only again and finalized again once it settles — one + re-embed per resume, never a treadmill. +- `longhand.codex.finalize_codex()`, a `settling` bucket and + `min_idle_seconds` on `scan_codex_sessions()` / `sync_codex()`, and + `codex_unfinalized` / `codex_settling` / `codex_finalized` on the + `reconcile` report and MCP tool. + +### Changed + +- `doctor`'s "Codex sessions archived" row is now "Codex finalizer": green + while archived threads are still being written, yellow only when one has + been quiet past the bound and nothing has finalized it yet. + ## [1.1.0] — 2026-09-07 Longhand now keeps one archive for Claude Code **and** Codex. This is the diff --git a/CLAUDE.md b/CLAUDE.md index 58e8d19..7930fd8 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -55,4 +55,4 @@ Beyond the decision tree above: `get_session_timeline` with `tail` (the last N e ## Codex sessions (1.1.0+) -Codex Desktop / CLI threads live in the same archive with `codex:`-prefixed session ids. This server lists and pages them (`list_sessions`, `get_session_timeline`) and `find_commits` sees commits made from Codex — but `recall` and semantic `search` only see a Codex session after `longhand codex-sync --semantic` has indexed it. For "what did I do in Codex" questions, use the `longhand-shared` server's keyword search (literal phrases; its session listing takes source="codex"), or tell the user to run the semantic sync. +Codex Desktop / CLI threads live in the same archive with `codex:`-prefixed session ids. This server lists and pages them (`list_sessions`, `get_session_timeline`) and `find_commits` sees commits made from Codex — but `recall` and semantic `search` only see a Codex session once it has been finalized — automatically, 30 minutes after the thread goes quiet (`longhand codex-sync --semantic` does it immediately). For "what did I do in Codex" questions about a thread that is still active, use the `longhand-shared` server's keyword search (literal phrases; its session listing takes source="codex"). diff --git a/README.md b/README.md index ff00e23..473e540 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ [![PyPI version](https://img.shields.io/pypi/v/longhand?label=PyPI&color=blue)](https://pypi.org/project/longhand/) ![Python](https://img.shields.io/badge/python-3.10+-blue) ![License](https://img.shields.io/badge/license-MIT-green) -![Tests](https://img.shields.io/badge/tests-580%20passing-brightgreen) +![Tests](https://img.shields.io/badge/tests-588%20passing-brightgreen) ![Local](https://img.shields.io/badge/100%25-local-informational) [![SafeSkill 93/100](https://img.shields.io/badge/SafeSkill-93%2F100_Verified%20Safe-brightgreen)](https://safeskill.dev/scan/wynelson94-longhand) @@ -87,7 +87,7 @@ longhand analyze --all # fill in episodes + vectors whenever, safe to Exact-text search, timelines, file history, and commit lookup all work after `--skip-analysis`. Semantic `recall` needs the `analyze --all` pass to complete. Typical throughput on an M-class Mac is ~1–2 sessions/sec for full analysis. -> *Status: v1.1.0 — stable, daily-driver tested, security-audited (zero critical findings), on PyPI, available as a Claude Code plugin. Validated against 433 real Claude Code sessions across 37 inferred projects (measured 2026-08-12). 580 unit tests passing.* +> *Status: v1.1.0 — stable, daily-driver tested, security-audited (zero critical findings), on PyPI, available as a Claude Code plugin. Validated against 433 real Claude Code sessions across 37 inferred projects (measured 2026-08-12). 588 unit tests passing.* **Full docs:** [Longhand Wiki](https://github.com/Wynelson94/longhand/wiki) — getting started, CLI reference, MCP tools reference, architecture, and troubleshooting. @@ -448,7 +448,7 @@ claude mcp add --scope user longhand-shared -- longhand shared-mcp # keyword s codex mcp add longhand -- longhand shared-mcp # the same server from Codex (Desktop: config.toml, see the docs) ``` -From then on `reconcile --fix` captures new Codex threads too, so the scheduled reconciler keeps both clients current. Codex sessions are stored exact-record-only — verbatim and searchable, no model loaded — until `longhand codex-sync --semantic` makes them recallable. Threads Codex spawns for itself are skipped, UI mirrors are never stored twice, and unknown record shapes surface in `doctor` like any other drift. Setup, bounds, and the macOS launchd template: **[docs/codex.md](docs/codex.md)**. +From then on `reconcile --fix` captures new Codex threads too, so the scheduled reconciler keeps both clients current. A thread is stored exact-record-only while it is being written — verbatim and searchable, no model loaded — and gets the full pipeline 30 minutes after it goes quiet, so `recall` sees it with no manual step (`codex-sync --semantic` does it immediately). Threads Codex spawns for itself are skipped, UI mirrors are never stored twice, and unknown record shapes surface in `doctor` like any other drift. Setup, bounds, and the macOS launchd template: **[docs/codex.md](docs/codex.md)**. --- @@ -547,7 +547,7 @@ Longhand is flat-cost: the cap is per-call, not per-corpus. Recalling across 10 --- -580 unit tests passing. All 13 MCP tools stress-tested. Full security audit: zero critical findings, zero high findings. `~/.longhand/` created with 0700 permissions, all SQL parameterized, all inputs bounded. Dependencies: chromadb, typer, rich, pydantic, mcp. +588 unit tests passing. All 13 MCP tools stress-tested. Full security audit: zero critical findings, zero high findings. `~/.longhand/` created with 0700 permissions, all SQL parameterized, all inputs bounded. Dependencies: chromadb, typer, rich, pydantic, mcp. --- diff --git a/docs/codex.md b/docs/codex.md index d190497..d20f05f 100644 --- a/docs/codex.md +++ b/docs/codex.md @@ -51,7 +51,7 @@ commands — one archive is the whole point. | ------------------------------------------------- | ----------------------------- | -------------------------- | | Lists and pages Codex sessions | yes | yes | | Keyword search across both clients | no | yes (`search`, literal) | -| `recall` and semantic search over Codex sessions | after `codex-sync --semantic` | no | +| `recall` and semantic search over Codex sessions | 30 min after a thread quiets | no | | Commits made in Codex | yes (`find_commits`) | through `search` | | Loads the embedding model | yes | never | @@ -63,21 +63,32 @@ Chroma, and never loads a model; long texts and raw records page in 8,000-character slices. A broad query on a large archive can hit the server's instruction budget — narrow it to a session. -Codex sessions are captured exact-record-only by default (ingestion stage -`archived`): every message, reasoning summary, tool call, and output is stored -verbatim and searchable, but not embedded. `longhand codex-sync --semantic` -runs the full pipeline — embeddings, episodes, project inference — on captured -sessions so `recall` and semantic `search` see them. It is the remedy `doctor` -names for archived sessions; `analyze` is not, because it never embeds events. +Capture runs in two passes, the Codex twin of Claude Code's Stop and SessionEnd +hooks. A new or changed rollout is first captured exact-record-only (ingestion +stage `archived`): every message, reasoning summary, tool call, and output is +stored verbatim and keyword-searchable, with no vector model loaded — the same +reason Claude's per-turn Stop hook skips embeddings. Codex sends no session-end +signal, so quiet stands in for it: once a rollout has been untouched for 30 +minutes (`--finalize-after`, in seconds) it gets the full pipeline — +embeddings, episodes, project inference — and `recall` and semantic `search` +see it. A finalized thread that resumes is captured exact-only again and +finalized again once it settles: one re-embed per resume. `longhand codex-sync +--semantic` runs the full pipeline on everything immediately, and +`--no-finalize` keeps a run exact-only. `doctor` shows a "Codex finalizer" row +while any thread is archived; `analyze` never embeds events, so it is not the +remedy for one. ## Keeping capture current -`longhand reconcile --fix` captures new or changed Codex rollouts along with -everything it already does for Claude transcripts, so the scheduled reconciler -(`longhand schedule install-reconciler`, every 30 minutes on macOS) keeps Codex -current with no extra setup. For an immediate capture run `longhand codex-sync`; -for a foreground loop run `longhand codex-sync --watch` (every 60 seconds until -interrupted). +`longhand reconcile --fix` runs both passes — captures new or changed Codex +rollouts and finalizes the quiet ones — along with everything it already does +for Claude transcripts, so the scheduled reconciler (`longhand schedule +install-reconciler`, every 30 minutes on macOS) keeps Codex current and +recallable with no extra setup. For an immediate run use `longhand codex-sync`; +for a foreground loop, `longhand codex-sync --watch` (every 60 seconds until +interrupted). On a 60-second schedule a thread is recallable about 30 minutes +after its last message; the poller loads the embedding model only on the run +that has a quiet thread to finalize, one thread per run. ### Faster capture with launchd (macOS) @@ -125,7 +136,8 @@ Scheduler. - **Bounds.** Per run: up to 50 sessions, each up to 16 MiB and 20,000 events. Larger rollouts are reported as `deferred`, never partially imported. Raise the bounds with `--limit`, `--max-file-kb`, and `--max-events`. These are - input bounds, not a memory ceiling — the default capture never loads a model. + input bounds, not a memory ceiling — the exact-record pass never loads a + model, and the finalizer loads it only when a quiet thread is waiting. - **Drift is never silent.** A record shape Longhand does not recognize is preserved as an `unknown` event with its raw JSON intact, and surfaces in `longhand doctor`'s "Transcript format" row as `response_item/` or diff --git a/longhand/cli/_commands.py b/longhand/cli/_commands.py index 054a6bd..cd920c1 100644 --- a/longhand/cli/_commands.py +++ b/longhand/cli/_commands.py @@ -32,7 +32,12 @@ _resolve_prefix, console, ) -from longhand.codex import DEFAULT_MAX_EVENTS, DEFAULT_MAX_FILE_BYTES, DEFAULT_SESSION_LIMIT +from longhand.codex import ( + DEFAULT_FINALIZE_AFTER_SECONDS, + DEFAULT_MAX_EVENTS, + DEFAULT_MAX_FILE_BYTES, + DEFAULT_SESSION_LIMIT, +) from longhand.parser import JSONLParser, discover_sessions from longhand.recall import recall as recall_pipeline from longhand.replay import ReplayEngine @@ -327,6 +332,18 @@ def codex_sync_cmd( help="Also capture threads Codex spawned for itself (e.g. its approval reviewer); " "skipped by default because they re-quote the parent thread", ), + finalize_after: int = typer.Option( + DEFAULT_FINALIZE_AFTER_SECONDS, + "--finalize-after", + min=0, + help="Seconds a thread must stay quiet before it gets the full pipeline " + "(embeddings, episodes, project inference) so recall and search see it", + ), + no_finalize: bool = typer.Option( + False, + "--no-finalize", + help="Capture exact records only; never run the full pipeline from this command", + ), dry_run: bool = typer.Option( False, "--dry-run", @@ -345,10 +362,14 @@ def codex_sync_cmd( ): """Capture Codex history into the same archive Claude uses. - Exact records only by default, with no vector model loaded. `reconcile - --fix` runs this capture too, so a scheduled reconciler keeps Codex - current on its own; run this directly for an immediate capture, or with - --watch for a 60-second loop. + Two passes, the Codex twin of Claude Code's Stop and SessionEnd hooks: + every new or changed rollout is captured exact-record-only (no vector + model loaded), then any rollout quiet for --finalize-after seconds gets + the full pipeline so recall and search see it. The model loads only when + a quiet thread is waiting. `reconcile --fix` runs both passes too, so a + scheduled reconciler keeps Codex current on its own; run this directly + for an immediate capture, with --watch for a 60-second loop, or with + --semantic to run the full pipeline on everything now. """ import json import time @@ -356,6 +377,7 @@ def codex_sync_cmd( from longhand.codex import ( CodexArchiveStore, discover_codex_sessions, + finalize_codex, is_subagent_rollout, read_session_meta, sync_codex, @@ -385,6 +407,20 @@ def codex_sync_cmd( max_events=max_events, include_subagents=include_subagents, ) + if not semantic and not no_finalize: + finalized = finalize_codex( + store.sqlite, + codex_home, + lambda: _get_store(data_dir), + idle_seconds=finalize_after, + max_file_bytes=max_file_kb * 1024, + max_events=max_events, + include_subagents=include_subagents, + ) + report["finalized"] = finalized["finalized"] + report["settling"] = finalized["settling"] + report["deferred"] = list(dict.fromkeys(report["deferred"] + finalized["deferred"])) + report["errors"].extend(finalized["errors"]) console.print(json.dumps(report)) if not watch: if report["errors"]: @@ -607,6 +643,15 @@ def reconcile( console.print(f"[bold]Codex rollouts:[/bold] {report.codex_on_disk} on disk") console.print(f" [green]{current}[/green] captured and current") console.print(f" [yellow]{report.codex_pending}[/yellow] new or changed since capture") + if report.codex_unfinalized: + console.print( + f" [yellow]{report.codex_unfinalized}[/yellow] quiet thread(s) awaiting the " + "full pipeline" + ) + if report.codex_settling: + console.print( + f" [dim]{report.codex_settling} still being written (finalized once quiet)[/dim]" + ) if report.codex_skipped_subagents: console.print( f" [dim]{report.codex_skipped_subagents} subagent thread(s) skipped " @@ -619,8 +664,9 @@ def reconcile( ) fixable_claude = bool(report.missing or report.null_project or report.partially_indexed) + fixable_codex = bool(report.codex_pending or report.codex_unfinalized) if not fix: - if fixable_claude or report.codex_pending: + if fixable_claude or fixable_codex: console.print("\n[dim]Run with --fix to re-ingest.[/dim]") return @@ -628,7 +674,7 @@ def reconcile( console.print("[yellow]Another ingest is running — aborting reconcile.[/yellow]") raise typer.Exit(1) - if not fixable_claude and not report.codex_pending: + if not fixable_claude and not fixable_codex: console.print("\n[green]Nothing to fix.[/green]") return @@ -645,6 +691,10 @@ def reconcile( f" ([dim]{len(report.codex_deferred)} deferred[/dim])" if report.codex_deferred else "" ) console.print(f"[bold]Captured {report.codex_ingested}[/bold] Codex rollout(s){deferred}") + settling = ( + f" ([dim]{report.codex_settling} settling[/dim])" if report.codex_settling else "" + ) + console.print(f"[bold]Finalized {report.codex_finalized}[/bold] Codex thread(s){settling}") # ----------------------------------------------------------------------------- diff --git a/longhand/codex.py b/longhand/codex.py index 6dc1e67..b4da4e6 100644 --- a/longhand/codex.py +++ b/longhand/codex.py @@ -26,6 +26,7 @@ import json import os import re +import time from dataclasses import dataclass, field from itertools import islice from pathlib import Path @@ -36,7 +37,10 @@ from longhand.types import Event, EventType if TYPE_CHECKING: + from collections.abc import Callable + from longhand.storage.sqlite_store import SQLiteStore + from longhand.storage.store import LonghandStore from longhand.types import Session # Capture bounds. These are input bounds, not a memory ceiling: the default @@ -46,6 +50,11 @@ DEFAULT_MAX_FILE_BYTES = 16 * 1024 * 1024 DEFAULT_MAX_EVENTS = 20_000 +# Codex sends no session-end signal, so quiet stands in for it: a rollout +# untouched for this long gets the full pipeline (embeddings, episodes, +# project inference) — the Codex twin of Claude Code's SessionEnd hook. +DEFAULT_FINALIZE_AFTER_SECONDS = 1800 + # Top-level record types with nothing recallable in them — usage accounting # and world-state snapshots. Skipped, not stored: the Claude parser's # KNOWN_SKIP_ENTRY_TYPES rule. Every member needs a fixture line in @@ -384,10 +393,17 @@ class CodexScan: unchanged: list[Path] = field(default_factory=list) # captured at this exact size subagents: list[Path] = field(default_factory=list) # skipped unless include_subagents oversize: list[Path] = field(default_factory=list) # over max_file_bytes + settling: list[Path] = field(default_factory=list) # changed within min_idle_seconds @property def on_disk(self) -> int: - return len(self.candidates) + len(self.unchanged) + len(self.subagents) + len(self.oversize) + return ( + len(self.candidates) + + len(self.unchanged) + + len(self.subagents) + + len(self.oversize) + + len(self.settling) + ) def scan_codex_sessions( @@ -397,15 +413,23 @@ def scan_codex_sessions( max_file_bytes: int = DEFAULT_MAX_FILE_BYTES, include_subagents: bool = False, complete_stages: tuple[str, ...] = ("archived", "analyzed"), + min_idle_seconds: float | None = None, ) -> CodexScan: - """Classify every rollout on disk against the archive without parsing any.""" + """Classify every rollout on disk against the archive without parsing any. + + With `min_idle_seconds`, a rollout written more recently than that is + `settling` rather than a candidate: the thread may still be running, so + the caller waits for it to go quiet. `None` never waits (live capture). + """ scan = CodexScan() stages = sqlite.analysis_stages() + now = time.time() for path in discover_codex_sessions(codex_home): try: - size = path.stat().st_size + stat = path.stat() except OSError: continue + size = stat.st_size key = str(path) if sqlite.already_ingested(key, size) and stages.get(key) in complete_stages: scan.unchanged.append(path) @@ -413,6 +437,8 @@ def scan_codex_sessions( scan.subagents.append(path) elif size > max_file_bytes: scan.oversize.append(path) + elif min_idle_seconds is not None and now - stat.st_mtime < min_idle_seconds: + scan.settling.append(path) else: scan.candidates.append(path) return scan @@ -427,6 +453,7 @@ def sync_codex( max_events: int = DEFAULT_MAX_EVENTS, include_subagents: bool = False, claim_lock: bool = True, + min_idle_seconds: float | None = None, ) -> dict[str, Any]: """Import new or changed rollouts into the archive. @@ -436,8 +463,11 @@ def sync_codex( reconcile does. This is deliberately a separate discovery path: Codex support never changes which files Claude's hooks scan. - Report keys: ingested, skipped (unchanged), skipped_subagents, deferred - (over a bound, or past `limit` for this run), errors, locked. + `min_idle_seconds` leaves rollouts written more recently than that alone + (reported as `settling`); live capture passes `None` and never waits. + + Report keys: ingested, skipped (unchanged), skipped_subagents, settling, + deferred (over a bound, or past `limit` for this run), errors, locked. """ from longhand.parser import JSONLParser from longhand.recall.project_fallback import claim_ingest_lock, release_ingest_lock @@ -448,6 +478,7 @@ def sync_codex( "ingested": 0, "skipped": 0, "skipped_subagents": 0, + "settling": 0, "deferred": [], "errors": [], "locked": False, @@ -465,9 +496,11 @@ def sync_codex( max_file_bytes=max_file_bytes, include_subagents=include_subagents, complete_stages=complete, + min_idle_seconds=min_idle_seconds, ) report["skipped"] = len(scan.unchanged) report["skipped_subagents"] = len(scan.subagents) + report["settling"] = len(scan.settling) report["deferred"].extend(str(p) for p in scan.oversize) for index, path in enumerate(scan.candidates): if index >= limit: @@ -502,3 +535,64 @@ def sync_codex( if claim_lock: release_ingest_lock(store) return report + + +def finalize_codex( + sqlite: SQLiteStore, + codex_home: str | Path | None, + build_store: Callable[[], LonghandStore], + *, + idle_seconds: float = DEFAULT_FINALIZE_AFTER_SECONDS, + limit: int = 1, + max_file_bytes: int = DEFAULT_MAX_FILE_BYTES, + max_events: int = DEFAULT_MAX_EVENTS, + include_subagents: bool = False, + claim_lock: bool = True, +) -> dict[str, Any]: + """Run the full pipeline on rollouts that have gone quiet. + + The synthetic SessionEnd for Codex: a rollout not yet `analyzed` whose + file has been untouched for `idle_seconds` is re-ingested through the + LonghandStore (embeddings, episodes, project inference), exactly as + `codex-sync --semantic` would. A rollout written more recently is + `settling` and left for a later run. + + `build_store` is called only when there is something to finalize, so the + every-minute steady state never loads the embedding model. `limit` + bounds the work per run; the rest is `deferred` to the next one. + + Report keys: finalized, settling, deferred, errors, locked. + """ + scan = scan_codex_sessions( + sqlite, + codex_home, + max_file_bytes=max_file_bytes, + include_subagents=include_subagents, + complete_stages=("analyzed",), + min_idle_seconds=idle_seconds, + ) + if not scan.candidates: + return { + "finalized": 0, + "settling": len(scan.settling), + "deferred": [str(p) for p in scan.oversize], + "errors": [], + "locked": False, + } + report = sync_codex( + build_store(), + codex_home, + limit=limit, + max_file_bytes=max_file_bytes, + max_events=max_events, + include_subagents=include_subagents, + claim_lock=claim_lock, + min_idle_seconds=idle_seconds, + ) + return { + "finalized": report["ingested"], + "settling": report["settling"], + "deferred": report["deferred"], + "errors": report["errors"], + "locked": report["locked"], + } diff --git a/longhand/recall/reconcile.py b/longhand/recall/reconcile.py index 451ea76..e3c6beb 100644 --- a/longhand/recall/reconcile.py +++ b/longhand/recall/reconcile.py @@ -5,8 +5,10 @@ re-ingests the missing, null-project, and partially-indexed entries. Codex rollouts ride along when a Codex home exists: new or changed ones are -captured exact-record-only under the same ingest lock, so the scheduled -reconciler keeps both clients' history current with no extra setup. +captured exact-record-only under the same ingest lock, and rollouts that have +gone quiet get the full pipeline (the Codex twin of Claude's SessionEnd hook), +so the scheduled reconciler keeps both clients' history current and recallable +with no extra setup. """ from __future__ import annotations @@ -41,6 +43,9 @@ class ReconcileReport: codex_oversize: int = 0 codex_ingested: int = 0 codex_deferred: list[str] = field(default_factory=list) + codex_unfinalized: int = 0 # quiet but not yet through the full pipeline + codex_settling: int = 0 # written too recently to finalize this run + codex_finalized: int = 0 def to_dict(self) -> dict[str, object]: return { @@ -64,6 +69,9 @@ def to_dict(self) -> dict[str, object]: "codex_oversize": self.codex_oversize, "codex_ingested": self.codex_ingested, "codex_deferred": self.codex_deferred, + "codex_unfinalized": self.codex_unfinalized, + "codex_settling": self.codex_settling, + "codex_finalized": self.codex_finalized, } @@ -75,11 +83,20 @@ def run_reconcile( Without `fix`: returns counts only. With `fix=True`: acquires the ingest lock, re-ingests missing, null-project, and partially-indexed Claude entries using current project - inference, and captures new or changed Codex rollouts (exact records - only). If another ingest is running, returns with - `lock_unavailable=True` and nothing ingested. + inference, captures new or changed Codex rollouts (exact records only), + and finalizes Codex rollouts that have been quiet for + DEFAULT_FINALIZE_AFTER_SECONDS (full pipeline, so recall sees them). If + another ingest is running, returns with `lock_unavailable=True` and + nothing ingested. """ - from longhand.codex import CodexArchiveStore, scan_codex_sessions, sync_codex + from longhand.codex import ( + DEFAULT_FINALIZE_AFTER_SECONDS, + DEFAULT_SESSION_LIMIT, + CodexArchiveStore, + finalize_codex, + scan_codex_sessions, + sync_codex, + ) files = discover_sessions() @@ -121,6 +138,12 @@ def run_reconcile( fully_indexed += 1 codex_scan = scan_codex_sessions(store.sqlite, codex_home) + finalize_scan = scan_codex_sessions( + store.sqlite, + codex_home, + complete_stages=("analyzed",), + min_idle_seconds=DEFAULT_FINALIZE_AFTER_SECONDS, + ) report = ReconcileReport( files_on_disk=len(files), @@ -133,13 +156,15 @@ def run_reconcile( codex_pending=len(codex_scan.candidates), codex_skipped_subagents=len(codex_scan.subagents), codex_oversize=len(codex_scan.oversize), + codex_unfinalized=len(finalize_scan.candidates), + codex_settling=len(finalize_scan.settling), ) if not fix: return report to_process = missing + null_project + partial - if not to_process and not codex_scan.candidates: + if not to_process and not codex_scan.candidates and not finalize_scan.candidates: report.fix_applied = True return report @@ -171,6 +196,20 @@ def run_reconcile( report.codex_ingested = codex_report["ingested"] report.codex_deferred = list(codex_report["deferred"]) report.errors.extend(codex_report["errors"]) + + if finalize_scan.candidates: + # The store already holds the model and the lock; as the catch-up + # job, clear the whole quiet backlog rather than one per run. + finalized = finalize_codex( + store.sqlite, + codex_home, + lambda: store, + limit=DEFAULT_SESSION_LIMIT, + claim_lock=False, + ) + report.codex_finalized = finalized["finalized"] + report.codex_settling = finalized["settling"] + report.errors.extend(finalized["errors"]) finally: release_ingest_lock(store) diff --git a/longhand/setup_commands.py b/longhand/setup_commands.py index 49084c1..7c99f6e 100644 --- a/longhand/setup_commands.py +++ b/longhand/setup_commands.py @@ -1144,6 +1144,53 @@ class token is already there, so the split needs no new plumbing. return parts[2].removesuffix(":") +def _codex_finalize_status(store: LonghandStore) -> str | None: + """Rich-formatted state of Codex rollouts captured exact-record-only. + + `archived` is a normal transient stage now: a thread is finalized (full + pipeline) once it has been quiet for DEFAULT_FINALIZE_AFTER_SECONDS, by + the codex-sync poller or the scheduled reconciler. So the row is green + while every archived rollout is still being written, and yellow only when + one has been quiet long enough and nothing has finalized it yet. None + when nothing is archived. + """ + import time + + from longhand.codex import DEFAULT_FINALIZE_AFTER_SECONDS + + try: + with store.sqlite.connect() as conn: + rows = conn.execute( + "SELECT transcript_path FROM ingestion_log WHERE analysis_stage = 'archived'" + ).fetchall() + except Exception: + return None + settling = waiting = 0 + now = time.time() + for row in rows: + try: + idle = now - Path(row[0]).stat().st_mtime + except OSError: + continue # rollout gone from disk: nothing can finalize it + if idle < DEFAULT_FINALIZE_AFTER_SECONDS: + settling += 1 + else: + waiting += 1 + minutes = DEFAULT_FINALIZE_AFTER_SECONDS // 60 + if waiting: + return ( + f"[yellow]⚠[/yellow] {waiting} awaiting the finalizer — " + f"[bold]longhand codex-sync[/bold] and [bold]reconcile --fix[/bold] run it on " + "their schedules; run codex-sync now to index immediately" + ) + if settling: + return ( + f"[green]✓[/green] {settling} Codex thread(s) active — " + f"indexed {minutes} min after they go quiet" + ) + return None + + def _archived_session_count(store: LonghandStore) -> int: """Sessions captured exact-record-only (Codex `archived` stage): no vectors yet.""" try: @@ -1448,8 +1495,8 @@ def _row(label: str, value: str) -> None: # Archived Codex sessions have no outcome row either, but `analyze` is # the wrong remedy for them: it rebuilds outcomes and episodes without - # ever embedding the events, so search would stay blind. Name the remedy - # that actually works for that class (Promise 5). + # ever embedding the events, so search would stay blind. They get their + # own row, whose remedy is the finalizer (Promise 5). archived = _archived_session_count(store) sessions_needing_analysis = max( 0, stats.get("sessions", 0) - stats.get("outcomes", 0) - archived @@ -1459,12 +1506,9 @@ def _row(label: str, value: str) -> None: "Sessions needing analysis", f"[yellow]{sessions_needing_analysis}[/yellow] (run [bold]longhand analyze --all[/bold])", ) - if archived > 0: - _row( - "Codex sessions archived", - f"[yellow]{archived}[/yellow] captured exact-record only — recall and search " - "skip them (run [bold]longhand codex-sync --semantic[/bold] to index them)", - ) + finalize_row = _codex_finalize_status(store) + if finalize_row: + _row("Codex finalizer", finalize_row) # 7. Storage footprint — the corpus grows forever by design, but nothing # surfaced how big it had gotten (3 GB observed in the wild). diff --git a/tests/test_codex.py b/tests/test_codex.py index 569ef64..4735093 100644 --- a/tests/test_codex.py +++ b/tests/test_codex.py @@ -382,7 +382,8 @@ def test_reconcile_captures_codex_rollouts_under_its_own_lock(temp_store, tmp_pa assert fixed.codex_ingested == 1 assert fixed.errors == [] assert temp_store.sqlite.get_events(session_id="codex:primary") - # The archive path stamps `archived`, never the semantic stages. + # Freshly written, so still settling: the exact pass stamps `archived` + # and the finalizer waits for the thread to go quiet. assert temp_store.sqlite.analysis_stages()[str(home / "sessions/primary.jsonl")] == "archived" # Reconcile owned the lock throughout and released it afterwards. @@ -421,6 +422,8 @@ def test_shared_mcp_and_codex_sync_are_registered_cli_commands(): assert options["max_events"].default == 20000 assert options["include_subagents"].default is False assert "--include-subagents" in options["include_subagents"].param_decls + assert options["finalize_after"].default == 1800 + assert "--no-finalize" in options["no_finalize"].param_decls def test_doctor_codex_capture_row(tmp_path, monkeypatch): @@ -475,3 +478,277 @@ def test_doctor_drift_row_names_the_codex_kind(tmp_path): ) row = _transcript_format_status(store) assert "event_msg/brand_new_kind ×1" in row + + +def test_scan_puts_recently_written_rollouts_in_settling(tmp_path): + import os + import time + + from longhand.codex import scan_codex_sessions + + source = tmp_path / "codex/sessions/rollout.jsonl" + write_rollout(source) + sqlite = SQLiteStore(tmp_path / "longhand.db") + # Without an idle bound the scan is unchanged: a fresh rollout is a candidate. + assert scan_codex_sessions(sqlite, tmp_path / "codex").candidates == [source] + + scan = scan_codex_sessions(sqlite, tmp_path / "codex", min_idle_seconds=600) + assert (scan.candidates, scan.settling) == ([], [source]) + assert scan.on_disk == 1 + + quiet = time.time() - 31 * 60 + os.utime(source, (quiet, quiet)) + scan = scan_codex_sessions(sqlite, tmp_path / "codex", min_idle_seconds=600) + assert (scan.candidates, scan.settling) == ([source], []) + + +def _analyzing_store(sqlite, source): + """A stand-in for LonghandStore: ingests into SQLite and stamps `analyzed`.""" + store = MagicMock() + store.data_dir = source.parents[2] / "archive" + store.sqlite = sqlite + + def ingest(session, events, run_analysis=True): + sqlite.upsert_session(session) + sqlite.insert_events(events) + sqlite.log_ingestion( + session.transcript_path, session.session_id, source.stat().st_size, len(events) + ) + sqlite.set_analysis_stage(session.transcript_path, "analyzed") + + store.ingest_session.side_effect = ingest + return store + + +def test_finalize_waits_for_quiet_then_runs_full_pipeline(tmp_path): + import os + import time + + from longhand.codex import CodexArchiveStore, finalize_codex + + source = tmp_path / "codex/sessions/rollout.jsonl" + write_rollout(source) + archive = CodexArchiveStore(tmp_path / "archive") + assert sync_codex(archive, tmp_path / "codex")["ingested"] == 1 + assert archive.sqlite.analysis_stages()[str(source)] == "archived" + + # Still settling: nothing is finalized and the semantic store is never built. + report = finalize_codex( + archive.sqlite, + tmp_path / "codex", + lambda: pytest.fail("built the semantic store while the thread was settling"), + idle_seconds=600, + ) + assert (report["finalized"], report["settling"]) == (0, 1) + assert archive.sqlite.analysis_stages()[str(source)] == "archived" + + quiet = time.time() - 31 * 60 + os.utime(source, (quiet, quiet)) + semantic = _analyzing_store(archive.sqlite, source) + report = finalize_codex(archive.sqlite, tmp_path / "codex", lambda: semantic, idle_seconds=600) + assert (report["finalized"], report["settling"], report["errors"]) == (1, 0, []) + assert archive.sqlite.analysis_stages()[str(source)] == "analyzed" + semantic.ingest_session.assert_called_once() + + # Nothing left to do: unchanged and analyzed. + report = finalize_codex(archive.sqlite, tmp_path / "codex", lambda: semantic, idle_seconds=600) + assert report["finalized"] == 0 + semantic.ingest_session.assert_called_once() + + +def test_finalize_after_append_regresses_then_recovers(tmp_path): + import os + import time + + from longhand.codex import CodexArchiveStore, finalize_codex + + source = tmp_path / "codex/sessions/rollout.jsonl" + write_rollout(source) + archive = CodexArchiveStore(tmp_path / "archive") + semantic = _analyzing_store(archive.sqlite, source) + quiet = time.time() - 31 * 60 + os.utime(source, (quiet, quiet)) + sync_codex(archive, tmp_path / "codex") + assert finalize_codex(archive.sqlite, tmp_path / "codex", lambda: semantic)["finalized"] == 1 + + # The thread resumes: live capture is exact-only again and the stage regresses. + with source.open("a") as f: + f.write( + json.dumps( + record( + "response_item", + { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": "Anything else?"}], + }, + ) + ) + + "\n" + ) + assert sync_codex(archive, tmp_path / "codex")["ingested"] == 1 + assert archive.sqlite.analysis_stages()[str(source)] == "archived" + assert finalize_codex(archive.sqlite, tmp_path / "codex", lambda: semantic)["settling"] == 1 + + # Ten quiet minutes later it is finalized once more — one re-embed per resume. + os.utime(source, (quiet, quiet)) + assert finalize_codex(archive.sqlite, tmp_path / "codex", lambda: semantic)["finalized"] == 1 + assert archive.sqlite.analysis_stages()[str(source)] == "analyzed" + assert len(archive.sqlite.get_events(session_id="codex:test-id")) == 11 + + +def test_reconcile_finalizes_quiet_codex_rollouts(temp_store, tmp_path, monkeypatch): + import os + import time + + from longhand.recall import reconcile as reconcile_mod + from longhand.recall.reconcile import run_reconcile + + monkeypatch.setattr(reconcile_mod, "discover_sessions", lambda: []) + home = tmp_path / "codex" + quiet_path = home / "sessions/quiet.jsonl" + active_path = home / "sessions/active.jsonl" + write_rollout(quiet_path, "quiet") + write_rollout(active_path, "active") + quiet = time.time() - 31 * 60 + os.utime(quiet_path, (quiet, quiet)) + + dry = run_reconcile(temp_store, fix=False, codex_home=home) + assert (dry.codex_pending, dry.codex_unfinalized, dry.codex_settling) == (2, 1, 1) + assert dry.to_dict()["codex_unfinalized"] == 1 + assert dry.codex_finalized == 0 + + fixed = run_reconcile(temp_store, fix=True, codex_home=home) + assert fixed.errors == [] + assert (fixed.codex_ingested, fixed.codex_finalized, fixed.codex_settling) == (2, 1, 1) + stages = temp_store.sqlite.analysis_stages() + assert stages[str(quiet_path)] == "analyzed" + assert stages[str(active_path)] == "archived" + # The quiet thread is now recallable: its events reached the vector store. + assert temp_store.vectors.events_collection.get( + where={"session_id": "codex:quiet"}, include=[] + )["ids"] + + after = run_reconcile(temp_store, fix=False, codex_home=home) + assert (after.codex_pending, after.codex_unfinalized, after.codex_settling) == (0, 0, 1) + + +def _cli_json(output): + """Rich wraps long lines at the runner's width; rejoin before parsing.""" + return json.loads("".join(output.splitlines())) + + +def test_codex_sync_cli_never_loads_model_while_threads_settle(tmp_path, monkeypatch): + from typer.testing import CliRunner + + from longhand.cli import app + + home = tmp_path / "codex" + write_rollout(home / "sessions/active.jsonl", "active") + monkeypatch.setenv("CODEX_HOME", str(home)) + monkeypatch.setenv("LONGHAND_DATA_DIR", str(tmp_path / "archive")) + monkeypatch.setattr( + "longhand.storage.store.VectorStore", lambda *a, **k: pytest.fail("loaded vectors") + ) + result = CliRunner().invoke(app, ["codex-sync"]) + assert result.exit_code == 0, result.output + report = _cli_json(result.output) + assert (report["ingested"], report["finalized"], report["settling"]) == (1, 0, 1) + + +def test_codex_sync_cli_finalizes_quiet_threads(tmp_path, monkeypatch): + import os + import time + + from typer.testing import CliRunner + + from longhand.cli import _commands, app + + home = tmp_path / "codex" + source = home / "sessions/quiet.jsonl" + write_rollout(source, "quiet") + quiet = time.time() - 31 * 60 + os.utime(source, (quiet, quiet)) + archive = tmp_path / "archive" + monkeypatch.setenv("CODEX_HOME", str(home)) + monkeypatch.setenv("LONGHAND_DATA_DIR", str(archive)) + archive.mkdir() + sqlite = SQLiteStore(archive / "longhand.db") + semantic = _analyzing_store(sqlite, source) + semantic.data_dir = archive + monkeypatch.setattr(_commands, "_get_store", lambda data_dir=None: semantic) + + result = CliRunner().invoke(app, ["codex-sync"]) + assert result.exit_code == 0, result.output + report = _cli_json(result.output) + assert (report["ingested"], report["finalized"], report["settling"]) == (1, 1, 0) + assert sqlite.analysis_stages()[str(source)] == "analyzed" + + # Opting out leaves the quiet thread archived: nothing else changes. + with sqlite.connect() as conn: + conn.execute("UPDATE ingestion_log SET analysis_stage = 'archived'") + semantic.ingest_session.reset_mock() + result = CliRunner().invoke(app, ["codex-sync", "--no-finalize"]) + assert result.exit_code == 0, result.output + assert "finalized" not in _cli_json(result.output) + semantic.ingest_session.assert_not_called() + + +def test_doctor_codex_finalizer_row(tmp_path, monkeypatch): + import os + import time + + from longhand.codex import CodexArchiveStore + from longhand.setup_commands import _codex_finalize_status + + store = CodexArchiveStore(tmp_path / "archive") + home = tmp_path / "codex" + source = home / "sessions/active.jsonl" + write_rollout(source, "active") + monkeypatch.setenv("CODEX_HOME", str(home)) + # Nothing captured exact-only yet: no row. + assert _codex_finalize_status(store) is None + + sync_codex(store, home) + settling = _codex_finalize_status(store) + assert "1 Codex thread(s) active" in settling + assert "30 min" in settling + + quiet = time.time() - 31 * 60 + os.utime(source, (quiet, quiet)) + waiting = _codex_finalize_status(store) + assert "1 awaiting the finalizer" in waiting + assert "longhand codex-sync" in waiting + + store.sqlite.set_analysis_stage(str(source), "analyzed") + assert _codex_finalize_status(store) is None + + +def test_reconcile_cli_reports_codex_finalization(temp_store, tmp_path, monkeypatch): + import os + import time + + from typer.testing import CliRunner + + from longhand.cli import app + from longhand.codex import CodexArchiveStore + from longhand.recall import reconcile as reconcile_mod + + monkeypatch.setattr(reconcile_mod, "discover_sessions", lambda: []) + home = tmp_path / "codex" + source = home / "sessions/quiet.jsonl" + write_rollout(source, "quiet") + quiet = time.time() - 31 * 60 + os.utime(source, (quiet, quiet)) + monkeypatch.setenv("CODEX_HOME", str(home)) + sync_codex(CodexArchiveStore(temp_store.data_dir, sqlite=temp_store.sqlite), home) + + runner = CliRunner() + dry = runner.invoke(app, ["reconcile", "--data-dir", str(temp_store.data_dir)]) + assert dry.exit_code == 0, dry.output + assert "1 quiet thread(s) awaiting the full pipeline" in dry.output + + fixed = runner.invoke(app, ["reconcile", "--fix", "--data-dir", str(temp_store.data_dir)]) + assert fixed.exit_code == 0, fixed.output + assert "Finalized 1 Codex thread(s)" in fixed.output + assert temp_store.sqlite.analysis_stages()[str(source)] == "analyzed"