diff --git a/.beads/issues.jsonl b/.beads/issues.jsonl index c34b92d542..d343e89656 100644 --- a/.beads/issues.jsonl +++ b/.beads/issues.jsonl @@ -57,7 +57,7 @@ {"_type":"issue","id":"polylogue-aagkt","title":"Closure discipline: closing with deferred scope requires a linked successor bead (recurring pattern, 3 instances found 2026-08-03)","description":"Process finding of the 2026-08-03 reindex-gate-hunt (feeds wwph1). Same failure shape found three independent times in one afternoon: a bead closes on its forward-looking fix while deferring the retroactive-repair or root-cause AC, and no successor bead is created — the deferred scope becomes anonymous debt discoverable only by archaeology.\n\nInstances: (1) gysk3 — root cause (18 positional-id parser call sites) deferred into docs/plans/position-derived-identity-acks.json, a lint-suppression registry whose every entry references the closed bead itself; zero open tracking bead until polylogue-slshy (this batch). (2) 8b10 — closed although its own notes state the closure gate (post-rebuild thinking-count check) was never met; no carrier until the v50-verify successor (this batch). (3) mvq8 — closed on AC(1) detector fix; AC(2) retroactive reclassification never done, no successor until polylogue-s8s54 (this batch). All three were on or adjacent to the 818fy critical path.\n\nOPERATOR RULING (2026-08-03): the fix is NOT machinery. A closure lint pattern-matching close-reason prose was explicitly rejected as itself a bloat class (programmatic interfaces to natural language yield false positives, regex-pleasing phrasing, rotting allowlists); the rule now lives in global CLAUDE.md (sinnix 8760308). The fix is the structured-successor convention in acceptance, applied by judgment.\n","acceptance_criteria":"Convention recorded in .agent/CONVENTIONS.md: a bd close whose reason defers any AC/root-cause/repair scope must name an ALREADY-CREATED successor bead id, linked via a structured dep/related edge, in the same batch as the close. The three found instances are each carried by a real successor (slshy for gysk3; the v50-verify successor for 8b10; s8s54 for mvq8). NO lint/gate greps prose for this — enforcement is judgment at close/review time per the global CLAUDE.md rule (sinnix 8760308).","notes":"2026-08-06 campaign graph promotion: this bead is a direct prerequisite or process guard for proof-carrying reindex acceptance. Its implementation cannot substitute for the terminal receipt, but its output is consumed by the campaign ledger and final gate.","status":"open","priority":0,"issue_type":"chore","owner":"ezo.dev@gmail.com","created_at":"2026-08-03T12:19:08Z","created_by":"Sinity","updated_at":"2026-08-06T05:00:41Z","dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"polylogue-r9xsj","title":"acceptance: schema-inference gate requires zero dupes, zero surviving quarantine, full reconciliation vs .claude/.codex/GDPR-export ground truth","description":"Operator hard requirement 2026-08-03: before running real schema inference (tnqqt), the production blobstore must be FULLY reconciled against its ground-truth acquisition sources - not just internally deduped. Concretely: every codex-session raw traceable/reconciled against ~/.codex/sessions; every claude-code-session raw against ~/.claude/projects; every export-origin raw (chatgpt-export, claude-ai-export, gemini/drive, grok-export) against its GDPR/Takeout export bundle. Zero tolerance: no byte-duplicate raws left unresolved (6753s), no revision_authority='quarantined' survivors at all (not just deduped away - the CONCEPT is meant to retire per lb39z/w6hql, not persist as a smaller pile). SCOPE NUANCE: hooks (raw_hook_events) and browser-capture origins have NO external filesystem ground truth to reconcile against - they ARE the origin (live capture, not a re-scannable export/session-dir), so this AC applies only to origins with an external ground-truth corpus. AC: (1) per-origin reconciliation census: source-tier raw count/bytes vs the actual ground-truth directory/export contents, with an explicit accounting for every logical source (materialized | superseded-duplicate | legitimately-excluded-non-conversation, never bare 'quarantined'); (2) zero unresolved raw_authority_blockers; (3) zero rows with revision_authority='quarantined' and no logical_source_key. Depends on 6753s (dedup), lkrc+hjpx (reconciler to fixed point), lb39z (drain fake quarantine), w6hql (collapse vocabulary - quarantine as a distinct concept goes away). Gates tnqqt and 818fy.","design":"DESIGN (2026-08-03): make each gate item a concrete, runnable query against named tables, so \"pass\" is mechanical.\n\nAll queries read-only against /realm/db/polylogue (sqlite3 'file:...?mode=ro' URIs).\n\nGATE QUERIES (all must return 0 / empty):\n1. Zero surviving quarantine — the CONCEPT retires, not just the pile shrinks:\n `SELECT COUNT(*) FROM raw_sessions WHERE revision_authority='quarantined';` == 0 on source.db.\n (Column: raw_sessions.revision_authority, closed 3-value RawRevisionAuthority vocabulary, DEFAULT 'quarantined' — archive_tiers/source.py:56. Note migration 023's byte-dup supersession actuator promotes to 'byte_proven' with a receipt; migration 019/020 actuators cover membership-writeback and append-chain populations. Whatever rows remain after lb39z/lkrc drain to fixed point must be individually accounted, not left as a smaller pile.)\n2. Zero orphan quarantine (the never-reconciled subclass): `SELECT COUNT(*) FROM raw_sessions WHERE revision_authority='quarantined' AND logical_source_key IS NULL;` == 0. (Subset of query 1; listed separately because this was the 4,305-row/17.6-GiB blind spot 6753s found — every reconciliation path keys off logical_source_key.)\n3. Zero open authority blockers: `SELECT COUNT(*) FROM raw_authority_blockers WHERE resolved_at_ms IS NULL;` == 0. (Open-row predicate matches idx_raw_authority_blockers_open_plan, source.py:378-380.)\n4. Zero unexplained byte-duplicates: every quarantine-era duplicate must carry a receipt row in raw_byte_duplicate_supersession_receipts (migration 023) or have been drained by the reconciler; check: no raw_sessions row shares a blob_hash with an indexed twin while itself being neither materialized, receipted-superseded, nor typed-excluded.\n\nRECONCILIATION CENSUS (instrument exists — do not build a new one): .agent/scripts/corpus-fidelity-audit.py (polylogue-f1vg) already computes absent-documents + revision-shortfall + unacquired-attachments against ground truth. Last run 2026-08-02: FAIL — 1,267 absent documents, 100 shortfall sessions (explained: 76 Hermes staleness + quarantined claude-code cohorts, both resolved by drain+reindex), 9,767 unacquired attachment refs. AC is that this audit reaches PASS, or every residual failure line carries an explicit typed explanation (materialized | superseded-duplicate | legitimately-excluded-non-conversation), never bare 'quarantined'.\nPer-origin ground-truth denominators: codex-session vs ~/.codex/sessions; claude-code-session vs ~/.claude/projects; chatgpt-export / claude-ai-export / gemini-drive / grok-export vs their export bundles under /realm/data/exports/chatlog. Hooks + browser-capture are exempt (no external ground truth — they ARE the origin), per description.\n\nSEQUENCING NOTE: queries 1-4 are meaningful only AFTER the deploy/migration step (9qnzy/a7gmk) and the raw-authority drain (lb39z/lkrc/hjpx/yla8 live halves) have run — this bead is the acceptance instrument for that program, not new mechanism. Its only buildable artifact is a small census script/registry check bundling queries 1-4 + the fidelity-audit verdict into one PASS/FAIL receipt the 818fy runbook can cite.\n","acceptance_criteria":"All against /realm/db/polylogue read-only, AFTER the drain program (lb39z/lkrc/hjpx/yla8 live halves) and deploy (9qnzy/a7gmk):\n1. `SELECT COUNT(*) FROM raw_sessions WHERE revision_authority='quarantined'` == 0 (source.db).\n2. `SELECT COUNT(*) FROM raw_sessions WHERE revision_authority='quarantined' AND logical_source_key IS NULL` == 0.\n3. `SELECT COUNT(*) FROM raw_authority_blockers WHERE resolved_at_ms IS NULL` == 0.\n4. No raw_sessions row shares a blob_hash with an indexed twin while being neither materialized, receipted in raw_byte_duplicate_supersession_receipts, nor typed-excluded.\n5. .agent/scripts/corpus-fidelity-audit.py (polylogue-f1vg) returns PASS, or every residual failure line carries a typed explanation (materialized | superseded-duplicate | legitimately-excluded-non-conversation) — never bare 'quarantined'. Per-origin census covers codex-session (~/.codex/sessions), claude-code-session (~/.claude/projects), and each export origin vs its export bundle; hooks + browser-capture exempt (no external ground truth).\n6. The queries above are bundled into one re-runnable PASS/FAIL receipt (script or registry check) that the 818fy runbook cites as its go/no-go input.","notes":"SCOPE-COMPLETENESS AUDIT (dissection, 2026-08-03; full numbers in .agent/scratch/reindex-baseline-2026-08-03.md — operator bar: blobstore contains EXACTLY {all relevant blobs from every source incl sidecars/attachments/codex DBs} plus {browser-capture, hooks, marginal}, all valid, no dupes, no weirdness). ADDITIONS this bead's description does not yet name: (1) ORIGINS MISSING FROM SCOPE: hermes-session (356 raws from ~/.hermes incl state.db + observability trajectories), antigravity-session (232, ~/.gemini/antigravity/brain), gemini-cli-session (43, ~/.gemini/tmp) — all locally reconcilable, add them; aistudio-drive (397, native_id all NULL — reconciliation key must be defined). (2) CODEX DATABASES: state_5.sqlite acquired ~9x AS quarantined SESSIONS + goals_1.sqlite + memories_1.sqlite — misrouted artifact-class acquisitions (omsw/1fijp-arm-4); decide their taxonomy before reconciliation counts them as unexplained. (3) SIDECAR TAXONOMY: the live classification is raw_artifacts (17,886 rows: agent_transcript 8,201, agent_sidecar_meta 5,273, coordinator_session_stream 3,915...); history_sidecars is EMPTY (vestigial? verify writer exists). Claude reconciliation denominator: 12,424 jsonl under ~/.claude/projects vs 10,693 distinct native ids + artifacts — reconcile via the artifact taxonomy, not filename. (4) PATH-LEVEL RECONCILIATION IS IMPOSSIBLE: 12,100 raws (42%) carry STALE-ROOT source_paths (~/.local/share/polylogue, dead archive root); GDPR bundles at /realm/data/exports have ZERO raws referencing them (ingested via old inbox paths) — must match content-level (conversation ids). (5) NO-WEIRDNESS ITEMS: 10 REPO/DEV raws from .cache/dev-loop (test artifacts in production blobstore — excise); 960 byte-duplicate (blob_hash,origin,native_id) groups (baseline for the zero-dupes AC); ~1,322 unreferenced blob files on disk (GC candidates, lease-aware sweep only AFTER 0v4tn fixes the refs). (6) BLOCKERS FILED FROM THIS AUDIT: 0v4tn (73,427+1,336 orphaned blob_refs — the GC/reference substrate must be true before 'pristine' means anything), mhx95 (daemon frozen since 07-31 — the drain assumes a working daemon). (7) verified_blob_receipts=0 EVER: add one full blob hash-verification pass (70GB read) to this gate's AC or Phase F — 'completely valid' is currently asserted, not measured; presence samples (300+100) pass.\nPROVENANCE MAPPING (operator question 2026-08-03 re the 42% stale-root paths): source_path is first-acquisition provenance and never rewrites (content-hash idempotency skips re-encounters), so the pre-root-move history is a fossil record — bytes all present, but raw→export-bundle linkage is only string-recoverable (bundle name survives in the inbox path basename, e.g. chatgpt-data-2026-04-23-*). REQUIRED ADDITION to this gate: the content-level reconciliation must PERSIST its raw↔bundle/source mapping (not merely check it) — that mapping IS the durable provenance repair for the 12,100 stale-root rows and the zero-reference GDPR home. Store as reconciliation receipts (existing receipt patterns), not as source_path rewrites (never mutate acquisition evidence).","status":"in_progress","priority":0,"issue_type":"task","assignee":"Sinity","owner":"ezo.dev@gmail.com","created_at":"2026-08-03T08:23:03Z","created_by":"Sinity","updated_at":"2026-08-04T07:52:08Z","started_at":"2026-08-04T07:52:08Z","lease_expires_at":"2026-08-04T07:57:08Z","heartbeat_at":"2026-08-04T07:52:08Z","dependencies":[{"issue_id":"polylogue-r9xsj","depends_on_id":"polylogue-0v4tn","type":"blocks","created_at":"2026-08-04T06:14:12Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-r9xsj","depends_on_id":"polylogue-6753s","type":"blocks","created_at":"2026-08-03T10:23:18Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-r9xsj","depends_on_id":"polylogue-84ake","type":"blocks","created_at":"2026-08-04T06:36:46Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-r9xsj","depends_on_id":"polylogue-lb39z","type":"blocks","created_at":"2026-08-03T10:23:18Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-r9xsj","depends_on_id":"polylogue-lkrc","type":"blocks","created_at":"2026-08-03T10:23:18Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-r9xsj","depends_on_id":"polylogue-nhbvf","type":"blocks","created_at":"2026-08-04T07:07:24Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-r9xsj","depends_on_id":"polylogue-w6hql","type":"blocks","created_at":"2026-08-03T10:23:18Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-r9xsj","depends_on_id":"polylogue-zm4w8","type":"blocks","created_at":"2026-08-04T00:05:14Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":8,"dependent_count":1,"comment_count":0} {"_type":"issue","id":"polylogue-6753s","title":"raw-authority: formalize byte-duplicate supersession - 4,305 quarantined heads (17.6 GiB) are byte-identical to already-indexed content, sitting in limbo instead of a terminal state","description":"Corrected finding 2026-08-03 (operator caught a 4x overstatement in the earlier framing - see corrected notes on lkrc/tnqqt/t0m73). Of 7,200 unindexed logical-source heads, 4,305 (77% of bytes, 17.6 of 22.9 GiB) are BYTE-IDENTICAL (same blob_hash) to a raw that IS already indexed under a different raw_id - re-acquisitions/re-syncs of sessions already in the archive, not missing content. These sit as revision_authority='quarantined' with no logical_source_key (never touched by the reconciler) in perpetual limbo, inflating every 'archive is X% unconverged' measurement and every schema-inference sampling frame (though the sampler's own blob_hash dedup already protects it from THIS specific mass). THIS is the actual 'make the blobstore pristine' fix, not bulk reingestion. AC: (1) query/classify: for every quarantined head with no logical_source_key, check whether its blob_hash matches an already-indexed raw; (2) for matches, resolve to a terminal 'superseded-by-byte-identical-indexed-raw' state (not full reconciler machinery - this is a strictly simpler byte-equality check, no content parsing needed) with a receipt recording which indexed raw_id it duplicates; (3) this must NOT touch/re-parse/re-write the indexed twin - purely marks the duplicate as resolved; (4) leaves the genuinely-novel ~2,895/~5.3GiB heads untouched for lkrc/hjpx's real reconciler to materialize; (5) verify against live archive read-only before/after counts. Related: lkrc, hjpx, t0m73 (I1's classification needs this bucket too).","status":"closed","priority":0,"issue_type":"task","assignee":"Sinity","owner":"ezo.dev@gmail.com","created_at":"2026-08-03T08:21:36Z","created_by":"Sinity","updated_at":"2026-08-03T10:30:25Z","started_at":"2026-08-03T10:30:25Z","closed_at":"2026-08-03T10:30:25Z","close_reason":"Fixed: PR #3624 (9a7e58080). Byte-duplicate supersession classifier + actuator implemented and dry-run-verified against live archive (5,515 duplicates / 17.86 GiB matching the corrected census).","dependency_count":0,"dependent_count":3,"comment_count":0} -{"_type":"issue","id":"polylogue-3hdz2","title":"Whale fixture pack: scale-outlier synthetic corpus + bounds regression","description":"Synthesize outlier-shaped fixtures from the live census (90MB-session shape, multi-million event streams, deep revision chains, giant base64 attachments) and assert bounding behavior: parse succeeds or quarantines LOUDLY with a typed reason. Class-H fragility stops regressing silently; 5iz4's fix gets a permanent guard.","acceptance_criteria":"1. Fixture pack in tests/benchmarks or infra with generation script. 2. Each outlier axis has a bounds assertion. 3. 5iz4's repro is in the pack red-first.","notes":"2026-08-06 campaign graph promotion: this bead is a direct prerequisite or process guard for proof-carrying reindex acceptance. Its implementation cannot substitute for the terminal receipt, but its output is consumed by the campaign ledger and final gate.","status":"open","priority":0,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-08-03T07:41:34Z","created_by":"Sinity","updated_at":"2026-08-06T05:00:40Z","labels":["area:test"],"dependencies":[{"issue_id":"polylogue-3hdz2","depends_on_id":"polylogue-1xc","type":"parent-child","created_at":"2026-08-03T09:41:33Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-3hdz2","depends_on_id":"polylogue-5iz4","type":"relates-to","created_at":"2026-08-03T09:41:33Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-3hdz2","depends_on_id":"polylogue-dhil","type":"relates-to","created_at":"2026-08-03T09:41:33Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":0,"dependent_count":0,"comment_count":0} +{"_type":"issue","id":"polylogue-3hdz2","title":"Whale fixture pack: scale-outlier synthetic corpus + bounds regression","description":"Synthesize outlier-shaped fixtures from the live census (90MB-session shape, multi-million event streams, deep revision chains, giant base64 attachments) and assert bounding behavior: parse succeeds or quarantines LOUDLY with a typed reason. Class-H fragility stops regressing silently; 5iz4's fix gets a permanent guard.","acceptance_criteria":"1. Fixture pack in tests/benchmarks or infra with generation script. 2. Each outlier axis has a bounds assertion. 3. 5iz4's repro is in the pack red-first.","notes":"2026-08-06 campaign graph promotion: this bead is a direct prerequisite or process guard for proof-carrying reindex acceptance. Its implementation cannot substitute for the terminal receipt, but its output is consumed by the campaign ledger and final gate.","status":"closed","priority":0,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-08-03T07:41:34Z","created_by":"Sinity","updated_at":"2026-08-08T15:54:51Z","closed_at":"2026-08-08T15:54:51Z","close_reason":"Satisfied: deterministic generator and manifest cover the 804-revision, 90,822,451-byte, two-million-event, and 12 MiB inline-image axes; focused parser and generator tests pass; the production acquisition, source-remediation, crash/resume, inactive-candidate, promotion, and public/canonical proof passes in 501.04 seconds; devtools verify --quick passes all 24 steps.","labels":["area:test"],"dependencies":[{"issue_id":"polylogue-3hdz2","depends_on_id":"polylogue-1xc","type":"parent-child","created_at":"2026-08-03T09:41:33Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-3hdz2","depends_on_id":"polylogue-5iz4","type":"relates-to","created_at":"2026-08-03T09:41:33Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-3hdz2","depends_on_id":"polylogue-dhil","type":"relates-to","created_at":"2026-08-03T09:41:33Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"polylogue-rrxe4","title":"test harness: hermetic convergence-property loop - generate corpus, ingest, converge, assert registry green; metamorphic properties on top","description":"The master loop unifying zoo (yazae), composer (amrpx), and registry (t0m73) into ONE elegant mechanism: property test = for corpus-program P and ingestion order sigma: build hermetic archive from (P, sigma) via PRODUCTION ingest+convergence (tests/infra/convergence_harness.py already adapts production writers/daemon stages and 'deliberately owns no alternate convergence state machine' - the exact seam), then assert the t0m73 registry green on the result. Every registry predicate thereby gets exercised against synthetic pathological corpora in CI, and every composer pathology is validated end-to-end through real code. METAMORPHIC properties on top (each kills a whole bug class): (1) INGESTION-ORDER INVARIANCE - final archive state equivalent for all sigma (catches the deferred-tail/parent-arrival order-dependence class); (2) INCREMENTAL==BULK - trickle-converged archive equivalent to bulk-rebuilt archive on the same corpus (this is hjwr done RIGHT: as a cheap synthetic property, not a live-archive lane - note on hjwr); (3) IDEMPOTENCE - re-ingest is a no-op; (4) APPEND-PREFIX consistency - ingest(full) == ingest(full_prefix)+ingest(append_delta). Use hypothesis.stateful RuleBasedStateMachine for order/interruption exploration (precedent: test_write_path_state_machine.py, test_fts_identity_state_machine.py); crash-injection at stage boundaries as a later extension. Depends on amrpx (corpus source) + t0m73 (predicates).","design":"DESIGN (2026-08-03; the description carries the mechanism — this fixes the file/seam map): SEAM: tests/infra/convergence_harness.py (already adapts production writers/daemon stages, owns no alternate state machine). LOOP: for corpus-program P (from the zoo/composer, yazae; amrpx closed = corpus source available) and ingestion order sigma: build a hermetic archive via PRODUCTION ingest+convergence, then assert the t0m73 registry green (binding (a) of its four). METAMORPHIC PROPERTIES, each its own test module, priority order: (1) ingestion-order invariance (kills the deferred-tail/parent-arrival class); (2) incremental==bulk (hjwr done right, synthetic); (3) idempotence (re-ingest no-op — hash-skip already exists, this proves it end-to-end); (4) append-prefix consistency (full == prefix+delta). MECHANISM: hypothesis.stateful RuleBasedStateMachine for order/interruption exploration (precedents: test_write_path_state_machine.py, test_fts_identity_state_machine.py); crash-injection at stage boundaries is a later extension, not v1. PITFALLS: archive equivalence needs a canonical comparison (session/message/block rows modulo generation-scoped ids and timestamps — define one comparator, reuse across all four properties); keep corpora small (seconds per example) and rely on Hypothesis shrinking; register slow variants behind --lab, not the default testmon loop. Depends on t0m73 (predicates) — start with the registry subset that exists today rather than waiting for full migration.\n","acceptance_criteria":"1. The property loop exists: (corpus-program P, order sigma) -> hermetic archive via production ingest+convergence (tests/infra/convergence_harness.py seam) -> t0m73 registry asserted green; runs in CI on small corpora.\n2. All four metamorphic properties implemented as separate modules sharing ONE canonical archive-equivalence comparator (also used by 0x7nh): order-invariance, incremental==bulk, idempotence, append-prefix consistency.\n3. Hypothesis stateful machine explores orders/interruptions (precedent files cited in design); failures shrink to minimal corpora.\n4. At least one historical bug class is demonstrably caught: re-introducing a known fixed order-dependence bug makes property (1) fail (anti-vacuity).\n5. Slow variants gated behind --lab; default testmon loop stays fast. Verify: devtools test -k convergence_property.","notes":"Promoted P0 2026-08-03, scope extended per operator direction: this is the closest existing bead to 'the test suite generated from schemas IS the integrity checker, run against the real archive when wanted.' Extend explicitly: corpus size can exceed the real archive, generated corpus is never checked into the repo (only the seeding schemas are, as the corpus's deterministic seeds), test invocations can specify which slice/intersection of the corpus to run. Depends on already-closed polylogue-amrpx (generator) and polylogue-t0m73 (registry) -- both landed, this is unblocked.\nSCOPE CORRECTION (operator-prompted verification, 2026-08-04): this bead's premise partially does not hold yet. Read tests/infra/pathology_composer.py (amrpx's actual output): it ships 6 fixed, non-composable pathology functions and ZERO ingestion-order control. rrxe4's own first metamorphic property (ingestion-order invariance over sigma) has no input to vary sigma OVER -- nothing generates multiple orderings of a corpus today. Filed as its own gap: [new bead, see notes]. rrxe4 cannot fully execute as designed until that lands (or rrxe4 absorbs building minimal order-control itself as part of its own scope, which may be the cheaper path -- rrxe4 is the one consumer that needs it, so building it inline here rather than reopening amrpx is worth considering).","status":"open","priority":0,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-08-03T07:09:11Z","created_by":"Sinity","updated_at":"2026-08-03T22:31:04Z","dependencies":[{"issue_id":"polylogue-rrxe4","depends_on_id":"polylogue-4v2d3","type":"blocks","created_at":"2026-08-06T07:02:25Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-rrxe4","depends_on_id":"polylogue-amrpx","type":"blocks","created_at":"2026-08-03T09:09:12Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-rrxe4","depends_on_id":"polylogue-canonical-snapshot","type":"blocks","created_at":"2026-08-06T07:02:24Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-rrxe4","depends_on_id":"polylogue-ehzfn","type":"blocks","created_at":"2026-08-06T17:13:59Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-rrxe4","depends_on_id":"polylogue-t0m73","type":"blocks","created_at":"2026-08-03T09:09:12Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-rrxe4","depends_on_id":"polylogue-un60n","type":"blocks","created_at":"2026-08-06T07:02:24Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-rrxe4","depends_on_id":"polylogue-yazae","type":"blocks","created_at":"2026-08-06T07:02:24Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":7,"dependent_count":2,"comment_count":0} {"_type":"issue","id":"polylogue-xselt","title":"reindex: stamp fingerprints at bootstrap - 818fy rebuild must write parser+lowering semantics fingerprints per session","description":"Split from polylogue-kea7p (its tiny prerequisite, decoupled from the big differential planner). Add two additive index-tier columns (sessions.parser_fingerprint, sessions.lowering_fingerprint) + per-origin fingerprint derivation (origin_specs home), and have the 818fy blue-green rebuild WRITE them for every session. Why P0 and why it gates 818fy: without stamps, the next semantic fix after the reindex forces another FULL rebuild (kea7p soundness analysis: hash-skip alone is unsound; the skip oracle needs fingerprint conjunction). With stamps, every future parser/lowering fix becomes an origin-scoped reparse and the whole 'we must root out ALL bugs before reindexing' pressure collapses to 'fix stamp-poisoners + durable-corrupters before; everything else cheaply after'. Blocked by the stamp-poisoner bugs (7zp4 NFC gaps, gysk3 position-derived identity) - fingerprints computed over buggy identity/hash semantics would bootstrap poisoned stamps. Scope: columns + derivation + rebuild wiring + registry check (every session row carries current-format stamps). NOT in scope: the differential planner/skip logic (stays kea7p).","design":"DESIGN (2026-08-03, grounded against current master: SOURCE=24, INDEX=57):\n\nPREMISE UPDATE: both declared blockers are now closed (7zp4 closed 2026-08-03, was already fixed at index-v46 commit 5e23e6abf; gysk3 closed via PR #3604). The stamp-poisoner gate on this bead is satisfied EXCEPT that fsgdd's K-class list also names 0qfy/uqwd (vintage-volatile comparison axes) as stamp-poisoners; coordinator should confirm whether those two must also gate this bead before stamps bootstrap. Otherwise this bead is ready to implement.\n\nWHAT GETS STAMPED (two additive nullable TEXT columns on index-tier `sessions`, archive_tiers/index.py):\n- sessions.parser_fingerprint: SHA-256 hex over the ORIGIN-SCOPED parser semantics — the normalized source of the parser module set that produced this session's parse. Derivation home is polylogue/sources/origin_specs.py (OriginSpec already carries the prose `semantic_reparse` field per origin; add a `parser_fingerprint()` that hashes the origin's declared parser surface). Mechanism precedent: docs/plans/classifier-fingerprints.json (polylogue-gucv gate) already computes per-function source fingerprints — reuse that function-source-hash machinery, do not invent a second normalizer.\n- sessions.lowering_fingerprint: SHA-256 over the SHARED lowering path semantics (sources/dispatch.py detect_provider/_lower_payload_specs/_parse_lowered_spec + pipeline/ids.py identity/hash functions). One global value per code vintage, same for all origins; a lowering change invalidates everything, a parser change invalidates one origin. This split is exactly what makes kea7p's T0 skip origin-scoped.\n\nNAMING HAZARD: source-tier census tables already have a `parser_fingerprint` column (revision_backfill.py:392 `_resource_blocked_parser_fingerprint`) — that fingerprints the RESOURCE ENVELOPE (max_payload_bytes), not parser semantics. Different concept, different tier. Document the distinction at both sites; do not unify them.\n\nWRITE POINT: write_parsed_session_to_archive (storage/sqlite/archive_tiers/write.py:307) — the single choke point both live incremental ingest and full raw replay/reindex go through, so ordinary ingest and the 818fy rebuild stamp identically for free. Stamps MUST be written in the same transaction as the session row, never backfilled asynchronously — an unstamped-but-indexed session makes kea7p's skip oracle unsound.\n\nSCHEMA MECHANICS: INDEX_SCHEMA_VERSION bump + IndexDeltaDeclaration in storage/sqlite/lifecycle.py. Correct class: SEMANTIC_REPARSE (the column's value is parser-semantics-dependent by definition; only reparse populates it). That routes existing archives to the full rebuild — which is precisely 818fy; land this bump so the one planned rebuild bootstraps the stamps, rather than paying a second rebuild later. `devtools lab policy schema-versioning` enforces the declaration.\n\nSKIP-PREDICATE CONTRACT (kea7p consumes, xselt guarantees): a session is reparse-candidate iff parser_fingerprint != current(origin) OR lowering_fingerprint != current() OR its raw head is unindexed; skippable otherwise. xselt's deliverable is that the conjunction inputs exist and are trustworthy on every row; the planner itself stays in kea7p.\n\nREGISTRY CHECK: add an ArchiveVerificationCheckSpec to ARCHIVE_VERIFICATION_CHECKS (polylogue/maintenance/archive_verification.py) asserting every sessions row carries non-NULL, hex-shaped fingerprints, with per-origin distinct-value counts as evidence (a healthy post-rebuild archive has exactly one parser_fingerprint per origin and one lowering_fingerprint globally). This is the \"registry check\" AC item and lives in both verification planes per 60gzo.\n","acceptance_criteria":"1. Additive nullable columns sessions.parser_fingerprint + sessions.lowering_fingerprint exist in the index-tier DDL (archive_tiers/index.py), with an INDEX_SCHEMA_VERSION bump carrying a declared IndexDeltaDeclaration (SEMANTIC_REPARSE) in storage/sqlite/lifecycle.py; `devtools lab policy schema-versioning` passes.\n2. write_parsed_session_to_archive stamps both columns in the same transaction as the session row, on both ordinary ingest and raw replay — proven by a unit test through the real write path (not a mock), verify: `devtools test -k fingerprint`.\n3. Per-origin derivation lives in origin_specs.py and reuses the classifier-fingerprints function-source-hash mechanism; a unit test proves the derived fingerprint CHANGES when a parser-semantics input changes and is STABLE across process restarts.\n4. After rebuilding a fixture archive, 100% of sessions rows carry both stamps, one distinct parser_fingerprint per origin, one lowering_fingerprint globally — enforced by a new ArchiveVerificationCheckSpec in ARCHIVE_VERIFICATION_CHECKS with a red-twin fixture test.\n5. Explicitly NOT in scope: the differential planner/skip logic (stays polylogue-kea7p); this bead only guarantees the skip oracle's inputs exist and are trustworthy.","notes":"2026-08-03 (reindex-gate-hunt, team-lead): PREMISE CORRECTION. The premise-update claiming \"both declared blockers are now closed (7zp4, gysk3) ... ready to implement\" is FALSE as stated: gysk3 closure fixed only the identity-hash call-site symptom and explicitly deferred the root cause (18 parser call sites baking positional strings into provider_message_id; the \"acks\" registry docs/plans/position-derived-identity-acks.json is a lint-suppression list whose every entry references the closed gysk3, not a tracking bead). New K-class blocker polylogue-slshy (blocks edge added) carries that root cause and MUST land before this bead writes bootstrap stamps — stamps computed over position-derived identity would poison every future differential reparse. Adjudicated K by the 2026-08-03 gate-hunt; ruling and evidence on that bead.\n2026-08-04 prerequisite update: polylogue-slshy landed as PR #3730. Current-master audit finds no position-derived provider_message_id assignments in parser routes, and docs/plans/position-derived-identity-acks.json is empty. The parser identity precondition for bootstrap stamps is satisfied; retain the remaining explicit xselt dependencies.\n2026-08-04 correction: slshy was reopened after an adversarial review found three identity defects in merged PR #3730. Do not implement or merge stamps until the reopened slshy repair is merged and re-verified.","status":"open","priority":0,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-08-03T06:24:45Z","created_by":"Sinity","updated_at":"2026-08-04T06:30:07Z","dependencies":[{"issue_id":"polylogue-xselt","depends_on_id":"polylogue-0qfy","type":"blocks","created_at":"2026-08-03T13:19:29Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-xselt","depends_on_id":"polylogue-7zp4","type":"blocks","created_at":"2026-08-03T08:24:57Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-xselt","depends_on_id":"polylogue-gysk3","type":"blocks","created_at":"2026-08-03T08:24:57Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-xselt","depends_on_id":"polylogue-slshy","type":"blocks","created_at":"2026-08-03T14:17:57Z","created_by":"Sinity","metadata":"{}"},{"issue_id":"polylogue-xselt","depends_on_id":"polylogue-uqwd","type":"blocks","created_at":"2026-08-03T13:19:29Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":5,"dependent_count":2,"comment_count":0} {"_type":"issue","id":"polylogue-4v2d3","title":"verification: oracle-integrity lint - every tested symbol must be production-reachable (dead-engine test suites certify nothing)","description":"From test-class taxonomy 2026-08-03. Two instances found in one day: test_topology_cycle_rejection.py + parts of test_delegations_view.py pin queries/session_links.py (zero production imports - H4/4ts.10); test_source_laws.py pins parse_drive_payload (zero production callers - M5/194qk). Both suites stay green forever regardless of live behavior. Class design: an import-graph reachability lint - for each test module, the primary tested symbols must be reachable from a production entry point (cli/mcp/daemon/api roots); a test whose entire target set is production-unreachable fails the lint with 'certifies dead code'. Needs a small allowlist for deliberate infra (tests/infra, fixtures). Cheap to approximate: intersect grep-derived test imports with the layering/topology import graph devtools already builds. This is the mechanical form of the anti-vacuity doctrine and directly attacks the 1:1-LoC-coverage paradox.","acceptance_criteria":"1. Every test module whose primary target is production behavior is reachable from a production entrypoint, or is explicitly classified as test infrastructure.\n2. Hermetic tests cannot read real user/archive paths unless an explicit production-safe fixture boundary is declared.\n3. A controlled dead-symbol mutation makes the reachability check fail, and a controlled path escape makes the hermeticity check fail.\n4. Focused devtools tests and devtools verify --quick pass.","notes":"2026-08-03: scope explicitly includes the HERMETICITY guard (kmqwm class - tests must not read real user paths like ~/.codex/sessions; enforce via fixture-env guard or lint), same family as reachability: both are 'the test exercises something other than what it claims'.\n2026-08-06 campaign graph promotion: this bead is a direct prerequisite or process guard for proof-carrying reindex acceptance. Its implementation cannot substitute for the terminal receipt, but its output is consumed by the campaign ledger and final gate.\nCodex closed-PR audit 2026-08-06: merged PR #3836 review found dead-engine and hermeticity gaps remain. This is a campaign prerequisite and cannot close from green tests that target production-unreachable symbols.","status":"open","priority":0,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-08-03T05:47:34Z","created_by":"Sinity","updated_at":"2026-08-06T19:24:17Z","dependency_count":0,"dependent_count":1,"comment_count":0} diff --git a/devtools/pytest_timeout_overrides.toml b/devtools/pytest_timeout_overrides.toml index 952c941ef0..0c59f1ee26 100644 --- a/devtools/pytest_timeout_overrides.toml +++ b/devtools/pytest_timeout_overrides.toml @@ -10,5 +10,5 @@ rationale = "The single-process coverage gate needs a bounded diagnostic budget [[exception]] path = "tests/unit/scenarios/test_codex_804_live_proof.py" -value = 420 -rationale = "The sanitized 804-revision proof runs source remediation plus an interrupted and resumed production replay; the measured replay takes about 311 seconds and needs a bounded incident-scale budget." +value = 900 +rationale = "The sanitized 804-revision proof generates and acquires the outlier corpus, runs source remediation, then exercises pre-checkpoint failure, interrupted replay, fresh-process resume, and promotion under one bounded incident-scale budget." diff --git a/polylogue/daemon/cli.py b/polylogue/daemon/cli.py index 9d665c2a46..4913760c7f 100644 --- a/polylogue/daemon/cli.py +++ b/polylogue/daemon/cli.py @@ -17,7 +17,7 @@ from datetime import UTC, datetime from http.server import ThreadingHTTPServer from pathlib import Path -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, Final import click @@ -66,6 +66,10 @@ daemon_write_coordinator, ) from polylogue.logging import configure_logging, get_logger +from polylogue.product.raw_authority import ( + RAW_MATERIALIZATION_ORDINARY_BLOB_LIMIT_BYTES, + RAW_MATERIALIZATION_WHALE_BLOB_LIMIT_BYTES, +) from polylogue.sources.live import LiveWatcher, WatchSource from polylogue.sources.live.sqlite_locking import is_transient_sqlite_lock from polylogue.sources.live.watcher import INBOX_SOURCE_SUFFIXES, default_sources @@ -106,7 +110,7 @@ # -- consuming a prefetch hit costs a receipt write, not a reparse, so this # does not meaningfully extend the writer hold. _RAW_MATERIALIZATION_PARSE_STAGE_WARM_LIMIT = 64 -_RAW_MATERIALIZATION_DAEMON_BLOB_LIMIT_BYTES = 64 * 1024 * 1024 +_RAW_MATERIALIZATION_DAEMON_BLOB_LIMIT_BYTES: Final = RAW_MATERIALIZATION_ORDINARY_BLOB_LIMIT_BYTES # polylogue-de2a: declared, enforced ceiling on how long ONE ordinary trickle # pass may hold the process-wide writer coordinator. Live evidence showed # ``_RAW_MATERIALIZATION_CONVERGENCE_BATCH_LIMIT`` alone did not bound hold @@ -157,7 +161,7 @@ # contention). 8 GiB comfortably covers the live witness (codex:019f49d8, # 6.33GB/788 raws) with headroom; override via # ``raw_authority_whale_payload_bytes`` / POLYLOGUE_RAW_AUTHORITY_WHALE_PAYLOAD_BYTES. -_RAW_MATERIALIZATION_WHALE_BLOB_LIMIT_BYTES = 8 * 1024 * 1024 * 1024 +_RAW_MATERIALIZATION_WHALE_BLOB_LIMIT_BYTES: Final = RAW_MATERIALIZATION_WHALE_BLOB_LIMIT_BYTES _RAW_MATERIALIZATION_LIVE_SPOOL_BACKOFF_SECONDS = 60 # A spool file younger than this is in the live route's normal debounce/ # batch flow, not stalled; only older cursor-less files park the conveyor. diff --git a/polylogue/product/raw_authority.py b/polylogue/product/raw_authority.py index 071c65c489..7a86e1c04d 100644 --- a/polylogue/product/raw_authority.py +++ b/polylogue/product/raw_authority.py @@ -9,7 +9,7 @@ from dataclasses import dataclass from pathlib import Path -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, Final from polylogue.config import Config from polylogue.core.json import JSONDocument @@ -18,6 +18,10 @@ from polylogue.sources.revision_backfill import RawParsePrefetchCache +RAW_MATERIALIZATION_ORDINARY_BLOB_LIMIT_BYTES: Final = 64 * 1024 * 1024 +RAW_MATERIALIZATION_WHALE_BLOB_LIMIT_BYTES: Final = 8 * 1024 * 1024 * 1024 + + @dataclass(frozen=True, slots=True) class RawMaterializationCounts: """Separate units produced by one bounded maintenance pass. diff --git a/polylogue/sources/parsers/codex.py b/polylogue/sources/parsers/codex.py index 772edfeb31..d236c0ab7d 100644 --- a/polylogue/sources/parsers/codex.py +++ b/polylogue/sources/parsers/codex.py @@ -4,12 +4,15 @@ import hashlib import json +import pickle import re import shlex +import tempfile from collections import defaultdict -from collections.abc import Iterable, Sequence +from collections.abc import Iterable, Iterator, Sequence from dataclasses import dataclass, replace from datetime import datetime +from typing import BinaryIO from pydantic import ValidationError @@ -78,6 +81,22 @@ _STRUCTURAL_BYTE_KEYS = frozenset({"bytes", "byte_count", "bytes_written", "size_bytes", "written_bytes"}) +@dataclass(slots=True) +class _PickleRecordReplay: + """Re-iterable disk spool for preserving parser lookahead without retaining records.""" + + handle: BinaryIO + + def __iter__(self) -> Iterator[object]: + self.handle.seek(0) + while True: + try: + # This reads only the private spool written immediately above. + yield pickle.load(self.handle) + except EOFError: + return + + @dataclass(frozen=True, slots=True) class _CodexExecChildType: kind: str @@ -1555,7 +1574,7 @@ def _response_inner_record(item: object) -> dict[str, object] | None: return inner if inner is not None and not _is_message(inner) else None -def _code_mode_exec_envelopes(records: Sequence[object]) -> dict[int, _CodexExecEnvelope]: +def _code_mode_exec_envelopes(records: Iterable[object]) -> dict[int, _CodexExecEnvelope]: call_occurrences: dict[str, list[tuple[int, _CodexExecEnvelope]]] = defaultdict(list) output_occurrences: dict[str, list[tuple[int, dict[str, object]]]] = defaultdict(list) envelopes_by_record: dict[int, _CodexExecEnvelope] = {} @@ -2160,11 +2179,34 @@ def _sanitize_codex_data_url(value: str) -> str: return value header, encoded = value.split(",", 1) mime = header.removeprefix("data:").split(";", 1)[0] or "image/unknown" - digest = hashlib.sha256(encoded.encode("ascii", errors="ignore")).hexdigest() - approx_bytes = (len(encoded.rstrip("=")) * 3) // 4 + digest_builder = hashlib.sha256() + for offset in range(0, len(encoded), 1024 * 1024): + digest_builder.update(encoded[offset : offset + 1024 * 1024].encode("ascii", errors="ignore")) + digest = digest_builder.hexdigest() + padding = 2 if encoded.endswith("==") else 1 if encoded.endswith("=") else 0 + approx_bytes = max(0, (len(encoded) * 3) // 4 - padding) return f"" +def _codex_inline_image_blocks(content: object) -> tuple[ParsedContentBlock, ...]: + """Return typed, bounded evidence for inline images without authored prose inflation.""" + if not isinstance(content, list): + return () + blocks: list[ParsedContentBlock] = [] + for item in content: + if not isinstance(item, dict) or item.get("type") not in {"input_image", "image"}: + continue + image_url = item.get("image_url") + if not isinstance(image_url, str): + continue + summary = _sanitize_codex_data_url(image_url) + if summary != image_url: + header = image_url.split(",", 1)[0] + mime = header.removeprefix("data:").split(";", 1)[0] or "image/unknown" + blocks.append(ParsedContentBlock(type=BlockType.IMAGE, text=summary, media_type=mime)) + return tuple(blocks) + + def _message_signature(role: Role | str, text: str | None) -> tuple[str, str]: role_value = role.value if isinstance(role, Role) else str(role) return (role_value, " ".join((text or "").split())) @@ -2337,7 +2379,7 @@ def is_supported_session_stream(payload: Sequence[object]) -> bool: return has_session_header or has_direct_record or has_envelope_record -def _parse_records(records: Iterable[object], fallback_id: str) -> ParsedSession: +def _parse_records(records: Iterable[object], fallback_id: str, *, _reiterable: bool = False) -> ParsedSession: """Parse Codex JSONL session file using typed CodexRecord model. Supports two format generations via CodexRecord.format_type: @@ -2350,9 +2392,17 @@ def _parse_records(records: Iterable[object], fallback_id: str) -> ParsedSession - text_content: Extracted text from any format - format_type: Detected format generation """ - record_list = list(records) - code_mode_envelopes = _code_mode_exec_envelopes(record_list) - response_signatures = _response_message_signatures(record_list) + if not isinstance(records, Sequence) and not _reiterable: + # The parser needs two lookahead-derived indexes before the materializing + # pass. Persist a private replay spool so a multi-million-record JSONL + # stream stays bounded by one decoded record instead of list(records). + with tempfile.TemporaryFile(mode="w+b") as spool: + for item in records: + pickle.dump(item, spool, protocol=pickle.HIGHEST_PROTOCOL) + return _parse_records(_PickleRecordReplay(spool), fallback_id, _reiterable=True) + + code_mode_envelopes = _code_mode_exec_envelopes(records) + response_signatures = _response_message_signatures(records) messages: list[ParsedMessage] = [] session_events: list[ParsedSessionEvent] = [] session_id = fallback_id @@ -2393,7 +2443,7 @@ def _parse_records(records: Iterable[object], fallback_id: str) -> ParsedSession session_model_provider: str | None = None session_developer_instructions: str | None = None - for idx, item in enumerate(record_list, start=1): + for idx, item in enumerate(records, start=1): record = _dict_record(item) if record is None: continue @@ -2750,12 +2800,15 @@ def _parse_records(records: Iterable[object], fallback_id: str) -> ParsedSession raw_role = _effective_role(message_record) content = _effective_content(message_record) text = extract_codex_text(content) + inline_image_blocks = _codex_inline_image_blocks(content) timestamp_pair = parse_timestamp_pair(_message_timestamp(record, message_record)) timestamp = timestamp_pair[1] if timestamp_pair is not None else None content_blocks = content_blocks_from_segments(content) + content_blocks.extend(inline_image_blocks) has_structured = any( - cb.type in (BlockType.TOOL_USE, BlockType.TOOL_RESULT, BlockType.THINKING) for cb in content_blocks + cb.type in (BlockType.TOOL_USE, BlockType.TOOL_RESULT, BlockType.THINKING, BlockType.IMAGE) + for cb in content_blocks ) if not raw_role or raw_role == "unknown": continue diff --git a/tests/infra/generate_whale_fixture.py b/tests/infra/generate_whale_fixture.py new file mode 100644 index 0000000000..ecb5f41e36 --- /dev/null +++ b/tests/infra/generate_whale_fixture.py @@ -0,0 +1,23 @@ +"""Generate the private-data-free Codex whale fixture pack on disk.""" + +from __future__ import annotations + +import argparse +from collections.abc import Sequence +from pathlib import Path + +from tests.infra.whale_fixtures import write_codex_whale_fixture_pack + + +def main(argv: Sequence[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("output_dir", type=Path, help="directory for codex-whale.jsonl and manifest.json") + args = parser.parse_args(argv) + source_path, manifest_path = write_codex_whale_fixture_pack(args.output_dir) + print(source_path) + print(manifest_path) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/infra/reindex_campaign.py b/tests/infra/reindex_campaign.py index 4f8f7eae84..0561d0c641 100644 --- a/tests/infra/reindex_campaign.py +++ b/tests/infra/reindex_campaign.py @@ -25,6 +25,7 @@ from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root from tests.infra.source_builders import SyntheticAntigravityLanguageServerClient +from tests.infra.whale_fixtures import WHALE_FIXTURE_DIMENSIONS REINDEX_CAMPAIGN_REQUIRED_ORIGINS = frozenset( { @@ -68,6 +69,7 @@ class ReindexCampaignManifest: fts_queries: tuple[str, ...] origin_session_counts: tuple[tuple[str, int], ...] denominators: tuple[tuple[str, int], ...] + fixture_dimensions: tuple[tuple[str, int | str], ...] def denominator(self, name: str) -> int: try: @@ -364,6 +366,7 @@ def _campaign_manifest( fts_queries=("generated", "fixture", "failed"), origin_session_counts=origin_session_counts, denominators=denominators, + fixture_dimensions=WHALE_FIXTURE_DIMENSIONS.manifest_dimensions(), ) manifest.assert_positive() if parser_failure_residuals < len(parser_failure_raw_ids): diff --git a/tests/infra/whale_fixtures.py b/tests/infra/whale_fixtures.py new file mode 100644 index 0000000000..03b254ab53 --- /dev/null +++ b/tests/infra/whale_fixtures.py @@ -0,0 +1,351 @@ +"""Deterministic, private-data-free scale outlier fixtures. + +The builders in this module own wire evidence only. Acquisition, parsing, +materialization, and replay remain the production services exercised by the +consuming tests. +""" + +from __future__ import annotations + +import asyncio +import base64 +import hashlib +import json +import os +import shutil +from collections.abc import Iterator +from dataclasses import dataclass +from pathlib import Path +from typing import BinaryIO, Final + +from polylogue.product.raw_authority import ( + RAW_MATERIALIZATION_ORDINARY_BLOB_LIMIT_BYTES, + RAW_MATERIALIZATION_WHALE_BLOB_LIMIT_BYTES, +) + +_TERMINAL_WIRE_BYTES: Final = 90_822_451 +_REVISION_COUNT: Final = 804 +_STREAM_EVENT_COUNT: Final = 2_000_000 +_GIANT_ATTACHMENT_RAW_BYTES: Final = 12 * 1024 * 1024 +_NEAR_TERMINAL_PREDECESSOR_BYTES: Final = 32 * 1024 * 1024 +_GENERATOR_CHUNK_BYTES: Final = 1024 * 1024 + + +@dataclass(frozen=True, slots=True) +class WhaleFixtureDimensions: + """Expected dimensions and the resource envelope for one fixture pack.""" + + fixture_id: str = "codex-whale-bounds-v2" + revision_count: int = _REVISION_COUNT + terminal_wire_bytes: int = _TERMINAL_WIRE_BYTES + near_terminal_predecessor_bytes: int = _NEAR_TERMINAL_PREDECESSOR_BYTES + stream_event_count: int = _STREAM_EVENT_COUNT + giant_attachment_raw_bytes: int = _GIANT_ATTACHMENT_RAW_BYTES + ordinary_blob_limit_bytes: int = RAW_MATERIALIZATION_ORDINARY_BLOB_LIMIT_BYTES + whale_blob_limit_bytes: int = RAW_MATERIALIZATION_WHALE_BLOB_LIMIT_BYTES + + def manifest_dimensions(self) -> tuple[tuple[str, int | str], ...]: + return ( + ("fixture_id", self.fixture_id), + ("revision_count", self.revision_count), + ("terminal_wire_bytes", self.terminal_wire_bytes), + ("near_terminal_predecessor_bytes", self.near_terminal_predecessor_bytes), + ("stream_event_count", self.stream_event_count), + ("giant_attachment_raw_bytes", self.giant_attachment_raw_bytes), + ("ordinary_blob_limit_bytes", self.ordinary_blob_limit_bytes), + ("whale_blob_limit_bytes", self.whale_blob_limit_bytes), + ) + + +WHALE_FIXTURE_DIMENSIONS = WhaleFixtureDimensions() + + +def _wire_target_bytes(revision: int, dimensions: WhaleFixtureDimensions) -> int: + if revision < dimensions.revision_count - 4: + return 4_096 + revision * 640 + return { + dimensions.revision_count - 4: dimensions.near_terminal_predecessor_bytes // 4, + dimensions.revision_count - 3: dimensions.near_terminal_predecessor_bytes // 2, + dimensions.revision_count - 2: dimensions.near_terminal_predecessor_bytes, + dimensions.revision_count - 1: dimensions.terminal_wire_bytes, + }[revision] + + +def _write_record(handle: BinaryIO, record: dict[str, object]) -> int: + encoded = json.dumps(record, sort_keys=True, separators=(",", ":")).encode("utf-8") + b"\n" + return handle.write(encoded) + + +def _write_base64_pattern(handle: BinaryIO, byte_count: int) -> int: + """Write deterministic base64 in bounded chunks, returning encoded bytes.""" + pattern = b"POLYLOGUE_PRIVATE_FREE_ATTACHMENT_0001\n" + remaining = byte_count + encoded_bytes = 0 + while remaining: + chunk_size = min(3 * 1024 * 1024, remaining) + chunk_size -= chunk_size % 3 + if chunk_size == 0: + chunk_size = remaining + raw_chunk = (pattern * ((chunk_size // len(pattern)) + 1))[:chunk_size] + encoded = base64.b64encode(raw_chunk) + handle.write(encoded) + encoded_bytes += len(encoded) + remaining -= chunk_size + return encoded_bytes + + +def _write_repeated_byte(handle: BinaryIO, value: bytes, byte_count: int) -> None: + """Write one repeated byte without allocating an outlier-sized buffer.""" + if len(value) != 1: + raise ValueError("value must contain exactly one byte") + chunk = value * min(byte_count, _GENERATOR_CHUNK_BYTES) + remaining = byte_count + while remaining: + current = chunk if remaining >= len(chunk) else chunk[:remaining] + written = handle.write(current) + if written <= 0: + raise OSError("fixture writer made no progress") + remaining -= written + + +def _file_sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(_GENERATOR_CHUNK_BYTES), b""): + digest.update(chunk) + return digest.hexdigest() + + +def _write_terminal_attachment_record(handle: BinaryIO, session_id: str, byte_count: int) -> None: + record = { + "type": "response_item", + "payload": { + "type": "message", + "id": f"{session_id}-terminal-803", + "role": "assistant", + "content": [ + {"type": "output_text", "text": "sanitized terminal response revision 803"}, + {"type": "input_image", "image_url": ""}, + ], + }, + } + encoded = json.dumps(record, sort_keys=True, separators=(",", ":")).encode("utf-8") + prefix, suffix = encoded.split(b'""', maxsplit=1) + handle.write(prefix + b'"data:image/png;base64,') + _write_base64_pattern(handle, byte_count) + handle.write(b'"' + suffix + b"\n") + + +def _write_padding_record(handle: BinaryIO, revision: int, target_bytes: int, current_bytes: int) -> None: + template = json.dumps( + {"payload": {"padding": "", "type": "token_count"}, "revision": revision, "type": "response_item"}, + sort_keys=True, + separators=(",", ":"), + ).encode("utf-8") + prefix, suffix = template.split(b'""', maxsplit=1) + padding_size = target_bytes - current_bytes - len(prefix) - len(suffix) - 3 + if padding_size <= 0: + raise AssertionError(f"padding underflow at revision {revision}: {padding_size}") + handle.write(prefix + b'"') + _write_repeated_byte(handle, b"x", padding_size) + handle.write(b'"' + suffix + b"\n") + + +@dataclass(frozen=True, slots=True) +class CodexRevisionChainFixture: + """A single-path 804-revision Codex chain built one snapshot at a time.""" + + dimensions: WhaleFixtureDimensions = WHALE_FIXTURE_DIMENSIONS + session_native_id: str = "codex-sanitized-804-session" + + def write_revision(self, source_path: Path, revision: int) -> int: + if not 0 <= revision < self.dimensions.revision_count: + raise ValueError(f"revision must be in [0, {self.dimensions.revision_count})") + source_path.parent.mkdir(parents=True, exist_ok=True) + temporary_path = source_path.with_name(f".{source_path.name}.revision-{revision:03d}.tmp") + previous_size = source_path.stat().st_size if revision and source_path.exists() else 0 + try: + with temporary_path.open("wb") as handle: + if revision: + with source_path.open("rb") as previous: + shutil.copyfileobj(previous, handle, length=_GENERATOR_CHUNK_BYTES) + else: + _write_record( + handle, + { + "type": "session_meta", + "payload": { + "id": self.session_native_id, + "timestamp": "2026-07-31T04:25:20Z", + "cwd": "/sanitized/codex-804", + }, + }, + ) + for message_index, role in enumerate(("user", "assistant")): + _write_record( + handle, + { + "type": "response_item", + "timestamp": "2026-07-31T04:25:20Z", + "payload": { + "type": "message", + "id": f"{self.session_native_id}-message-{message_index}", + "role": role, + "content": [ + { + "type": "output_text" if role == "assistant" else "input_text", + "text": f"sanitized incident witness {role} baseline", + } + ], + }, + }, + ) + if revision in {1, 800, 801, 802}: + _write_record( + handle, + { + "type": "response_item", + "timestamp": f"2026-07-31T04:25:{20 + revision % 40:02d}Z", + "payload": { + "type": "message", + "id": f"{self.session_native_id}-milestone-{revision:03d}", + "role": "assistant", + "content": [ + { + "type": "output_text", + "text": f"sanitized parsed milestone revision {revision}", + } + ], + }, + }, + ) + if revision == self.dimensions.revision_count - 1: + _write_record( + handle, + { + "type": "compacted", + "payload": { + "message": "sanitized compaction summary at the whale boundary", + "replacement_history": [{"role": "user", "content": "sanitized prior context"}], + }, + }, + ) + _write_terminal_attachment_record( + handle, + self.session_native_id, + self.dimensions.giant_attachment_raw_bytes, + ) + current_bytes = handle.tell() + target_bytes = _wire_target_bytes(revision, self.dimensions) + if target_bytes <= max(previous_size, current_bytes): + raise AssertionError( + f"revision {revision} is not strictly larger: previous={previous_size}, " + f"current={current_bytes}, target={target_bytes}" + ) + _write_padding_record(handle, revision, target_bytes, current_bytes) + final_size = handle.tell() + if final_size != target_bytes: + raise AssertionError(f"revision {revision} has {final_size} bytes, expected {target_bytes}") + os.replace(temporary_path, source_path) + except BaseException: + temporary_path.unlink(missing_ok=True) + raise + return final_size + + def iter_revisions(self, source_path: Path) -> Iterator[tuple[int, int, str]]: + """Write one revision, yielding it before the next replaces the file.""" + for revision in range(self.dimensions.revision_count): + size = self.write_revision(source_path, revision) + yield revision, size, _file_sha256(source_path) + + def write_manifest(self, path: Path, sizes: tuple[int, ...], sha256s: tuple[str, ...]) -> Path: + if len(sizes) != self.dimensions.revision_count or len(sha256s) != self.dimensions.revision_count: + raise AssertionError("revision manifest does not cover every revision") + if any(len(value) != 64 for value in sha256s): + raise AssertionError("revision manifest contains an invalid SHA-256 digest") + payload = { + "fixture_id": self.dimensions.fixture_id, + "dimensions": dict(self.dimensions.manifest_dimensions()), + "session_native_id": self.session_native_id, + "revision_sizes": list(sizes), + "revision_sha256": list(sha256s), + "terminal_features": ["compaction", "giant-base64-attachment", "codex-stream-dispatch"], + } + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload, sort_keys=True, indent=2) + "\n", encoding="utf-8") + return path + + +def write_codex_whale_fixture_pack( + output_dir: Path, + fixture: CodexRevisionChainFixture = CodexRevisionChainFixture(), +) -> tuple[Path, Path]: + """Generate the final wire snapshot and its complete revision manifest.""" + output_dir.mkdir(parents=True, exist_ok=True) + source_path = output_dir / "codex-whale.jsonl" + observations = tuple(fixture.iter_revisions(source_path)) + sizes = tuple(size for _revision, size, _sha256 in observations) + sha256s = tuple(sha256 for _revision, _size, sha256 in observations) + manifest_path = fixture.write_manifest(output_dir / "manifest.json", sizes, sha256s) + return source_path, manifest_path + + +def acquire_codex_revision_chain( + archive_root: Path, + fixture: CodexRevisionChainFixture, + source_path: Path, +) -> tuple[tuple[str, ...], tuple[int, ...], tuple[str, ...]]: + """Acquire all snapshots through ``AcquisitionService`` with one live path.""" + from polylogue.config import Source + from polylogue.pipeline.services.acquisition import AcquisitionService + from polylogue.storage.sqlite import SQLiteBackend + + async def _run() -> tuple[tuple[str, ...], tuple[int, ...], tuple[str, ...]]: + backend = SQLiteBackend(db_path=archive_root / "index.db") + try: + service = AcquisitionService(backend) + raw_ids: list[str] = [] + sizes: list[int] = [] + sha256s: list[str] = [] + for revision, size, sha256 in fixture.iter_revisions(source_path): + result = await service.acquire_sources([Source(name="codex", path=source_path)]) + if result.errors: + raise AssertionError( + f"acquisition reported {result.errors} error(s) at revision {revision}: {result.counts}" + ) + raw_ids.extend(result.raw_ids) + sizes.append(size) + sha256s.append(sha256) + return tuple(raw_ids), tuple(sizes), tuple(sha256s) + finally: + await backend.close() + + return asyncio.run(_run()) + + +def multi_million_codex_stream( + dimensions: WhaleFixtureDimensions = WHALE_FIXTURE_DIMENSIONS, +) -> Iterator[dict[str, object]]: + """Yield a million-scale Codex stream without allocating each state row.""" + yield {"type": "session_meta", "payload": {"id": "codex-stream-million", "timestamp": "2026-08-06T00:00:00Z"}} + yield { + "type": "response_item", + "payload": { + "type": "message", + "id": "codex-stream-million-message", + "role": "user", + "content": [{"type": "input_text", "text": "sanitized streaming boundary"}], + }, + } + for sequence in range(dimensions.stream_event_count): + yield {"record_type": "state", "sequence": sequence} + + +__all__ = [ + "CodexRevisionChainFixture", + "WHALE_FIXTURE_DIMENSIONS", + "WhaleFixtureDimensions", + "acquire_codex_revision_chain", + "multi_million_codex_stream", + "write_codex_whale_fixture_pack", +] diff --git a/tests/unit/infra/test_whale_fixtures.py b/tests/unit/infra/test_whale_fixtures.py new file mode 100644 index 0000000000..ef9e45b303 --- /dev/null +++ b/tests/unit/infra/test_whale_fixtures.py @@ -0,0 +1,106 @@ +"""Contracts for the shared scale outlier fixture pack.""" + +from __future__ import annotations + +import json +from pathlib import Path + +from tests.infra.whale_fixtures import ( + WHALE_FIXTURE_DIMENSIONS, + CodexRevisionChainFixture, + WhaleFixtureDimensions, + _write_padding_record, + multi_million_codex_stream, + write_codex_whale_fixture_pack, +) + + +def test_whale_fixture_manifest_pins_all_outlier_axes() -> None: + dimensions = dict(WHALE_FIXTURE_DIMENSIONS.manifest_dimensions()) + + assert dimensions == { + "fixture_id": "codex-whale-bounds-v2", + "revision_count": 804, + "terminal_wire_bytes": 90_822_451, + "near_terminal_predecessor_bytes": 32 * 1024 * 1024, + "stream_event_count": 2_000_000, + "giant_attachment_raw_bytes": 12 * 1024 * 1024, + "ordinary_blob_limit_bytes": 64 * 1024 * 1024, + "whale_blob_limit_bytes": 8 * 1024 * 1024 * 1024, + } + + assert WHALE_FIXTURE_DIMENSIONS.giant_attachment_raw_bytes > 0 + assert ( + WHALE_FIXTURE_DIMENSIONS.giant_attachment_raw_bytes + < WHALE_FIXTURE_DIMENSIONS.near_terminal_predecessor_bytes + < WHALE_FIXTURE_DIMENSIONS.ordinary_blob_limit_bytes + < WHALE_FIXTURE_DIMENSIONS.terminal_wire_bytes + < WHALE_FIXTURE_DIMENSIONS.whale_blob_limit_bytes + ) + assert WHALE_FIXTURE_DIMENSIONS.stream_event_count > WHALE_FIXTURE_DIMENSIONS.revision_count + + +def test_multi_million_stream_emits_realistic_distinct_records() -> None: + stream = multi_million_codex_stream() + first = next(stream) + second = next(stream) + assert first["type"] == "session_meta" + assert second["type"] == "response_item" + first_state = next(stream) + second_state = next(stream) + assert first_state is not second_state + assert first_state == {"record_type": "state", "sequence": 0} + assert second_state == {"record_type": "state", "sequence": 1} + + +def test_padding_generator_never_writes_an_outlier_sized_chunk() -> None: + class RecordingSink: + position = 0 + max_write = 0 + + def write(self, value: bytes) -> int: + self.position += len(value) + self.max_write = max(self.max_write, len(value)) + return len(value) + + sink = RecordingSink() + target_bytes = 4 * 1024 * 1024 + _write_padding_record(sink, revision=803, target_bytes=target_bytes, current_bytes=0) # type: ignore[arg-type] + + assert sink.position == target_bytes + assert sink.max_write <= 1024 * 1024 + + +def test_fixture_pack_generator_writes_a_complete_manifest(tmp_path: Path) -> None: + dimensions = WhaleFixtureDimensions( + fixture_id="codex-whale-bounds-test", + revision_count=5, + terminal_wire_bytes=1024 * 1024, + near_terminal_predecessor_bytes=256 * 1024, + stream_event_count=100, + giant_attachment_raw_bytes=32 * 1024, + ordinary_blob_limit_bytes=512 * 1024, + whale_blob_limit_bytes=2 * 1024 * 1024, + ) + fixture = CodexRevisionChainFixture(dimensions=dimensions, session_native_id="codex-whale-test") + + source_path, manifest_path = write_codex_whale_fixture_pack(tmp_path, fixture) + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + + assert source_path.stat().st_size == dimensions.terminal_wire_bytes + assert manifest["fixture_id"] == dimensions.fixture_id + assert manifest["revision_sizes"] == [4_096, 64 * 1024, 128 * 1024, 256 * 1024, 1024 * 1024] + assert len(manifest["revision_sha256"]) == dimensions.revision_count + assert all(len(value) == 64 for value in manifest["revision_sha256"]) + assert manifest["terminal_features"] == ["compaction", "giant-base64-attachment", "codex-stream-dispatch"] + + rerun_source_path, rerun_manifest_path = write_codex_whale_fixture_pack(tmp_path, fixture) + assert rerun_source_path == source_path + assert rerun_source_path.stat().st_size == dimensions.terminal_wire_bytes + assert json.loads(rerun_manifest_path.read_text(encoding="utf-8")) == manifest + + changed_fixture = CodexRevisionChainFixture(dimensions=dimensions, session_native_id="codex-whale-changed") + _changed_source, changed_manifest_path = write_codex_whale_fixture_pack(tmp_path / "changed", changed_fixture) + changed_manifest = json.loads(changed_manifest_path.read_text(encoding="utf-8")) + assert changed_manifest["revision_sizes"] == manifest["revision_sizes"] + assert changed_manifest["revision_sha256"] != manifest["revision_sha256"] diff --git a/tests/unit/maintenance/test_reindex_campaign.py b/tests/unit/maintenance/test_reindex_campaign.py index 50ba441546..e407eafd67 100644 --- a/tests/unit/maintenance/test_reindex_campaign.py +++ b/tests/unit/maintenance/test_reindex_campaign.py @@ -126,6 +126,8 @@ def test_reindex_campaign_manifest_has_positive_denominators(tmp_path: Path) -> assert corpus.manifest.duplicate_raw_ids assert corpus.manifest.restart_session_ids assert set(corpus.manifest.parser_failure_raw_ids).isdisjoint(corpus.manifest.duplicate_raw_ids) + assert dict(corpus.manifest.fixture_dimensions)["fixture_id"] == "codex-whale-bounds-v2" + assert dict(corpus.manifest.fixture_dimensions)["revision_count"] == 804 def test_real_inactive_rebuild_and_canary_preserve_active_and_reject_parser_as_duplicate(tmp_path: Path) -> None: diff --git a/tests/unit/scenarios/test_codex_804_live_proof.py b/tests/unit/scenarios/test_codex_804_live_proof.py index 613eb28c3e..ff05eca0b8 100644 --- a/tests/unit/scenarios/test_codex_804_live_proof.py +++ b/tests/unit/scenarios/test_codex_804_live_proof.py @@ -26,11 +26,13 @@ import time from dataclasses import replace from datetime import datetime -from functools import cache +from itertools import pairwise from pathlib import Path import pytest +from polylogue.config import Config +from polylogue.product import raw_authority from polylogue.scenarios import ( MeasurementScope, WorkloadPhaseObservation, @@ -45,162 +47,48 @@ from polylogue.storage.index_generation import IndexGenerationStore, rebuild_source_evidence_snapshot from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_active_archive_root from tests.infra.archive_canonical_snapshot import archive_snapshot -from tests.infra.corpus_program import ( - Acquire, - CorpusProgram, - ProductionCorpusRuntime, - RawArtifact, -) from tests.infra.rebuild_receipt import write_valid_rebuild_receipt +from tests.infra.whale_fixtures import ( + WHALE_FIXTURE_DIMENSIONS, + CodexRevisionChainFixture, + acquire_codex_revision_chain, +) -REVISION_COUNT = 804 -TERMINAL_WIRE_BYTES = 90_822_451 -SESSION_NATIVE_ID = "codex-sanitized-804-session" +REVISION_COUNT = WHALE_FIXTURE_DIMENSIONS.revision_count +TERMINAL_WIRE_BYTES = WHALE_FIXTURE_DIMENSIONS.terminal_wire_bytes +SESSION_NATIVE_ID = CodexRevisionChainFixture().session_native_id SOURCE_PATH = "codex/incident-804-sanitized.jsonl" -NEAR_TERMINAL_PREDECESSOR_BYTES = 32 * 1024 * 1024 _BASELINE_MESSAGE_IDS = tuple(f"{SESSION_NATIVE_ID}-message-{index}" for index in range(2)) _BASELINE_MESSAGE_TIMESTAMPS = ("2026-07-31T04:25:20Z", "2026-07-31T04:25:20Z") -def _wire_target_bytes(revision: int) -> int: - if revision < REVISION_COUNT - 4: - # Keep every prefix strictly larger than its predecessor without - # making the 800 ordinary snapshots consume hundreds of megabytes. - return 4_096 + revision * 640 - return { - REVISION_COUNT - 4: 8 * 1024 * 1024, - REVISION_COUNT - 3: 16 * 1024 * 1024, - REVISION_COUNT - 2: NEAR_TERMINAL_PREDECESSOR_BYTES, - REVISION_COUNT - 1: TERMINAL_WIRE_BYTES, - }[revision] - - -@cache -def _codex_payload(revision: int, *, terminal: bool) -> bytes: - revision_timestamp = f"2026-07-31T04:{25 + revision // 60:02d}:{20 + revision % 60:02d}Z" - if revision == 0: - records: list[dict[str, object]] = [ - { - "type": "session_meta", - "payload": { - "id": SESSION_NATIVE_ID, - "timestamp": "2026-07-31T04:25:20Z", - "cwd": "/sanitized/codex-804", - }, - }, - ] - for message_index, role in enumerate(("user", "assistant")): - text = f"sanitized incident witness {role} baseline" - records.append( - { - "type": "response_item", - "timestamp": _BASELINE_MESSAGE_TIMESTAMPS[message_index], - "payload": { - "type": "message", - "id": f"{SESSION_NATIVE_ID}-message-{message_index}", - "role": role, - "content": [{"type": "output_text" if role == "assistant" else "input_text", "text": text}], - }, - } - ) - previous_payload = b"" - else: - records = [] - previous_payload = _codex_payload(revision - 1, terminal=False) - # Parsed content grows by containment. The first milestone proves that - # revisions differ semantically, while the later milestones keep the - # production membership classifier's accepted frontier moving toward the - # terminal snapshot without making the fixture 804 messages wide. - milestone_revisions = (1, 800, 801, 802) - if revision in milestone_revisions: - records.append( - { - "type": "response_item", - "timestamp": revision_timestamp, - "payload": { - "type": "message", - "id": f"{SESSION_NATIVE_ID}-milestone-{revision:03d}", - "role": "assistant", - "content": [ - { - "type": "output_text", - "text": f"sanitized parsed milestone revision {revision}", - } - ], - }, - } - ) - if terminal: - records.append( - { - "type": "response_item", - "timestamp": revision_timestamp, - "payload": { - "type": "message", - "id": f"{SESSION_NATIVE_ID}-terminal-803", - "role": "assistant", - "content": [{"type": "output_text", "text": "sanitized terminal response revision 803"}], - }, - } - ) - prefix = previous_payload + b"".join(json.dumps(record, sort_keys=True).encode() + b"\n" for record in records) - wire_template = json.dumps( - { - "payload": { - "padding": "", - "type": "token_count", - }, - "revision": revision, - "type": "response_item", - }, - sort_keys=True, - ).encode() - padding_prefix, padding_suffix = wire_template.split(b'""', maxsplit=1) - padding_size = _wire_target_bytes(revision) - len(prefix) - len(padding_prefix) - len(padding_suffix) - 3 - if padding_size <= 0: - raise AssertionError(f"terminal padding underflow: {padding_size}") - payload = prefix + padding_prefix + b'"' + (b"x" * padding_size) + b'"' + padding_suffix + b"\n" - if terminal: - assert len(payload) == TERMINAL_WIRE_BYTES - else: - assert len(payload) == _wire_target_bytes(revision) - return payload - - -def _baseline_message_timestamps(payload: bytes) -> tuple[str, ...]: +def _baseline_message_timestamps(path: Path) -> tuple[str, ...]: timestamps: dict[str, str] = {} - for line in payload.splitlines(): - record = json.loads(line) - if not isinstance(record, dict): - continue - record_payload = record.get("payload") - if not isinstance(record_payload, dict) or record_payload.get("type") != "message": - continue - message_id = record_payload.get("id") - timestamp = record.get("timestamp") - if isinstance(message_id, str) and message_id in _BASELINE_MESSAGE_IDS: - if not isinstance(timestamp, str): - raise AssertionError(f"baseline message {message_id} has no string timestamp") - timestamps[message_id] = timestamp + with path.open("rb") as handle: + for line in handle: + record = json.loads(line) + if not isinstance(record, dict): + continue + record_payload = record.get("payload") + if not isinstance(record_payload, dict) or record_payload.get("type") != "message": + continue + message_id = record_payload.get("id") + timestamp = record.get("timestamp") + if isinstance(message_id, str) and message_id in _BASELINE_MESSAGE_IDS: + if not isinstance(timestamp, str): + raise AssertionError(f"baseline message {message_id} has no string timestamp") + timestamps[message_id] = timestamp + if len(timestamps) == len(_BASELINE_MESSAGE_IDS): + break return tuple(timestamps[message_id] for message_id in _BASELINE_MESSAGE_IDS) -def _incident_program() -> CorpusProgram: - operations = tuple( - Acquire( - f"revision-{revision:03d}", - RawArtifact( - artifact_id=f"revision-{revision:03d}", - payload=_codex_payload(revision, terminal=revision == REVISION_COUNT - 1), - source_name="codex", - source_path=SOURCE_PATH, - source_index=revision, - metadata={"incident": "codex-804-sanitized", "revision": revision}, - ), - ) - for revision in range(REVISION_COUNT) - ) - return CorpusProgram(operations=operations) +def _file_sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() def _current_rss_bytes() -> int | None: @@ -296,35 +184,27 @@ def _readiness_count(readiness: dict[str, object], key: str) -> int: return int(value) -@pytest.mark.timeout(420) +@pytest.mark.timeout(900) @pytest.mark.uses_real_clock("waits for a real subprocess replay checkpoint and kill/resume boundary") -def test_sanitized_codex_804_revision_recovery_proof( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch, request: pytest.FixtureRequest -) -> None: +def test_sanitized_codex_804_revision_recovery_proof(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: """Exercise acquisition through candidate promotion at incident scale. Anti-vacuity: deleting the production ``AcquisitionService`` call from - ``ProductionCorpusRuntime.acquire`` leaves the source row count at zero; + ``acquire_codex_revision_chain`` leaves the source row count at zero; deleting parser/convergence leaves the terminal index session absent; deleting the resumable candidate path leaves no paused transaction or inactive generation for the recovery assertions below. """ - _codex_payload.cache_clear() - request.addfinalizer(_codex_payload.cache_clear) root = tmp_path / "codex-804-proof" monkeypatch.setenv("POLYLOGUE_ARCHIVE_ROOT", str(root)) initialize_active_archive_root(root) - runtime = ProductionCorpusRuntime(root) + fixture = CodexRevisionChainFixture() + source_path = root / "fixture-sources" / SOURCE_PATH setup_before = _resource_sample(root) setup_started = time.perf_counter() - program = _incident_program() profile_id = _schema_profile_id() - program_json = program.to_json() - # The JSON program is intentionally the canonical witness identity. Its - # 90 MiB terminal payload is a faithful wire-shape fixture, not a parser - # replacement or a pre-populated database. setup_after = _resource_sample(root) phases: list[WorkloadPhaseObservation] = [ _phase( @@ -332,40 +212,47 @@ def test_sanitized_codex_804_revision_recovery_proof( setup_before, setup_after, setup_started, - completed=REVISION_COUNT, + completed=0, total=REVISION_COUNT, ) ] acquire_before = _resource_sample(root) acquire_started = time.perf_counter() - run = program.run(runtime) + acquired_raw_ids, sizes, fixture_sha256s = acquire_codex_revision_chain(root, fixture, source_path) + with sqlite3.connect(root / "source.db") as conn: + persisted_sizes = tuple( + int(row[0]) for row in conn.execute("SELECT blob_size FROM raw_sessions ORDER BY blob_size") + ) + manifest_path = fixture.write_manifest(root / "fixture-manifest.json", sizes, fixture_sha256s) + fixture_manifest = manifest_path.read_text(encoding="utf-8") acquire_after = _resource_sample(root) phases.append( _phase( "acquire", acquire_before, acquire_after, acquire_started, completed=REVISION_COUNT, total=REVISION_COUNT ) ) - assert len(run.state.artifacts) == REVISION_COUNT - assert run.state.crashed is False - assert len(program_json) > REVISION_COUNT - first_revision_payload = _codex_payload(0, terminal=False) - second_revision_payload = _codex_payload(1, terminal=False) - assert first_revision_payload != second_revision_payload - for revision in range(1, REVISION_COUNT): - current_payload = _codex_payload(revision, terminal=revision == REVISION_COUNT - 1) - previous_payload = _codex_payload(revision - 1, terminal=False) - assert current_payload.startswith(previous_payload) - assert b"incident witness user baseline" in first_revision_payload - assert b"parsed milestone revision 1" in second_revision_payload - # Mutation that assigns every recursively extended baseline message the - # current revision timestamp fails this exact multi-revision comparison. - for revision, terminal in ((0, False), (1, False), (800, False), (803, True)): - assert _baseline_message_timestamps(_codex_payload(revision, terminal=terminal)) == ( - _BASELINE_MESSAGE_TIMESTAMPS - ) - - program_digest = hashlib.sha256(program_json.encode("utf-8")).hexdigest() + assert len(acquired_raw_ids) == REVISION_COUNT + assert len(set(acquired_raw_ids)) == REVISION_COUNT + assert len(sizes) == REVISION_COUNT + assert sizes[0] == 4_096 + assert sizes[-2] == WHALE_FIXTURE_DIMENSIONS.near_terminal_predecessor_bytes + assert sizes[-1] == TERMINAL_WIRE_BYTES + assert all(current > previous for previous, current in pairwise(sizes)) + assert persisted_sizes == sizes + manifest_payload = json.loads(fixture_manifest) + assert manifest_payload["fixture_id"] == WHALE_FIXTURE_DIMENSIONS.fixture_id + assert manifest_payload["session_native_id"] == SESSION_NATIVE_ID + assert manifest_payload["revision_sizes"] == list(sizes) + assert manifest_payload["revision_sha256"] == list(fixture_sha256s) + assert manifest_payload["dimensions"] == dict(WHALE_FIXTURE_DIMENSIONS.manifest_dimensions()) + with source_path.open("rb") as handle: + first_source_bytes = handle.read(8_192) + assert b"sanitized incident witness user baseline" in first_source_bytes + assert b"sanitized parsed milestone revision 1" in first_source_bytes + assert _baseline_message_timestamps(source_path) == _BASELINE_MESSAGE_TIMESTAMPS + + fixture_manifest_digest = hashlib.sha256(fixture_manifest.encode("utf-8")).hexdigest() census_before = _resource_sample(root) census_started = time.perf_counter() @@ -382,6 +269,19 @@ def test_sanitized_codex_804_revision_recovery_proof( assert pre_recovery_unresolved_count == REVISION_COUNT assert pre_recovery_membership_count == 0 assert pre_recovery_census_count == 0 + whale_candidate = raw_authority.whale_pass_candidate( + Config(archive_root=root, render_root=root / "render", sources=[]), + ordinary_max_payload_bytes=WHALE_FIXTURE_DIMENSIONS.ordinary_blob_limit_bytes, + whale_max_payload_bytes=WHALE_FIXTURE_DIMENSIONS.whale_blob_limit_bytes, + ) + assert whale_candidate is not None + assert whale_candidate == acquired_raw_ids[0] + whale_component_bytes = sum(sizes) + assert ( + WHALE_FIXTURE_DIMENSIONS.ordinary_blob_limit_bytes + < whale_component_bytes + <= WHALE_FIXTURE_DIMENSIONS.whale_blob_limit_bytes + ) phases.append( _phase( "census", @@ -644,14 +544,13 @@ async def recording_replay(*args, **kwargs): assert source_path_count == 1 assert parse_error_count == 0 assert authorities == ("byte_proven",) - assert tuple(row[1] for row in raw_rows) == tuple( - _wire_target_bytes(revision) for revision in range(REVISION_COUNT) - ) + assert tuple(row[1] for row in raw_rows) == sizes + assert tuple(row[4] for row in raw_rows) == fixture_sha256s assert raw_rows[-1][1] == TERMINAL_WIRE_BYTES - assert raw_rows[-2][1] >= NEAR_TERMINAL_PREDECESSOR_BYTES + assert raw_rows[-2][1] == WHALE_FIXTURE_DIMENSIONS.near_terminal_predecessor_bytes terminal_raw_id = raw_rows[-1][2] terminal_blob_hash = raw_rows[-1][4] - assert raw_rows[-1][4] == hashlib.sha256(_codex_payload(REVISION_COUNT - 1, terminal=True)).hexdigest() + assert raw_rows[-1][4] == fixture_sha256s[-1] == _file_sha256(source_path) with sqlite3.connect(root / "source.db") as conn: post_recovery_membership_count = int( conn.execute("SELECT COUNT(DISTINCT raw_id) FROM raw_session_memberships").fetchone()[0] @@ -725,6 +624,20 @@ async def recording_replay(*args, **kwargs): ("%sanitized terminal response revision 803%",), ).fetchone()[0] ) + compaction_event_count = int( + conn.execute( + "SELECT COUNT(*) FROM session_events WHERE session_id = ? AND event_type = 'compaction'", + (f"codex-session:{SESSION_NATIVE_ID}",), + ).fetchone()[0] + ) + attachment_summary_count = int( + conn.execute("SELECT COUNT(*) FROM blocks WHERE search_text LIKE '%sha256_base64=%'").fetchone()[0] + ) + attachment_image_block_count = int( + conn.execute( + "SELECT COUNT(*) FROM blocks WHERE block_type = 'image' AND search_text LIKE '%sha256_base64=%'" + ).fetchone()[0] + ) selected_heads = conn.execute( "SELECT accepted_raw_id FROM raw_revision_heads WHERE logical_source_key = ?", (terminal_logical_source_key,), @@ -744,7 +657,7 @@ async def recording_replay(*args, **kwargs): assert len(selected_heads) == 1 selected_raw_id = str(selected_heads[0][0]) assert selected_raw_id == terminal_raw_id - assert indexed == [(f"codex-session:{SESSION_NATIVE_ID}", selected_raw_id, 7)] + assert indexed == [(f"codex-session:{SESSION_NATIVE_ID}", selected_raw_id, 8)] with sqlite3.connect(root / "source.db") as conn: selected_source_row = conn.execute( "SELECT blob_size, lower(hex(blob_hash)) FROM raw_sessions WHERE raw_id = ?", @@ -752,6 +665,9 @@ async def recording_replay(*args, **kwargs): ).fetchone() assert selected_source_row == (TERMINAL_WIRE_BYTES, terminal_blob_hash) assert terminal_block_count > 0 + assert compaction_event_count == 1 + assert attachment_summary_count > 0 + assert attachment_image_block_count > 0 expected_baseline_timestamps = tuple( int(datetime.fromisoformat(timestamp.replace("Z", "+00:00")).timestamp() * 1000) for timestamp in _BASELINE_MESSAGE_TIMESTAMPS @@ -790,8 +706,8 @@ async def recording_replay(*args, **kwargs): inputs=( replace( input_ref, - input_id=f"{input_ref.input_id}:program-json-sha256:{program_digest}", - corpus_id=f"program-json:sha256:{program_digest}", + input_id=f"{input_ref.input_id}:fixture-manifest-sha256:{fixture_manifest_digest}", + corpus_id=f"fixture-manifest:sha256:{fixture_manifest_digest}", ), ), ) @@ -808,7 +724,7 @@ async def recording_replay(*args, **kwargs): "fixture:codex-804-sanitized", f"schema-registry:{profile_id}", f"candidate-generation:{generation_id}", - f"program-json-sha256:{program_digest}", + f"fixture-manifest-sha256:{fixture_manifest_digest}", f"source-raw-count:{raw_count}", f"source-parse-error-count:{parse_error_count}", f"recovery-unresolved-before-crash:{pre_recovery_unresolved_count}", @@ -820,8 +736,8 @@ async def recording_replay(*args, **kwargs): cleanup_complete=True, notes=( "Sanitized structural witness only; no live /realm/db/polylogue access.", - "Fixture setup includes 804 payload construction, schema hashing, and canonical program serialization.", - f"Serialized program digest is sha256:{program_digest} and is bound into the receipt input identity.", + "Fixture setup includes 804 on-disk payload revisions, schema hashing, and a canonical fixture manifest.", + f"Serialized fixture manifest digest is sha256:{fixture_manifest_digest} and is bound into the receipt input identity.", "Replay and postflight subprocess RSS is unavailable because statm samples only the pytest parent; storage growth is not reported as write I/O.", "The crash boundary hard-exits after durable transaction creation with all 804 authority rows unresolved; a fresh process resumes and materializes the cohort into an inactive candidate.", "Live confidence remains open until the named successor receipts bind the active archive.", @@ -841,7 +757,7 @@ async def recording_replay(*args, **kwargs): "receipt_id": workload_receipt.receipt_id, "raw_count": raw_count, "max_blob_size": max_blob_size, - "program_digest": program_digest, + "fixture_manifest_digest": fixture_manifest_digest, "parse_error_count": parse_error_count, "terminal_raw_id": selected_raw_id, "indexed_session": indexed[0][0], diff --git a/tests/unit/sources/test_codex_whale_fixtures.py b/tests/unit/sources/test_codex_whale_fixtures.py new file mode 100644 index 0000000000..9c0ee135c8 --- /dev/null +++ b/tests/unit/sources/test_codex_whale_fixtures.py @@ -0,0 +1,78 @@ +"""Real Codex streaming-dispatch regression for the shared whale fixture.""" + +from __future__ import annotations + +from collections.abc import Iterator + +from tests.infra.whale_fixtures import WHALE_FIXTURE_DIMENSIONS, multi_million_codex_stream + + +def test_multi_million_codex_stream_uses_real_stream_dispatch_without_truncation() -> None: + """Anti-vacuity: bypassing ``parse_stream_payload`` or shrinking the event boundary fails. + + State records are deliberately reused immutable evidence. The parser must + consume exactly two million of them through its streaming entry point, + while materializing only the one authored message in the resulting session. + """ + from polylogue.sources.dispatch import parse_stream_payload + + class CountingStream: + def __init__(self) -> None: + self.yielded = 0 + self._source = multi_million_codex_stream() + + def __iter__(self) -> CountingStream: + return self + + def __next__(self) -> dict[str, object]: + value = next(self._source) + self.yielded += 1 + return value + + stream = CountingStream() + sessions = parse_stream_payload("codex", stream, "codex-stream-million", source_path="million.jsonl") + + assert stream.yielded == WHALE_FIXTURE_DIMENSIONS.stream_event_count + 2 + assert len(sessions) == 1 + assert sessions[0].provider_session_id == "codex-stream-million" + assert len(sessions[0].messages) == 1 + assert sessions[0].messages[0].text == "sanitized streaming boundary" + + +def test_stream_dispatch_does_not_retain_distinct_input_records() -> None: + """A restored list(records) fails from live-record accumulation before parsing.""" + from polylogue.sources.dispatch import parse_stream_payload + + class TrackedStateRecord(dict[str, object]): + live = 0 + + def __init__(self, sequence: int) -> None: + super().__init__(record_type="state", sequence=sequence) + type(self).live += 1 + + def __del__(self) -> None: + type(self).live -= 1 + + def __reduce__(self) -> tuple[object, tuple[dict[str, object]]]: + return dict, (dict(self),) + + def guarded_stream() -> Iterator[dict[str, object]]: + yield {"type": "session_meta", "payload": {"id": "bounded-stream"}} + yield { + "type": "response_item", + "payload": { + "type": "message", + "id": "bounded-stream-message", + "role": "user", + "content": [{"type": "input_text", "text": "bounded"}], + }, + } + for sequence in range(10_000): + if TrackedStateRecord.live > 4: + raise AssertionError("stream parser retained decoded input records") + yield TrackedStateRecord(sequence) + + sessions = parse_stream_payload("codex", guarded_stream(), "bounded-stream", source_path="bounded.jsonl") + + assert len(sessions) == 1 + assert sessions[0].messages[0].text == "bounded" diff --git a/tests/unit/sources/test_parsers_codex.py b/tests/unit/sources/test_parsers_codex.py index 6ea5fb6bb4..d449845008 100644 --- a/tests/unit/sources/test_parsers_codex.py +++ b/tests/unit/sources/test_parsers_codex.py @@ -1442,6 +1442,61 @@ def test_function_call_output_omits_inline_image_data_urls_from_text(self) -> No assert "mime=image/png" in (message.text or "") assert "sha256_base64=" in (message.text or "") + def test_message_preserves_bounded_inline_image_evidence(self) -> None: + payload = [ + { + "type": "response_item", + "payload": { + "type": "message", + "id": "message-image", + "role": "user", + "content": [ + {"type": "input_text", "text": "inspect this image"}, + {"type": "input_image", "image_url": "data:image/png;base64," + ("a" * 4096)}, + ], + }, + } + ] + + result = parse(payload, "fallback") + + assert len(result.messages) == 1 + message = result.messages[0] + assert "data:image/png;base64" not in (message.text or "") + assert message.text == "inspect this image" + assert all("data:image/png;base64" not in (block.text or "") for block in message.blocks) + image_blocks = [block for block in message.blocks if block.type is BlockType.IMAGE] + assert len(image_blocks) == 1 + assert image_blocks[0].media_type == "image/png" + assert "mime=image/png" in (image_blocks[0].text or "") + assert "sha256_base64=" in (image_blocks[0].text or "") + + def test_message_preserves_image_only_bounded_evidence(self) -> None: + payload = [ + { + "type": "response_item", + "payload": { + "type": "message", + "id": "message-image-only", + "role": "user", + "content": [ + {"type": "input_image", "image_url": "data:image/png;base64," + ("a" * 4096)}, + ], + }, + } + ] + + result = parse(payload, "fallback") + + assert len(result.messages) == 1 + message = result.messages[0] + assert message.text == "" + assert len(message.blocks) == 1 + assert message.blocks[0].type is BlockType.IMAGE + assert message.blocks[0].media_type == "image/png" + assert "data:image/png;base64" not in (message.blocks[0].text or "") + assert "sha256_base64=" in (message.blocks[0].text or "") + # ============================================================================= # Edge Cases