From 0e682fd1886e238efe9f37d52e421515312b7861 Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 4 Aug 2026 08:49:16 +0200 Subject: [PATCH 1/9] fix(daemon): classify raw failure lifecycle Problem Stopped daemon convergence presented retained hot captures, terminal input rejections, and unexplained parser failures as one retryable failure bucket. What changed Record closed source-tier evidence for deferred hot JSONL captures and terminal corrupt or unsupported payloads. Surface the lifecycle in daemon and direct CLI status, with a read-only preflight census and production-route tests. Compatibility Existing raw rows remain unexplained until a reviewed post-deploy route records new typed evidence. No archive data, cursor, or blob was mutated. Co-Authored-By: Codex --- devtools/docs_surface.py | 6 + docs/README.md | 1 + .../2026-08-04-raw-failure-preflight.md | 42 ++++++ docs/audits/README.md | 1 + docs/plans/layering.yaml | 2 +- polylogue/cli/commands/status.py | 30 +++++ polylogue/core/raw_failure_evidence.py | 50 +++++++ polylogue/daemon/health.py | 36 +++-- polylogue/daemon/status.py | 100 +++++++++++++- polylogue/daemon/status_snapshot.py | 3 + polylogue/sources/live/batch.py | 69 +++++++++- .../storage/sqlite/archive_tiers/archive.py | 22 ++++ .../archive_tiers/revision_governance.py | 37 ++++++ .../sqlite/archive_tiers/source_write.py | 35 ++++- tests/unit/cli/test_status.py | 70 ++++++++++ tests/unit/daemon/test_health_check_paths.py | 26 ++++ tests/unit/daemon/test_raw_failure_sample.py | 69 ++++++++++ tests/unit/sources/test_live_batch_support.py | 124 ++++++++++++------ 18 files changed, 664 insertions(+), 59 deletions(-) create mode 100644 docs/audits/2026-08-04-raw-failure-preflight.md create mode 100644 polylogue/core/raw_failure_evidence.py diff --git a/devtools/docs_surface.py b/devtools/docs_surface.py index 3da625e702..03075b8c19 100644 --- a/devtools/docs_surface.py +++ b/devtools/docs_surface.py @@ -450,6 +450,12 @@ def _entry(title: str, path: str, description: str, tier: DocsTier) -> DocsEntry "I3 live evidence, source-tier reconciliation safeguards, and the direct-reindex gate.", "archive", ), + _entry( + "Raw-Failure Preflight", + "audits/2026-08-04-raw-failure-preflight.md", + "Read-only raw-failure census before lifecycle evidence deployment.", + "archive", + ), _entry("Audit Record Index", "audits/README.md", "Index of dated investigation records.", "archive"), _entry( "1498 Cascade Retrospective", diff --git a/docs/README.md b/docs/README.md index 2928993e3e..4acd06a3ab 100644 --- a/docs/README.md +++ b/docs/README.md @@ -137,6 +137,7 @@ Start with **Guides** for a task, **Reference** for a surface contract, and **Ar | [Race Window Audit](audits/2026-07-09-race-window-audit.md) | Race-window investigation record. | | [Reindex Forcing-Class Audit](audits/2026-08-04-reindex-forcing-class-audit.md) | Forcing-class and reindex-gate evidence audit. | | [Blob-Reference Liveness Closure Audit](audits/2026-08-04-blob-ref-liveness-closure.md) | I3 live evidence, source-tier reconciliation safeguards, and the direct-reindex gate. | +| [Raw-Failure Preflight](audits/2026-08-04-raw-failure-preflight.md) | Read-only raw-failure census before lifecycle evidence deployment. | | [Audit Record Index](audits/README.md) | Index of dated investigation records. | | [1498 Cascade Retrospective](retro/2026-05-24-1498-cascade.md) | Historical cascade incident retrospective. | | [Retrospective Index](retro/README.md) | Index of historical incident retrospectives. | diff --git a/docs/audits/2026-08-04-raw-failure-preflight.md b/docs/audits/2026-08-04-raw-failure-preflight.md new file mode 100644 index 0000000000..89722dcef3 --- /dev/null +++ b/docs/audits/2026-08-04-raw-failure-preflight.md @@ -0,0 +1,42 @@ +# Raw-failure preflight, 2026-08-04 + +## Scope and method + +This is an immutable, read-only census of the active archive before deploying `polylogue-dyica`. It opened `/realm/db/polylogue/source.db` and `ops.db` with SQLite read-only mode, inspected the stopped user daemon, and compared source metadata with retained raw metadata. It did not start the daemon, run reprocessing, reset a cursor, or remove any source or blob data. + +Raw IDs and source paths are intentionally omitted. The captures and exports contain private operator data, and aggregate evidence is sufficient for the implementation decision. + +`polylogued.service` was stopped after an exit 143 at 2026-08-04 04:40:38. The one maintenance failure was a session-insights repair blocked by the live source-tier schema version. The deployed runtime expected version 25 while the archive was at version 24. This implementation does not migrate that archive. + +## Census + +The archive contained 112 failures: 111 raw parse failures and one maintenance failure. There were no raw validation failures. + +| Origin or route | Existing artifact kind | Count | Source mutability evidence | Retry eligibility before deployment | Classification | +| --- | --- | ---: | --- | --- | --- | +| `claude-code-session` | `coordinator_session_stream`, `supported_parseable` | 59 | Live Claude source; a later current file observation exists, but historical retained bytes have no recorded byte-prefix proof | Unexplained historical rows | Candidate deferred only after a future route proves the retained bytes are the strict current prefix of a larger source | +| `claude-code-session` | No artifact observation | 4 | Live Claude source, hot-file metadata | Unexplained | Lifecycle revision conflict | +| `codex-session` | No artifact observation | 14 | Live Codex source, hot-file metadata | Unexplained | Lifecycle revision conflict | +| `codex-session` | No artifact observation | 1 | Live Codex source, hot-file metadata | Unexplained | Unconvertible byte-head lifecycle failure | +| `unknown-export` | No artifact observation | 25 | Source missing from archive inbox or legacy inbox | No automatic retry | Terminal unsupported shape: parser produced no sessions | +| `unknown-export` | No artifact observation | 6 | Five legacy inputs remain immutable and available; one source is absent | No automatic retry | Terminal corrupt input: JSON decoding failed | +| `hermes-session` | No artifact observation | 2 | Live Hermes source, hot-file metadata | No automatic retry without a demonstrated parser path | Terminal unsupported shape: artifact produced no materializable sessions | +| maintenance replay | Failure routing record | 1 | Durable source tier is older than the deployed runtime requirement | Blocked pending backup-gated migration and reviewed retry | Explicit maintenance schema mismatch | + +The rows sum to 112. The 59 Claude rows are the only existing `raw_artifacts` observations for the 111 raw failures. Their current `supported_parseable` classification is historical parser metadata, not structural proof of a deferred capture. The remaining 52 raw failures have no artifact observation. + +The source mutability evidence groups the 111 raw failures as follows: 80 live-source rows with hot-file metadata, 6 immutable legacy inputs still available, and 25 source-missing archive or legacy inputs. Hot metadata alone is not enough to defer a raw. A future ingest must establish both conditions against the same captured bytes: the current source is larger and its prefix hash exactly matches the retained payload. + +## Implementation decision + +The real full-ingest route now writes a closed `raw_artifacts` outcome after retaining a raw failure: + +- `deferred_hot_jsonl_capture` only when the source has grown and its prefix exactly matches the retained incomplete JSONL payload; +- `terminal_corrupt_input` for an incomplete capture without that proof; +- `terminal_unsupported_shape` when parsing yields no positive conversational evidence. + +All three states retain the raw payload and its parser diagnostic. Deferred and terminal outcomes acknowledge the source record so the cursor does not retry an unchanged payload indefinitely. Status and health report deferred retryable work, terminal rejections, and unexplained failures separately. Historical rows stay unexplained until an explicitly reviewed route records new evidence. + +## Post-deploy boundary + +No live reprocessing occurred for this change. A future operator run requires a fresh verified backup, a targeted dry-run receipt, review of each proposed raw state transition, and an apply receipt. It must not reset cursors, bulk reprocess the archive, or delete source/blob data. The stopped daemon must resume convergence for deferred work; remaining unexplained lifecycle failures stay visible for separate diagnosis. diff --git a/docs/audits/README.md b/docs/audits/README.md index 7da12b0bb5..fad5bc9c00 100644 --- a/docs/audits/README.md +++ b/docs/audits/README.md @@ -12,3 +12,4 @@ and [Developer Tools](../devtools.md) references for present-tense behavior. - [Race window audit](2026-07-09-race-window-audit.md) - [Reindex forcing-class audit](2026-08-04-reindex-forcing-class-audit.md) - [Blob-reference liveness closure audit](2026-08-04-blob-ref-liveness-closure.md) +- [Raw-failure preflight](2026-08-04-raw-failure-preflight.md) diff --git a/docs/plans/layering.yaml b/docs/plans/layering.yaml index c5b93f9499..9f1093796f 100644 --- a/docs/plans/layering.yaml +++ b/docs/plans/layering.yaml @@ -57,7 +57,7 @@ writer_modules: [apply_source_raw_state_update, bind_source_raw_revision, record_capture_mode_observation, record_excised_blob_hash, write_history_sidecar, delete_source_hook_event, write_source_blob_refs, write_source_hook_event, write_source_raw_session, - write_source_raw_session_blob_ref] + write_source_raw_session_blob_ref, upsert_raw_artifact] - path: polylogue/storage/sqlite/archive_tiers/write.py surfaces: - tier: index diff --git a/polylogue/cli/commands/status.py b/polylogue/cli/commands/status.py index d30b209b74..2146c1891f 100644 --- a/polylogue/cli/commands/status.py +++ b/polylogue/cli/commands/status.py @@ -1549,6 +1549,9 @@ def _compact_raw_failure_status(status: dict[str, Any]) -> dict[str, Any]: "validation": "raw_validation_failures", "quarantined": "raw_quarantined", "maintenance": "raw_maintenance_failures", + "deferred_retryable": "raw_deferred_failures", + "terminal_rejections": "raw_terminal_rejections", + "unexplained": "raw_unexplained_failures", "detection_warnings": "raw_detection_warnings", } failures = {label: status[key] for label, key in keys.items() if key in status} @@ -1558,6 +1561,23 @@ def _compact_raw_failure_status(status: dict[str, Any]) -> dict[str, Any]: return failures +def _direct_raw_failure_status(root: Path) -> dict[str, Any]: + """Adapt the archive raw-failure ledger for the stopped-daemon surface.""" + from polylogue.daemon.status import raw_failure_info_for_root + + info = raw_failure_info_for_root(root) + return { + "raw_parse_failures": _safe_int(info.get("parse_failures")), + "raw_validation_failures": _safe_int(info.get("validation_failures")), + "raw_quarantined": _safe_int(info.get("quarantined")), + "raw_maintenance_failures": _safe_int(info.get("maintenance_failures")), + "raw_deferred_failures": _safe_int(info.get("deferred_failures")), + "raw_terminal_rejections": _safe_int(info.get("terminal_rejections")), + "raw_unexplained_failures": _safe_int(info.get("unexplained_failures")), + "raw_failure_samples": info.get("samples", []), + } + + def _show_daemon_status_unavailable_json(env: AppEnv) -> None: payload = { "daemon_liveness": True, @@ -1654,6 +1674,7 @@ def _show_direct_json( "next_action": diag.next_action, "diagnostic": diagnostic_payload(diag), } + payload.update(_direct_raw_failure_status(active_root)) if active_db is not None and active_db.exists(): payload["active_db_path"] = str(active_db) try: @@ -2241,6 +2262,15 @@ def _show_direct_status( env.ui.console.print(f" Sessions: {convs:,}") env.ui.console.print(f" Messages: {msgs:,}") env.ui.console.print(f" Raw records: {raw:,}") + raw_failure_status = _direct_raw_failure_status(active_root) + raw_total = raw_failure_status["raw_parse_failures"] + raw_failure_status["raw_validation_failures"] + if raw_total: + env.ui.console.print( + " Raw failures: " + f"{raw_total:,} total, {raw_failure_status['raw_deferred_failures']:,} deferred retryable, " + f"{raw_failure_status['raw_terminal_rejections']:,} terminal, " + f"{raw_failure_status['raw_unexplained_failures']:,} unexplained" + ) if unidentified: env.ui.console.print( f" Unidentified artifacts: [yellow]{unidentified:,}[/yellow] " diff --git a/polylogue/core/raw_failure_evidence.py b/polylogue/core/raw_failure_evidence.py new file mode 100644 index 0000000000..3ad12d0784 --- /dev/null +++ b/polylogue/core/raw_failure_evidence.py @@ -0,0 +1,50 @@ +"""Closed evidence vocabulary for raw parse outcomes. + +The source tier retains the original raw bytes and parser diagnostic. This +module adds the separate, machine-readable outcome needed to distinguish a +source that may progress from a payload that has reached a terminal refusal. +""" + +from __future__ import annotations + +from enum import StrEnum + +from polylogue.core.enums import ArtifactSupportStatus + + +class RawFailureEvidenceKind(StrEnum): + """Durable lifecycle evidence attached to a retained raw artifact.""" + + DEFERRED_HOT_JSONL_CAPTURE = "deferred_hot_jsonl_capture" + TERMINAL_CORRUPT_INPUT = "terminal_corrupt_input" + TERMINAL_UNSUPPORTED_SHAPE = "terminal_unsupported_shape" + + @property + def support_status(self) -> ArtifactSupportStatus: + if self is RawFailureEvidenceKind.DEFERRED_HOT_JSONL_CAPTURE: + return ArtifactSupportStatus.PARTIAL_DECODE + if self is RawFailureEvidenceKind.TERMINAL_CORRUPT_INPUT: + return ArtifactSupportStatus.DECODE_FAILED + return ArtifactSupportStatus.UNSUPPORTED_PARSEABLE + + @property + def lifecycle(self) -> str: + return "deferred" if self is RawFailureEvidenceKind.DEFERRED_HOT_JSONL_CAPTURE else "terminal" + + +RAW_FAILURE_EVIDENCE_KINDS = frozenset(kind.value for kind in RawFailureEvidenceKind) +RAW_FAILURE_DEFERRED_EVIDENCE_KINDS = frozenset({RawFailureEvidenceKind.DEFERRED_HOT_JSONL_CAPTURE.value}) +RAW_FAILURE_TERMINAL_EVIDENCE_KINDS = frozenset( + { + RawFailureEvidenceKind.TERMINAL_CORRUPT_INPUT.value, + RawFailureEvidenceKind.TERMINAL_UNSUPPORTED_SHAPE.value, + } +) + + +__all__ = [ + "RAW_FAILURE_DEFERRED_EVIDENCE_KINDS", + "RAW_FAILURE_EVIDENCE_KINDS", + "RAW_FAILURE_TERMINAL_EVIDENCE_KINDS", + "RawFailureEvidenceKind", +] diff --git a/polylogue/daemon/health.py b/polylogue/daemon/health.py index c1e15a1985..a66556eacf 100644 --- a/polylogue/daemon/health.py +++ b/polylogue/daemon/health.py @@ -677,7 +677,13 @@ def _check_raw_failures_medium() -> HealthAlert: quarantined = info.get("quarantined", 0) if isinstance(info.get("quarantined"), int) else 0 raw_maint = info.get("maintenance_failures", 0) maintenance = int(raw_maint) if isinstance(raw_maint, (int, float)) else 0 - total_failures = parse + validation + maintenance + raw_deferred = info.get("deferred_failures", 0) + deferred = int(raw_deferred) if isinstance(raw_deferred, (int, float)) else 0 + raw_terminal = info.get("terminal_rejections", 0) + terminal = int(raw_terminal) if isinstance(raw_terminal, (int, float)) else 0 + raw_unexplained = info.get("unexplained_failures", 0) + unexplained = int(raw_unexplained) if isinstance(raw_unexplained, (int, float)) else parse + validation + total_failures = unexplained + maintenance op_hint = "" if maintenance > 0: @@ -690,31 +696,41 @@ def _check_raw_failures_medium() -> HealthAlert: op_hint = f" (op={str(op_id)[:8]})" break - if total_failures == 0: + if total_failures == 0 and deferred == 0: severity = HealthSeverity.OK - message = "no raw failures" + message = ( + f"no unexplained raw failures ({terminal} terminal rejections recorded)" + if terminal + else "no raw failures" + ) + elif total_failures == 0: + severity = HealthSeverity.WARNING + message = f"{deferred} deferred retryable raw capture(s); daemon work remains pending" elif total_failures <= _RAW_FAILURE_WARN_COUNT: severity = HealthSeverity.WARNING message = ( - f"{total_failures} raw failures ({quarantined} quarantined, {maintenance} maintenance){op_hint}" + f"{total_failures} unexplained raw failures ({quarantined} quarantined, {maintenance} maintenance, " + f"{deferred} deferred, {terminal} terminal){op_hint}" if maintenance - else f"{total_failures} raw failures ({quarantined} quarantined)" + else f"{total_failures} unexplained raw failures ({quarantined} quarantined, {deferred} deferred, {terminal} terminal)" ) elif total_failures <= _RAW_FAILURE_ERROR_COUNT: severity = HealthSeverity.ERROR message = ( - f"{total_failures} raw failures ({quarantined} quarantined, {maintenance} maintenance){op_hint}" + f"{total_failures} unexplained raw failures ({quarantined} quarantined, {maintenance} maintenance, " + f"{deferred} deferred, {terminal} terminal){op_hint}" if maintenance - else f"{total_failures} raw failures ({quarantined} quarantined)" + else f"{total_failures} unexplained raw failures ({quarantined} quarantined, {deferred} deferred, {terminal} terminal)" ) else: severity = HealthSeverity.CRITICAL base = ( - f"{total_failures} raw failures ({quarantined} quarantined, {maintenance} maintenance){op_hint}" + f"{total_failures} unexplained raw failures ({quarantined} quarantined, {maintenance} maintenance, " + f"{deferred} deferred, {terminal} terminal){op_hint}" if maintenance - else f"{total_failures} raw failures ({quarantined} quarantined)" + else f"{total_failures} unexplained raw failures ({quarantined} quarantined, {deferred} deferred, {terminal} terminal)" ) - message = f"{base} — investigation needed" + message = f"{base}; investigation needed" return HealthAlert( check_name="raw_failures", tier=HealthTier.MEDIUM, diff --git a/polylogue/daemon/status.py b/polylogue/daemon/status.py index 401e4db276..7471b365d6 100644 --- a/polylogue/daemon/status.py +++ b/polylogue/daemon/status.py @@ -22,6 +22,10 @@ from polylogue.core.payload_coercion import required_str as _required_str from polylogue.core.payload_coercion import row_float as _row_float from polylogue.core.payload_coercion import row_int as _row_int +from polylogue.core.raw_failure_evidence import ( + RAW_FAILURE_DEFERRED_EVIDENCE_KINDS, + RAW_FAILURE_TERMINAL_EVIDENCE_KINDS, +) from polylogue.core.stats import percentile from polylogue.daemon.catchup_status import ( CatchupStatus as CatchupStatus, @@ -404,6 +408,7 @@ class RawFailureSample(BaseModel): source: Literal["ingest", "maintenance"] = "ingest" operation_id: str | None = None locator: str | None = None + lifecycle: Literal["deferred", "terminal", "unexplained"] | None = None @field_validator("redacted_error", mode="before") @classmethod @@ -486,6 +491,9 @@ class DaemonStatus(BaseModel): raw_validation_failures: int = 0 raw_quarantined: int = 0 raw_maintenance_failures: int = 0 + raw_deferred_failures: int = 0 + raw_terminal_rejections: int = 0 + raw_unexplained_failures: int = 0 raw_failure_samples: list[RawFailureSample] = Field(default_factory=list) raw_detection_warnings: int = 0 health: DaemonHealth = Field(default_factory=DaemonHealth) @@ -845,6 +853,9 @@ def _raw_failure_info() -> dict[str, object]: "validation_failures": 0, "quarantined": 0, "maintenance_failures": maintenance_count, + "deferred_failures": 0, + "terminal_rejections": 0, + "unexplained_failures": 0, "samples": maintenance_samples, } archive_info = _archive_raw_failure_info( @@ -871,6 +882,9 @@ def _raw_failure_info() -> dict[str, object]: "quarantined": 0, "detection_warnings": 0, "maintenance_failures": maintenance_count, + "deferred_failures": 0, + "terminal_rejections": 0, + "unexplained_failures": 0, "samples": maintenance_samples, } @@ -948,6 +962,9 @@ def _raw_failure_info() -> dict[str, object]: "quarantined": quarantined, "detection_warnings": detection_warnings_count, "maintenance_failures": maintenance_count, + "deferred_failures": 0, + "terminal_rejections": 0, + "unexplained_failures": parse_fail + validation_fail, "samples": combined, } except sqlite3.Error as exc: @@ -958,6 +975,9 @@ def _raw_failure_info() -> dict[str, object]: "quarantined": 0, "detection_warnings": 0, "maintenance_failures": maintenance_count, + "deferred_failures": 0, + "terminal_rejections": 0, + "unexplained_failures": 0, "samples": maintenance_samples, } @@ -1004,19 +1024,38 @@ def _archive_raw_failure_info( or 0 ) samples: list[RawFailureSample] = [] + deferred_failures = 0 + terminal_rejections = 0 + unexplained_failures = 0 for row in conn.execute( """ - SELECT raw_id, origin, parse_error, validation_status, validation_error - FROM raw_sessions + SELECT r.raw_id, r.origin, r.parse_error, r.validation_status, r.validation_error, + ( + SELECT a.artifact_kind + FROM raw_artifacts AS a + WHERE a.raw_id = r.raw_id + ORDER BY a.last_observed_at_ms DESC, a.artifact_id DESC + LIMIT 1 + ) AS artifact_kind + FROM raw_sessions AS r WHERE parse_error IS NOT NULL OR validation_status = 'failed' ORDER BY acquired_at_ms DESC - LIMIT 50 """ ).fetchall(): parse_err = str(row[2] or "") if row[2] else "" val_status = str(row[3] or "") if row[3] else "" val_err = str(row[4] or "") if row[4] else "" origin = str(row[1]) if row[1] else None + artifact_kind = str(row[5]) if row[5] is not None else None + lifecycle = _raw_failure_lifecycle(artifact_kind) + if lifecycle == "deferred": + deferred_failures += 1 + elif lifecycle == "terminal": + terminal_rejections += 1 + else: + unexplained_failures += 1 + if len(samples) >= 50: + continue if "JSONDecodeError" in parse_err or "decode error" in parse_err.lower(): kind: Literal["decode_error", "parse_error", "schema_violation", "unknown"] = "decode_error" elif val_status == "failed": @@ -1030,6 +1069,7 @@ def _archive_raw_failure_info( failure_kind=kind, provider_hint=origin, redacted_error=parse_err or val_err, + lifecycle=lifecycle, ) ) finally: @@ -1044,13 +1084,47 @@ def _archive_raw_failure_info( "quarantined": quarantined, "detection_warnings": detection_warnings_count, "maintenance_failures": maintenance_count, + "deferred_failures": deferred_failures, + "terminal_rejections": terminal_rejections, + "unexplained_failures": unexplained_failures, "samples": combined, } except sqlite3.Error: return None -def _maintenance_failure_info() -> tuple[list[RawFailureSample], int]: +def _raw_failure_lifecycle(artifact_kind: str | None) -> Literal["deferred", "terminal", "unexplained"]: + """Classify only closed source-tier evidence, never a diagnostic string.""" + if artifact_kind in RAW_FAILURE_DEFERRED_EVIDENCE_KINDS: + return "deferred" + if artifact_kind in RAW_FAILURE_TERMINAL_EVIDENCE_KINDS: + return "terminal" + return "unexplained" + + +def raw_failure_info_for_root(root: Path) -> dict[str, object]: + """Return raw lifecycle evidence from one archive root without a daemon.""" + maintenance_samples, maintenance_count = _maintenance_failure_info(root) + archive_info = _archive_raw_failure_info( + root / "source.db", + maintenance_samples=maintenance_samples, + maintenance_count=maintenance_count, + ) + if archive_info is not None: + return archive_info + return { + "parse_failures": 0, + "validation_failures": 0, + "quarantined": 0, + "maintenance_failures": maintenance_count, + "deferred_failures": 0, + "terminal_rejections": 0, + "unexplained_failures": 0, + "samples": maintenance_samples, + } + + +def _maintenance_failure_info(root: Path | None = None) -> tuple[list[RawFailureSample], int]: """Read routed maintenance failures into typed daemon samples (#1198). Returns a ``(samples, total_count)`` pair so the caller can both @@ -1063,9 +1137,9 @@ def _maintenance_failure_info() -> tuple[list[RawFailureSample], int]: ) try: - root = archive_root() - records = read_maintenance_failures(root) - total = count_maintenance_failures(root) + archive = root or archive_root() + records = read_maintenance_failures(archive) + total = count_maintenance_failures(archive) except Exception as exc: logger.warning("status: maintenance-failure read failed: %s", exc, exc_info=True) return [], 0 @@ -2659,6 +2733,9 @@ def _v(name: str, default: Any) -> Any: raw_validation_failures=_safe_int(raw_failures.get("validation_failures", 0)), raw_quarantined=_safe_int(raw_failures.get("quarantined", 0)), raw_maintenance_failures=_safe_int(raw_failures.get("maintenance_failures", 0)), + raw_deferred_failures=_safe_int(raw_failures.get("deferred_failures", 0)), + raw_terminal_rejections=_safe_int(raw_failures.get("terminal_rejections", 0)), + raw_unexplained_failures=_safe_int(raw_failures.get("unexplained_failures", 0)), raw_failure_samples=_typed_failure_samples(raw_failures.get("samples")), raw_detection_warnings=_safe_int(raw_failures.get("detection_warnings", 0)), daemon_liveness=_check_daemon_liveness(daemon_lifecycle), @@ -2856,6 +2933,9 @@ def daemon_status_payload( "raw_validation_failures": status.raw_validation_failures, "raw_quarantined": status.raw_quarantined, "raw_maintenance_failures": status.raw_maintenance_failures, + "raw_deferred_failures": status.raw_deferred_failures, + "raw_terminal_rejections": status.raw_terminal_rejections, + "raw_unexplained_failures": status.raw_unexplained_failures, "raw_detection_warnings": status.raw_detection_warnings, "raw_failure_samples": [s.model_dump() for s in status.raw_failure_samples], } @@ -3264,12 +3344,18 @@ def format_daemon_status_lines(payload: JSONDocument) -> list[str]: raw_val = _safe_int(payload.get("raw_validation_failures")) raw_quarantined = _safe_int(payload.get("raw_quarantined")) raw_maintenance = _safe_int(payload.get("raw_maintenance_failures")) + raw_deferred = _safe_int(payload.get("raw_deferred_failures")) + raw_terminal = _safe_int(payload.get("raw_terminal_rejections")) + raw_unexplained = _safe_int(payload.get("raw_unexplained_failures")) total_raw = raw_parse + raw_val + raw_maintenance if total_raw > 0: breakdown = f"{raw_parse} parse + {raw_val} validation" if raw_maintenance > 0: breakdown += f" + {raw_maintenance} maintenance" lines.append(f"Raw failures: {total_raw} total ({raw_quarantined} quarantined), {breakdown}") + lines.append( + f" Lifecycle: {raw_deferred} deferred retryable, {raw_terminal} terminal, {raw_unexplained} unexplained" + ) samples = payload.get("raw_failure_samples") if isinstance(samples, list) and samples: for s in samples[:5]: diff --git a/polylogue/daemon/status_snapshot.py b/polylogue/daemon/status_snapshot.py index 74cb27b489..59e7137b8f 100644 --- a/polylogue/daemon/status_snapshot.py +++ b/polylogue/daemon/status_snapshot.py @@ -299,6 +299,9 @@ def _minimal_status_payload(*, refresh_in_progress: bool = False, refresh_error: "raw_validation_failures": 0, "raw_quarantined": 0, "raw_maintenance_failures": 0, + "raw_deferred_failures": 0, + "raw_terminal_rejections": 0, + "raw_unexplained_failures": 0, "raw_detection_warnings": 0, "raw_failure_samples": [], "status_snapshot": { diff --git a/polylogue/sources/live/batch.py b/polylogue/sources/live/batch.py index 7cede39437..64901f9a7e 100644 --- a/polylogue/sources/live/batch.py +++ b/polylogue/sources/live/batch.py @@ -47,6 +47,7 @@ read_peak_rss_self_mb, ) from polylogue.core.provider_identity import canonical_acquisition_provider +from polylogue.core.raw_failure_evidence import RawFailureEvidenceKind from polylogue.logging import get_logger from polylogue.pipeline.ids import session_revision_projection from polylogue.pipeline.ingest_outcomes import ( @@ -164,6 +165,26 @@ def _file_observation(stat: os.stat_result) -> tuple[int, int, int, int, int]: return stat.st_dev, stat.st_ino, stat.st_size, stat.st_mtime_ns, stat.st_ctime_ns +def _hot_capture_prefix_is_proven(path: str, payload: bytes | None, *, blob_size: int) -> bool: + """Prove a rejected JSONL capture is a live prefix, never merely assume it. + + A later source size alone is insufficient because a rewrite can have the + same pathname. The retained bytes must still be the exact current prefix + and the source must have grown beyond them. + """ + if payload is None: + return False + source = Path(path) + try: + stat = source.stat() + if stat.st_size <= blob_size: + return False + fingerprint, _bytes_read = sha256_range_from_path(source, start_offset=0, end_offset=blob_size) + except (EOFError, OSError): + return False + return fingerprint == sha256(payload).hexdigest() + + def _write_codex_thread_state_evidence( archive: Any, snapshot: codex_state.CodexStateSnapshot, @@ -2164,7 +2185,43 @@ def _ingest_full_records_archive( blob_hash=blob_hash, blob_size=record.blob_size, ): - raise ValueError("captured JSONL payload ends before a complete record boundary") + if _hot_capture_prefix_is_proven( + record.source_path, + payload, + blob_size=record.blob_size, + ): + archive.record_raw_failure_evidence( + source_raw_id, + provider=provider, + source_path=record.source_path, + source_index=record.source_index or 0, + acquired_at_ms=acquired_at_ms, + kind=RawFailureEvidenceKind.DEFERRED_HOT_JSONL_CAPTURE, + ) + archive.mark_raw_parse_failed( + source_raw_id, + provider=provider, + error=ValueError("captured JSONL payload ends before a complete record boundary"), + ) + result.raw_ids[record.raw_id] = source_raw_id + _accumulate_stage_timings(result.stage_timings_s, record_timings) + continue + archive.record_raw_failure_evidence( + source_raw_id, + provider=provider, + source_path=record.source_path, + source_index=record.source_index or 0, + acquired_at_ms=acquired_at_ms, + kind=RawFailureEvidenceKind.TERMINAL_CORRUPT_INPUT, + ) + archive.mark_raw_parse_failed( + source_raw_id, + provider=provider, + error=ValueError("captured JSONL payload ends before a complete record boundary"), + ) + result.raw_ids[record.raw_id] = source_raw_id + _accumulate_stage_timings(result.stage_timings_s, record_timings) + continue t0 = time.perf_counter() cached_sessions = ( parsed_sessions_by_raw_id.pop(record.raw_id, None) if parsed_sessions_by_raw_id else None @@ -2260,6 +2317,14 @@ def _ingest_full_records_archive( time.perf_counter() - t0 ) if not sessions: + archive.record_raw_failure_evidence( + source_raw_id, + provider=provider, + source_path=record.source_path, + source_index=record.source_index or 0, + acquired_at_ms=acquired_at_ms, + kind=RawFailureEvidenceKind.TERMINAL_UNSUPPORTED_SHAPE, + ) archive.mark_raw_parse_failed( source_raw_id, provider=provider, @@ -2267,6 +2332,8 @@ def _ingest_full_records_archive( "parsed raw payload produced no sessions with positive conversational evidence" ), ) + result.raw_ids[record.raw_id] = source_raw_id + _accumulate_stage_timings(result.stage_timings_s, record_timings) continue record_raw_id = source_raw_id record_session_ids: list[str] = [] diff --git a/polylogue/storage/sqlite/archive_tiers/archive.py b/polylogue/storage/sqlite/archive_tiers/archive.py index 31616c07a4..ce05366e54 100644 --- a/polylogue/storage/sqlite/archive_tiers/archive.py +++ b/polylogue/storage/sqlite/archive_tiers/archive.py @@ -65,6 +65,7 @@ from polylogue.core.dates import parse_date from polylogue.core.enums import Origin, Provider from polylogue.core.json import JSONValue, require_json_value +from polylogue.core.raw_failure_evidence import RawFailureEvidenceKind from polylogue.core.refs import delegation_edge_object_id from polylogue.core.sources import origin_from_provider from polylogue.core.types import SessionId @@ -212,6 +213,7 @@ raw_revision_rebuild_selection, raw_revision_replay_adoptable, raw_revision_replay_plan, + record_raw_failure_evidence, release_provisional_full_revisions, replace_raw_membership_census, unclassified_raw_revision_rows, @@ -2524,6 +2526,26 @@ def finalize_raw_parse_state(self, raw_id: str, *, state: RawSessionStateUpdate) def mark_raw_parse_failed(self, raw_id: str, *, provider: Provider, error: BaseException) -> None: return mark_raw_parse_failed(self, raw_id, provider=provider, error=error) + def record_raw_failure_evidence( + self, + raw_id: str, + *, + provider: Provider, + source_path: str, + source_index: int, + acquired_at_ms: int, + kind: RawFailureEvidenceKind, + ) -> None: + return record_raw_failure_evidence( + self, + raw_id, + provider=provider, + source_path=source_path, + source_index=source_index, + acquired_at_ms=acquired_at_ms, + kind=kind, + ) + def mark_raw_parse_succeeded(self, raw_id: str, *, provider: Provider) -> None: return mark_raw_parse_succeeded(self, raw_id, provider=provider) diff --git a/polylogue/storage/sqlite/archive_tiers/revision_governance.py b/polylogue/storage/sqlite/archive_tiers/revision_governance.py index 6deba7b1d2..6420b79f74 100644 --- a/polylogue/storage/sqlite/archive_tiers/revision_governance.py +++ b/polylogue/storage/sqlite/archive_tiers/revision_governance.py @@ -129,6 +129,7 @@ ) from polylogue.archive.session_revision_membership import MembershipClassification, MembershipDecision from polylogue.core.enums import Origin, Provider +from polylogue.core.raw_failure_evidence import RawFailureEvidenceKind from polylogue.core.sources import origin_from_provider, provider_from_origin from polylogue.pipeline.ids import SessionRevisionProjection, session_content_hash, session_revision_projection from polylogue.pipeline.ids import session_id as make_session_id @@ -2830,6 +2831,42 @@ def mark_raw_parse_failed( finalize_raw_parse_state(store, raw_id, state=_raw_parse_failure_state(provider, error)) +def record_raw_failure_evidence( + store: RawRevisionGovernanceHost, + raw_id: str, + *, + provider: Provider, + source_path: str, + source_index: int, + acquired_at_ms: int, + kind: RawFailureEvidenceKind, +) -> None: + """Persist a closed parse-outcome classification beside retained bytes.""" + from hashlib import sha256 + + from polylogue.core.sources import origin_from_provider + from polylogue.storage.sqlite.archive_tiers.source_write import ArchiveSourceArtifact, upsert_raw_artifact + + artifact_id = "raw-failure:" + sha256(f"{raw_id}:{kind.value}".encode()).hexdigest() + upsert_raw_artifact( + store._ensure_source_conn(), + raw_id, + ArchiveSourceArtifact( + artifact_id=artifact_id, + origin=origin_from_provider(provider), + source_path=source_path, + source_index=source_index, + artifact_kind=kind.value, + classification_reason=kind.value, + support_status=kind.support_status, + parse_as_session=kind is RawFailureEvidenceKind.DEFERRED_HOT_JSONL_CAPTURE, + schema_eligible=kind is RawFailureEvidenceKind.DEFERRED_HOT_JSONL_CAPTURE, + first_observed_at_ms=acquired_at_ms, + last_observed_at_ms=acquired_at_ms, + ), + ) + + def mark_raw_parse_succeeded(store: RawRevisionGovernanceHost, raw_id: str, *, provider: Provider) -> None: """Finalize one retained raw payload after every derived session commits.""" finalize_raw_parse_state(store, raw_id, state=_raw_parse_success_state(provider)) diff --git a/polylogue/storage/sqlite/archive_tiers/source_write.py b/polylogue/storage/sqlite/archive_tiers/source_write.py index 9601febb1c..9825ea2d57 100644 --- a/polylogue/storage/sqlite/archive_tiers/source_write.py +++ b/polylogue/storage/sqlite/archive_tiers/source_write.py @@ -9,7 +9,7 @@ import json import sqlite3 from contextlib import nullcontext -from dataclasses import dataclass +from dataclasses import dataclass, replace from datetime import UTC, datetime from typing import Literal @@ -1192,6 +1192,22 @@ def _insert_artifact(conn: sqlite3.Connection, raw_id: str, artifact: ArchiveSou malformed_jsonl_lines, decode_error, cohort_id, link_group_key, sidecar_agent_type, first_observed_at_ms, last_observed_at_ms ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(artifact_id) DO UPDATE SET + raw_id = excluded.raw_id, + origin = excluded.origin, + source_path = excluded.source_path, + source_index = excluded.source_index, + artifact_kind = excluded.artifact_kind, + support_status = excluded.support_status, + classification_reason = excluded.classification_reason, + parse_as_session = excluded.parse_as_session, + schema_eligible = excluded.schema_eligible, + malformed_jsonl_lines = excluded.malformed_jsonl_lines, + decode_error = excluded.decode_error, + cohort_id = excluded.cohort_id, + link_group_key = excluded.link_group_key, + sidecar_agent_type = excluded.sidecar_agent_type, + last_observed_at_ms = excluded.last_observed_at_ms """, ( artifact.artifact_id, @@ -1215,6 +1231,22 @@ def _insert_artifact(conn: sqlite3.Connection, raw_id: str, artifact: ArchiveSou ) +def upsert_raw_artifact(conn: sqlite3.Connection, raw_id: str, artifact: ArchiveSourceArtifact) -> None: + """Attach or refresh typed artifact evidence for an existing raw row.""" + with conn: + existing = conn.execute( + """ + SELECT artifact_id + FROM raw_artifacts + WHERE origin = ? AND source_path = ? AND source_index = ? + """, + (_enum_value(artifact.origin), artifact.source_path, artifact.source_index), + ).fetchone() + if existing is not None: + artifact = replace(artifact, artifact_id=str(existing[0])) + _insert_artifact(conn, raw_id, artifact) + + def _insert_hook_event( conn: sqlite3.Connection, hook_event: ArchiveHookEvent, @@ -1328,6 +1360,7 @@ def _enum_value(value: object) -> str | None: "read_archive_raw_session_envelope", "record_capture_mode_observation", "record_excised_blob_hash", + "upsert_raw_artifact", "write_history_sidecar", "write_source_raw_session", "write_source_raw_session_blob_ref", diff --git a/tests/unit/cli/test_status.py b/tests/unit/cli/test_status.py index b12cae7899..da18ee217e 100644 --- a/tests/unit/cli/test_status.py +++ b/tests/unit/cli/test_status.py @@ -31,8 +31,10 @@ status_command, ) from polylogue.cli.shared.types import AppEnv +from polylogue.core.enums import ArtifactSupportStatus from polylogue.storage.sqlite.archive_tiers import ARCHIVE_VERSION_BY_TIER from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database +from polylogue.storage.sqlite.archive_tiers.source_write import ArchiveSourceArtifact, upsert_raw_artifact from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier from polylogue.storage.sqlite.archive_tiers.user_write import AssertionKind, upsert_assertion from tests.infra.frozen_clock import FrozenClock @@ -532,6 +534,74 @@ def test_direct_status_json_reads_archive_file_set_from_archive_tiers(self, tmp_ assert payload["messages"] == 3 assert payload["raw_records"] == 2 + def test_direct_status_json_keeps_stopped_daemon_lifecycle_visible(self, tmp_path: Path) -> None: + """The direct fallback retains retryable and unexplained distinctions.""" + env = _make_app_env() + index_db = tmp_path / "index.db" + source_db = tmp_path / "source.db" + with sqlite3.connect(index_db) as conn: + conn.executescript( + """ + CREATE TABLE sessions (session_id TEXT PRIMARY KEY, message_count INTEGER NOT NULL); + INSERT INTO sessions VALUES ('codex-session:one', 1); + """ + ) + initialize_archive_database(source_db, ArchiveTier.SOURCE) + with sqlite3.connect(source_db) as conn: + conn.execute( + """ + INSERT INTO raw_sessions ( + raw_id, origin, native_id, source_path, blob_hash, blob_size, + acquired_at_ms, parse_error, detection_warnings_json + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + "raw-deferred", + "claude-code-session", + "native-deferred", + "/data/deferred.jsonl", + bytes(32), + 32, + 1_770_000_000_000, + "captured JSONL payload ends before a complete record boundary", + "[]", + ), + ) + upsert_raw_artifact( + conn, + "raw-deferred", + ArchiveSourceArtifact( + artifact_id="deferred-evidence", + origin="claude-code-session", + source_path="/data/deferred.jsonl", + source_index=0, + artifact_kind="deferred_hot_jsonl_capture", + classification_reason="deferred_hot_jsonl_capture", + support_status=ArtifactSupportStatus.PARTIAL_DECODE, + parse_as_session=True, + schema_eligible=True, + ), + ) + + with ( + patch("polylogue.paths.db_path", return_value=tmp_path / "custom.sqlite"), + patch("polylogue.paths.archive_root", return_value=tmp_path), + ): + _show_direct_json(env) + + payload = json.loads(_combined_calls(env)) + assert payload["daemon_liveness"] is False + assert payload["raw_failures"] == { + "parse": 1, + "validation": 0, + "quarantined": 1, + "maintenance": 0, + "deferred_retryable": 1, + "terminal_rejections": 0, + "unexplained": 0, + "sample_count": 1, + } + def test_direct_status_json_reports_sqlite_maintenance_state(self, tmp_path: Path) -> None: env = _make_app_env() db_anchor = tmp_path / "custom.sqlite" diff --git a/tests/unit/daemon/test_health_check_paths.py b/tests/unit/daemon/test_health_check_paths.py index 1f7895be6a..ee8ea6db5f 100644 --- a/tests/unit/daemon/test_health_check_paths.py +++ b/tests/unit/daemon/test_health_check_paths.py @@ -496,6 +496,32 @@ def test_raw_failures_ok_degraded_and_recovery( assert good.message == "no raw failures" +def test_raw_failures_marks_deferred_work_retryable_not_unexplained( + workspace_env: dict[str, Path], + monkeypatch: pytest.MonkeyPatch, +) -> None: + import polylogue.daemon.status as status_module + + monkeypatch.setattr( + status_module, + "_raw_failure_info", + lambda: { + "parse_failures": 4, + "validation_failures": 0, + "quarantined": 4, + "maintenance_failures": 0, + "deferred_failures": 4, + "terminal_rejections": 0, + "unexplained_failures": 0, + }, + ) + + alert = _check_raw_failures_medium() + + assert alert.severity == HealthSeverity.WARNING + assert alert.message == "4 deferred retryable raw capture(s); daemon work remains pending" + + # --------------------------------------------------------------------------- # MEDIUM: stale_ingest_attempts # --------------------------------------------------------------------------- diff --git a/tests/unit/daemon/test_raw_failure_sample.py b/tests/unit/daemon/test_raw_failure_sample.py index dbe9f6d4ee..9cc832e062 100644 --- a/tests/unit/daemon/test_raw_failure_sample.py +++ b/tests/unit/daemon/test_raw_failure_sample.py @@ -10,12 +10,14 @@ import pytest from pydantic import ValidationError +from polylogue.core.enums import ArtifactSupportStatus from polylogue.daemon.status import ( DaemonStatus, RawFailureSample, _raw_failure_info, ) from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database +from polylogue.storage.sqlite.archive_tiers.source_write import ArchiveSourceArtifact, upsert_raw_artifact from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier @@ -318,6 +320,73 @@ def test_raw_failure_info_generic_parse_error_kind(self, tmp_path: Path) -> None # Non-JSON parse error → "parse_error" (the error IS a parse error, just not JSON-specific) assert sample.failure_kind == "parse_error" + def test_raw_failure_info_separates_closed_lifecycle_evidence(self, tmp_path: Path) -> None: + index_db = _seed_archive_raw_session( + tmp_path, + raw_id="raw-deferred", + origin="claude-code-session", + native_id="native-deferred", + source_path="/data/deferred.jsonl", + parse_error="captured JSONL payload ends before a complete record boundary", + acquired_at_ms=1_770_000_000_001, + ) + _seed_archive_raw_session( + tmp_path, + raw_id="raw-terminal", + origin="unknown-export", + native_id="native-terminal", + source_path="/data/terminal.json", + parse_error="parsed raw payload produced no sessions", + acquired_at_ms=1_770_000_000_002, + ) + _seed_archive_raw_session( + tmp_path, + raw_id="raw-unexplained", + origin="codex-session", + native_id="native-unexplained", + source_path="/data/unexplained.jsonl", + parse_error="raw revision CAS rejected an older accepted frontier", + acquired_at_ms=1_770_000_000_003, + ) + with sqlite3.connect(tmp_path / "source.db") as conn: + upsert_raw_artifact( + conn, + "raw-deferred", + ArchiveSourceArtifact( + artifact_id="deferred-evidence", + origin="claude-code-session", + source_path="/data/deferred.jsonl", + source_index=0, + artifact_kind="deferred_hot_jsonl_capture", + classification_reason="deferred_hot_jsonl_capture", + support_status=ArtifactSupportStatus.PARTIAL_DECODE, + parse_as_session=True, + schema_eligible=True, + ), + ) + upsert_raw_artifact( + conn, + "raw-terminal", + ArchiveSourceArtifact( + artifact_id="terminal-evidence", + origin="unknown-export", + source_path="/data/terminal.json", + source_index=0, + artifact_kind="terminal_unsupported_shape", + classification_reason="terminal_unsupported_shape", + support_status=ArtifactSupportStatus.UNSUPPORTED_PARSEABLE, + ), + ) + + with patch("polylogue.daemon.status._active_status_db_path", return_value=index_db): + info = _raw_failure_info() + + assert info["deferred_failures"] == 1 + assert info["terminal_rejections"] == 1 + assert info["unexplained_failures"] == 1 + samples = cast(list[RawFailureSample], info["samples"]) + assert {sample.lifecycle for sample in samples} == {"deferred", "terminal", "unexplained"} + def test_raw_failure_info_empty_when_no_failures(self, tmp_path: Path) -> None: db = tmp_path / "index.db" with sqlite3.connect(db) as conn: diff --git a/tests/unit/sources/test_live_batch_support.py b/tests/unit/sources/test_live_batch_support.py index 4ba40e5088..91116ea7a3 100644 --- a/tests/unit/sources/test_live_batch_support.py +++ b/tests/unit/sources/test_live_batch_support.py @@ -352,39 +352,7 @@ def test_full_ingest_empty_jsonl_is_not_misclassified_as_truncated( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: - """A zero-byte captured ``.jsonl`` must not be flagged as a mid-write - truncation. - - Before the fix, ``_captured_jsonl_ends_at_record_boundary`` returned - ``False`` whenever ``blob_size <= 0``, so every genuinely empty session - file raised "captured JSONL payload ends before a complete record - boundary" once it reached the archive write stage -- 59 raws in the live - archive carry exactly this error with ``blob_size = 0`` - (2026-07-29 raw-failure accounting). The live cases arise from a scan/ - write race (a byte sample taken at scan time observes real content, but - the file reads back as empty bytes moments later when the writer stage - re-reads it -- e.g. an atomic rewrite of the session file in between), - which is why the ordinary "is this even a session artifact" pre-filter - (``_jsonl_provider_and_session_artifact``, exercised by - ``test_full_ingest_heartbeats_small_file_groups_with_current_path`` above) - does not by itself prevent it: that decision is made before the write - stage's own read. Force the same "treat as a session artifact" decision - that pre-filter would make on real content, so this test exercises the - write stage's boundary check exactly as the race does. An empty payload - has zero records, none complete and none incomplete, so it is trivially - at a record boundary. - - polylogue-9ykn superseded this test's original outcome: an empty - capture used to "cleanly materialize as a legitimately empty parsed - session" -- exactly the silent-inflation default that bead eliminates. - The correct outcome for a zero-record capture is now the same bounded, - recorded ``require_positive_conversational_evidence`` refusal any other - zero-message parse gets (``failed``, not ``succeeded``), NOT the - misleading truncation-boundary error the original fix targeted. Both - halves of the original claim still hold: no misleading truncation - error, and no crash/quarantine loop -- just an honest "no positive - conversational evidence" outcome instead of a phantom session. - """ + """An empty session candidate is a terminal typed refusal, not a retry.""" root = tmp_path / "sessions" root.mkdir() path = root / "empty.jsonl" @@ -403,16 +371,94 @@ def test_full_ingest_empty_jsonl_is_not_misclassified_as_truncated( result = processor._ingest_full_paths_sync([path], source_name="codex") - assert result.succeeded == [] - assert result.failed == [path] + assert result.succeeded == [path] + assert result.failed == [] parsed_at_ms, parse_error = _raw_parse_state(tmp_path) assert parse_error != "captured JSONL payload ends before a complete record boundary" - # polylogue-9ykn: a zero-record capture carries no positive - # conversational evidence -- refused with a recorded, honest reason - # (never the misleading truncation-boundary error), not silently - # accepted as a phantom empty session. assert isinstance(parse_error, str) and "no sessions with positive conversational evidence" in parse_error assert parsed_at_ms is None + with sqlite3.connect(tmp_path / "source.db") as conn: + artifact = conn.execute("SELECT artifact_kind, support_status, parse_as_session FROM raw_artifacts").fetchone() + assert artifact == ("terminal_unsupported_shape", "unsupported_parseable", 0) + + +def test_full_ingest_defers_incomplete_jsonl_only_after_hot_prefix_proof( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The full-ingest route defers a capture only after byte-prefix proof. + + The empty-capture test above is the red twin. Both paths retain the raw + and advance the cursor, but only this test changes the source so its + captured bytes can be verified as a strict current prefix. + """ + from polylogue.sources.live import batch as live_batch + + root = tmp_path / "sessions" + root.mkdir() + path = root / "active.jsonl" + captured = b'{"type":"session_meta"' + path.write_bytes(captured) + db_path = tmp_path / "archive.sqlite" + processor = LiveBatchProcessor( + cast(Any, SimpleNamespace(archive_root=tmp_path, backend=SimpleNamespace(db_path=db_path))), + (WatchSource(name="codex", root=root),), + cursor=CursorStore(db_path), + parser_fingerprint="test-parser", + ) + monkeypatch.setattr( + "polylogue.sources.live.batch._jsonl_provider_and_session_artifact", + lambda _path, fallback_provider: (fallback_provider, True), + ) + captured_boundary_check = live_batch._captured_jsonl_ends_at_record_boundary + + def grow_source_after_capture(**kwargs: object) -> bool: + path.write_bytes(captured + b"\n") + return captured_boundary_check(**kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(live_batch, "_captured_jsonl_ends_at_record_boundary", grow_source_after_capture) + + result = processor._ingest_full_paths_sync([path], source_name="codex") + + assert result.succeeded == [path] + assert result.failed == [] + _parsed_at_ms, parse_error = _raw_parse_state(tmp_path) + assert isinstance(parse_error, str) and parse_error.endswith( + "captured JSONL payload ends before a complete record boundary" + ) + with sqlite3.connect(tmp_path / "source.db") as conn: + artifact = conn.execute("SELECT artifact_kind, support_status, parse_as_session FROM raw_artifacts").fetchone() + assert artifact == ("deferred_hot_jsonl_capture", "partial_decode", 1) + + +def test_full_ingest_rejects_incomplete_jsonl_without_hot_prefix_proof( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """An incomplete static capture is terminal evidence, never deferred.""" + root = tmp_path / "sessions" + root.mkdir() + path = root / "static.jsonl" + path.write_bytes(b'{"type":"session_meta"') + db_path = tmp_path / "archive.sqlite" + processor = LiveBatchProcessor( + cast(Any, SimpleNamespace(archive_root=tmp_path, backend=SimpleNamespace(db_path=db_path))), + (WatchSource(name="codex", root=root),), + cursor=CursorStore(db_path), + parser_fingerprint="test-parser", + ) + monkeypatch.setattr( + "polylogue.sources.live.batch._jsonl_provider_and_session_artifact", + lambda _path, fallback_provider: (fallback_provider, True), + ) + + result = processor._ingest_full_paths_sync([path], source_name="codex") + + assert result.succeeded == [path] + assert result.failed == [] + with sqlite3.connect(tmp_path / "source.db") as conn: + artifact = conn.execute("SELECT artifact_kind, support_status, parse_as_session FROM raw_artifacts").fetchone() + assert artifact == ("terminal_corrupt_input", "decode_failed", 0) def test_full_ingest_heartbeats_small_file_groups_with_current_path( From 1b0546d021b9a52ebb1c87293f09eb523a65fbff Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 4 Aug 2026 09:10:19 +0200 Subject: [PATCH 2/9] fix(live): replay completed terminal JSONL captures Problem: A terminal incomplete JSONL capture was retained as successful raw evidence, but its cursor enabled append-only parsing after the source completed. The append lacked the session prefix and deferred indefinitely. What changed: detect cursor prefixes backed by terminal raw-failure evidence and route later growth through full replay. The regression verifies durable terminal classification, successful raw retention, and eventual materialization of the completed source. Co-Authored-By: Codex --- polylogue/sources/live/batch.py | 41 ++++++++++++++++++- tests/unit/sources/test_live_batch_support.py | 26 +++++++----- 2 files changed, 56 insertions(+), 11 deletions(-) diff --git a/polylogue/sources/live/batch.py b/polylogue/sources/live/batch.py index 64901f9a7e..699a5273e9 100644 --- a/polylogue/sources/live/batch.py +++ b/polylogue/sources/live/batch.py @@ -47,7 +47,7 @@ read_peak_rss_self_mb, ) from polylogue.core.provider_identity import canonical_acquisition_provider -from polylogue.core.raw_failure_evidence import RawFailureEvidenceKind +from polylogue.core.raw_failure_evidence import RAW_FAILURE_TERMINAL_EVIDENCE_KINDS, RawFailureEvidenceKind from polylogue.logging import get_logger from polylogue.pipeline.ids import session_revision_projection from polylogue.pipeline.ingest_outcomes import ( @@ -3024,6 +3024,43 @@ def _resynthesize_cursor_from_source(self, path: Path) -> CursorRecord | None: mtime_ns=None, ) + def _cursor_references_terminal_raw_failure(self, path: Path, cursor: CursorRecord) -> bool: + """Return whether this cursor's complete prefix was terminally rejected. + + A terminally typed full capture is durably useful evidence, but it has + no materialized session from which an append-only parser can recover + the missing prefix. When the same file subsequently grows, route it + through full replay so the completed record is parsed with its header + and preceding messages intact. + """ + source_db = self._cursor._db_path.with_name("source.db") + if not source_db.exists() or cursor.content_fingerprint is None: + return False + placeholders = ", ".join("?" for _ in RAW_FAILURE_TERMINAL_EVIDENCE_KINDS) + try: + conn = sqlite3.connect(f"file:{source_db}?mode=ro", uri=True) + try: + return ( + conn.execute( + f""" + SELECT 1 + FROM raw_sessions AS r + JOIN raw_artifacts AS a ON a.raw_id = r.raw_id + WHERE lower(hex(r.blob_hash)) = ? + AND r.source_path = ? + AND r.parse_error IS NOT NULL + AND a.artifact_kind IN ({placeholders}) + LIMIT 1 + """, + (cursor.content_fingerprint, str(path), *sorted(RAW_FAILURE_TERMINAL_EVIDENCE_KINDS)), + ).fetchone() + is not None + ) + finally: + conn.close() + except sqlite3.Error: + return False + def _append_plan(self, path: Path, *, cursor: CursorRecord | None = None) -> _AppendPlan | _DeferredAppend | None: # Append planning is safe only for newline-delimited record streams. # Watch-source names describe acquisition routes, not file semantics: @@ -3061,6 +3098,8 @@ def _append_plan(self, path: Path, *, cursor: CursorRecord | None = None) -> _Ap or cursor.content_fingerprint is None ): return None + if self._cursor_references_terminal_raw_failure(path, cursor): + return None expected_prefix_hash = cursor_prefix_hash(cursor.tail_hash) if expected_prefix_hash is None: return None diff --git a/tests/unit/sources/test_live_batch_support.py b/tests/unit/sources/test_live_batch_support.py index 91116ea7a3..7e578e7df2 100644 --- a/tests/unit/sources/test_live_batch_support.py +++ b/tests/unit/sources/test_live_batch_support.py @@ -3126,17 +3126,20 @@ def test_incomplete_full_jsonl_capture_retries_without_losing_split_record( first = asyncio.run(processor.ingest_files([path])) assert first.full_file_count == 1 - assert first.succeeded_file_count == 0 - assert first.failed_file_count == 1 - failed_cursor = cursor.get_record(path) - assert failed_cursor is not None - assert failed_cursor.byte_offset == 0 - assert failed_cursor.content_fingerprint is None + # Retaining and classifying the captured raw bytes is a successful source + # write, even though the incomplete session is terminally unmaterialized. + assert first.succeeded_file_count == 1 + assert first.failed_file_count == 0 + captured_cursor = cursor.get_record(path) + assert captured_cursor is not None + assert captured_cursor.byte_offset == path.stat().st_size with sqlite3.connect(index_db) as conn: assert conn.execute("SELECT COUNT(*) FROM sessions").fetchone() == (0,) with sqlite3.connect(tmp_path / "source.db") as conn: parse_error = conn.execute("SELECT parse_error FROM raw_sessions").fetchone()[0] + artifact = conn.execute("SELECT artifact_kind, support_status, parse_as_session FROM raw_artifacts").fetchone() assert "complete record boundary" in str(parse_error) + assert artifact == ("terminal_corrupt_input", "decode_failed", 0) with path.open("ab") as handle: handle.write(split_record[split_at:]) @@ -3220,16 +3223,19 @@ def remove_source_after_capture(*args: Any, **kwargs: Any) -> _ArchiveFullWriteR result = asyncio.run(processor.ingest_files([path])) - assert result.succeeded_file_count == 0 - assert result.failed_file_count == 1 - failed_cursor = cursor.get_record(path) - assert failed_cursor is None or failed_cursor.byte_offset == 0 + # The acquired bytes were durably retained with a terminal classification; + # source disappearance cannot turn that completed archive write into a + # retryable transport failure. + assert result.succeeded_file_count == 1 + assert result.failed_file_count == 0 with sqlite3.connect(index_db) as conn: assert conn.execute("SELECT COUNT(*) FROM sessions").fetchone() == (0,) assert conn.execute("SELECT COUNT(*) FROM raw_revision_heads").fetchone() == (0,) with sqlite3.connect(tmp_path / "source.db") as conn: parse_error = conn.execute("SELECT parse_error FROM raw_sessions").fetchone()[0] + artifact = conn.execute("SELECT artifact_kind, support_status, parse_as_session FROM raw_artifacts").fetchone() assert "complete record boundary" in str(parse_error) + assert artifact == ("terminal_corrupt_input", "decode_failed", 0) def test_append_persistence_failure_preserves_frontier_for_next_tick( From 871a5da9aa1ebf44359bfb122c8a74a5720064c1 Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 4 Aug 2026 09:10:36 +0200 Subject: [PATCH 3/9] fix(daemon): retain legacy raw failure signals Problem: Health collectors that predate lifecycle classification omit unexplained_failures. Treating that absent field as zero hid their observed parse and validation failures. What changed: distinguish a missing lifecycle field from an explicit zero and derive the former from parse plus validation counts. Co-Authored-By: Codex --- polylogue/daemon/health.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/polylogue/daemon/health.py b/polylogue/daemon/health.py index a66556eacf..1bf57aeebb 100644 --- a/polylogue/daemon/health.py +++ b/polylogue/daemon/health.py @@ -681,7 +681,7 @@ def _check_raw_failures_medium() -> HealthAlert: deferred = int(raw_deferred) if isinstance(raw_deferred, (int, float)) else 0 raw_terminal = info.get("terminal_rejections", 0) terminal = int(raw_terminal) if isinstance(raw_terminal, (int, float)) else 0 - raw_unexplained = info.get("unexplained_failures", 0) + raw_unexplained = info.get("unexplained_failures") unexplained = int(raw_unexplained) if isinstance(raw_unexplained, (int, float)) else parse + validation total_failures = unexplained + maintenance From 413b4367003a6c64643ed3f46b2d7c4037648ae5 Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 4 Aug 2026 09:10:37 +0200 Subject: [PATCH 4/9] test(daemon): cover current liveness registry Problem: The hardcoded health scheduling contract predated two liveness checks now present in the verification registry. What changed: include excluded cursor vocabulary and stalled append cursor freshness in the explicit expected set. Co-Authored-By: Codex --- tests/unit/daemon/test_health_check_paths.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/unit/daemon/test_health_check_paths.py b/tests/unit/daemon/test_health_check_paths.py index ee8ea6db5f..2455eb5cd6 100644 --- a/tests/unit/daemon/test_health_check_paths.py +++ b/tests/unit/daemon/test_health_check_paths.py @@ -765,6 +765,8 @@ def test_archive_verification_registry_only_schedules_liveness_and_freshness_cla "planner_stats", "convergence_freshness", "user_tier_refs", + "excluded_cursor_vocabulary_honesty", + "stalled_append_cursor_freshness", } From 68fa9e3485e4e205e23d74f2f20fe8acf1450587 Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 4 Aug 2026 09:10:50 +0200 Subject: [PATCH 5/9] fix(status): catalog current facade methods Problem: The static status catalog fell behind the public async Polylogue facade, leaving similarity search and durable settings methods unclassified. What changed: route the four methods through their actual embeddings or user-tier paths and assert those classifications. Co-Authored-By: Codex --- polylogue/cli/commands/status.py | 8 ++++++++ tests/unit/cli/test_status.py | 8 ++++++++ 2 files changed, 16 insertions(+) diff --git a/polylogue/cli/commands/status.py b/polylogue/cli/commands/status.py index 2146c1891f..9e400c7ed7 100644 --- a/polylogue/cli/commands/status.py +++ b/polylogue/cli/commands/status.py @@ -438,6 +438,11 @@ def _archive_tier_files(root: Path) -> dict[str, Path]: "archive_count_sessions": ("archive_direct", "index", "current archive helper"), "archive_get_session": ("archive_direct", "index", "current archive helper"), "search": ("archive_routed", "index", "searches index.db block FTS"), + "search_similar_sessions": ( + "archive_routed", + "embeddings", + "ranks archived sessions through embeddings.db vectors", + ), "search_session_hits": ("archive_direct", "index", "projects FTS/hybrid search hits from index.db"), "search_envelope": ("archive_routed", "index", "builds envelopes from index.db"), "session_correlation_payload": ( @@ -451,6 +456,7 @@ def _archive_tier_files(root: Path) -> dict[str, Path]: "reads session-scoped usage/cost reconciliation from index.db", ), "set_metadata": ("archive_routed", "user", "writes user metadata through user.db"), + "set_setting": ("archive_routed", "user", "writes a typed durable user setting through user.db"), "stats": ("archive_routed", "index", "reads archive stats from index.db"), "storage_stats": ("archive_direct", "index", "reads lightweight archive counts from index.db"), "tool_call_latency_distribution": ( @@ -465,6 +471,8 @@ def _archive_tier_files(root: Path) -> dict[str, Path]: "index", "summarizes workflow-shape profiles from index.db", ), + "get_setting": ("archive_routed", "user", "reads one typed durable user setting from user.db"), + "list_settings": ("archive_routed", "user", "lists typed durable user settings from user.db"), "capture_assertion_candidate": ( "archive_routed", "user", diff --git a/tests/unit/cli/test_status.py b/tests/unit/cli/test_status.py index da18ee217e..81272bb020 100644 --- a/tests/unit/cli/test_status.py +++ b/tests/unit/cli/test_status.py @@ -252,6 +252,14 @@ def test_archive_facade_route_catalog_covers_public_async_facade() -> None: assert routing["routes"]["health_check"]["route"] == "archive_routed" assert routing["routes"]["get_session"]["route"] == "archive_routed" assert routing["routes"]["search"]["route"] == "archive_routed" + assert routing["routes"]["search_similar_sessions"] == { + "route": "archive_routed", + "tier": "embeddings", + "detail": "ranks archived sessions through embeddings.db vectors", + } + for method in ("get_setting", "list_settings", "set_setting"): + assert routing["routes"][method]["route"] == "archive_routed" + assert routing["routes"][method]["tier"] == "user" assert routing["unsupported_methods"] == [] From ca77b0bb3b2836eef0fb8ff01be6b6bcdb40c01d Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 4 Aug 2026 09:51:20 +0200 Subject: [PATCH 6/9] fix(live): replay deferred raw captures from source Typed raw failures retain incomplete source evidence without a materialized session. Route a later growth observation through full replay and resolve durable evidence from the configured archive root. Co-Authored-By: Codex --- polylogue/sources/live/batch.py | 31 +-- tests/unit/sources/test_live_batch_support.py | 198 +++++++++++++++++- 2 files changed, 212 insertions(+), 17 deletions(-) diff --git a/polylogue/sources/live/batch.py b/polylogue/sources/live/batch.py index 699a5273e9..905d7fc357 100644 --- a/polylogue/sources/live/batch.py +++ b/polylogue/sources/live/batch.py @@ -47,7 +47,7 @@ read_peak_rss_self_mb, ) from polylogue.core.provider_identity import canonical_acquisition_provider -from polylogue.core.raw_failure_evidence import RAW_FAILURE_TERMINAL_EVIDENCE_KINDS, RawFailureEvidenceKind +from polylogue.core.raw_failure_evidence import RAW_FAILURE_EVIDENCE_KINDS, RawFailureEvidenceKind from polylogue.logging import get_logger from polylogue.pipeline.ids import session_revision_projection from polylogue.pipeline.ingest_outcomes import ( @@ -1438,8 +1438,11 @@ def _converge_paths( def _latest_raw_fingerprint(self, path: Path) -> str | None: return self._latest_archive_tiers_raw_fingerprint(path) + def _archive_source_db_path(self) -> Path: + return Path(getattr(self._polylogue, "archive_root", self._cursor._db_path.parent)) / "source.db" + def _latest_archive_tiers_raw_fingerprint(self, path: Path) -> str | None: - source_db = self._cursor._db_path.with_name("source.db") + source_db = self._archive_source_db_path() if not source_db.exists(): return None try: @@ -2930,7 +2933,7 @@ def _resynthesize_cursor_from_source(self, path: Path) -> CursorRecord | None: or not a 'full' head -- callers fall through to the existing full-capture path exactly as before this fallback existed. """ - source_db = self._cursor._db_path.with_name("source.db") + source_db = self._archive_source_db_path() if not source_db.exists(): return None try: @@ -3024,19 +3027,19 @@ def _resynthesize_cursor_from_source(self, path: Path) -> CursorRecord | None: mtime_ns=None, ) - def _cursor_references_terminal_raw_failure(self, path: Path, cursor: CursorRecord) -> bool: - """Return whether this cursor's complete prefix was terminally rejected. + def _cursor_references_raw_failure_requiring_full_replay(self, path: Path, cursor: CursorRecord) -> bool: + """Return whether a typed raw failure invalidates append-only replay. - A terminally typed full capture is durably useful evidence, but it has - no materialized session from which an append-only parser can recover - the missing prefix. When the same file subsequently grows, route it - through full replay so the completed record is parsed with its header - and preceding messages intact. + Typed raw-failure evidence retains a source observation that did not + materialize a session. Whether it was terminally rejected or deferred + while hot, an append-only parser cannot recover its missing prefix. + When the file subsequently grows, replay the complete source so the + parser receives its header and preceding messages intact. """ - source_db = self._cursor._db_path.with_name("source.db") + source_db = self._archive_source_db_path() if not source_db.exists() or cursor.content_fingerprint is None: return False - placeholders = ", ".join("?" for _ in RAW_FAILURE_TERMINAL_EVIDENCE_KINDS) + placeholders = ", ".join("?" for _ in RAW_FAILURE_EVIDENCE_KINDS) try: conn = sqlite3.connect(f"file:{source_db}?mode=ro", uri=True) try: @@ -3052,7 +3055,7 @@ def _cursor_references_terminal_raw_failure(self, path: Path, cursor: CursorReco AND a.artifact_kind IN ({placeholders}) LIMIT 1 """, - (cursor.content_fingerprint, str(path), *sorted(RAW_FAILURE_TERMINAL_EVIDENCE_KINDS)), + (cursor.content_fingerprint, str(path), *sorted(RAW_FAILURE_EVIDENCE_KINDS)), ).fetchone() is not None ) @@ -3098,7 +3101,7 @@ def _append_plan(self, path: Path, *, cursor: CursorRecord | None = None) -> _Ap or cursor.content_fingerprint is None ): return None - if self._cursor_references_terminal_raw_failure(path, cursor): + if self._cursor_references_raw_failure_requiring_full_replay(path, cursor): return None expected_prefix_hash = cursor_prefix_hash(cursor.tail_hash) if expected_prefix_hash is None: diff --git a/tests/unit/sources/test_live_batch_support.py b/tests/unit/sources/test_live_batch_support.py index 7e578e7df2..b234fb3f36 100644 --- a/tests/unit/sources/test_live_batch_support.py +++ b/tests/unit/sources/test_live_batch_support.py @@ -21,7 +21,7 @@ RawRevisionKind, ) from polylogue.archive.session_revision_membership import MembershipClassification -from polylogue.core.enums import Provider +from polylogue.core.enums import ArtifactSupportStatus, Provider from polylogue.pipeline.ids import session_content_hash, session_revision_projection from polylogue.sources.dispatch import parse_payload from polylogue.sources.live import LiveWatcher, WatchSource @@ -47,10 +47,18 @@ from polylogue.storage.sqlite.archive_tiers import archive as archive_tier_module from polylogue.storage.sqlite.archive_tiers import revision_governance as archive_revision_governance from polylogue.storage.sqlite.archive_tiers.archive import ArchiveStore -from polylogue.storage.sqlite.archive_tiers.bootstrap import ARCHIVE_TIER_SPECS, initialize_active_archive_root +from polylogue.storage.sqlite.archive_tiers.bootstrap import ( + ARCHIVE_TIER_SPECS, + initialize_active_archive_root, + initialize_archive_database, +) from polylogue.storage.sqlite.archive_tiers.index import INDEX_SCHEMA_VERSION from polylogue.storage.sqlite.archive_tiers.source import SOURCE_SCHEMA_VERSION -from polylogue.storage.sqlite.archive_tiers.source_write import read_archive_raw_session_envelope +from polylogue.storage.sqlite.archive_tiers.source_write import ( + ArchiveSourceArtifact, + read_archive_raw_session_envelope, + upsert_raw_artifact, +) from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier _ARCHIVE_STORAGE_TIERS = ",".join(spec.tier.value for spec in ARCHIVE_TIER_SPECS.values()) @@ -3192,6 +3200,190 @@ def spy_record_convergence_debt(self: CursorStore, **kwargs: Any) -> None: ).fetchall() == [("zero",), ("one",)] +def test_deferred_full_jsonl_with_prior_session_replays_completed_snapshot( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A deferred full capture cannot resume through an append-only tail.""" + from polylogue.sources.live import batch as live_batch + + root = tmp_path / "sessions" + root.mkdir() + path = root / "prior-session.jsonl" + baseline = ( + b'{"type":"session_meta","payload":{"id":"prior-session"}}\n' + b'{"type":"response_item","payload":{"type":"message","id":"message-0","role":"user",' + b'"content":[{"type":"input_text","text":"zero"}]}}\n' + ) + completed_record = ( + b'{"type":"response_item","payload":{"type":"message","id":"message-1","role":"assistant",' + b'"content":[{"type":"output_text","text":"one"}]}}\n' + ) + split_at = len(completed_record) // 2 + path.write_bytes(baseline) + index_db = tmp_path / "index.db" + cursor = CursorStore(index_db) + processor = LiveBatchProcessor( + cast(Any, SimpleNamespace(archive_root=tmp_path, backend=SimpleNamespace(db_path=index_db))), + (WatchSource(name="codex", root=root),), + cursor=cursor, + parser_fingerprint="current-parser", + ) + + seeded = asyncio.run(processor.ingest_files([path])) + assert seeded.succeeded_file_count == 1 + with sqlite3.connect(index_db) as conn: + assert conn.execute("SELECT native_id FROM messages ORDER BY position").fetchall() == [("message-0",)] + + seed_cursor = cursor.get_record(path) + assert seed_cursor is not None + cursor.set( + path, + seed_cursor.byte_size, + byte_offset=seed_cursor.byte_offset, + last_complete_newline=seed_cursor.last_complete_newline, + parser_fingerprint="previous-parser", + content_fingerprint=seed_cursor.content_fingerprint, + tail_hash=seed_cursor.tail_hash, + source_name=seed_cursor.source_name, + st_dev=seed_cursor.st_dev, + st_ino=seed_cursor.st_ino, + mtime_ns=seed_cursor.mtime_ns, + ) + path.write_bytes(baseline + completed_record[:split_at]) + original_boundary_check = live_batch._captured_jsonl_ends_at_record_boundary + completed = False + + def complete_source_after_capture(**kwargs: object) -> bool: + nonlocal completed + if not completed: + path.write_bytes(baseline + completed_record) + completed = True + return original_boundary_check(**kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(live_batch, "_captured_jsonl_ends_at_record_boundary", complete_source_after_capture) + + deferred = asyncio.run(processor.ingest_files([path])) + + assert deferred.full_file_count == 1 + assert deferred.succeeded_file_count == 1 + with sqlite3.connect(tmp_path / "source.db") as conn: + artifact = conn.execute( + """ + SELECT a.artifact_kind, lower(hex(r.blob_hash)) + FROM raw_artifacts AS a + JOIN raw_sessions AS r ON r.raw_id = a.raw_id + WHERE a.artifact_kind = 'deferred_hot_jsonl_capture' + """ + ).fetchone() + assert artifact is not None + assert artifact[0] == "deferred_hot_jsonl_capture" + deferred_raw_fingerprint = str(artifact[1]) + completed_stat = path.stat() + cursor.set( + path, + len(baseline), + byte_offset=len(baseline), + last_complete_newline=len(baseline), + parser_fingerprint="current-parser", + content_fingerprint=deferred_raw_fingerprint, + tail_hash=_cursor_hash_authority(baseline), + source_name="codex", + st_dev=completed_stat.st_dev, + st_ino=completed_stat.st_ino, + mtime_ns=completed_stat.st_mtime_ns, + ) + + replayed = asyncio.run(processor.ingest_files([path])) + + assert replayed.full_file_count == 1 + assert replayed.append_file_count == 0 + assert replayed.succeeded_file_count == 1 + final_cursor = cursor.get_record(path) + assert final_cursor is not None + assert final_cursor.failure_count == 0 + with sqlite3.connect(index_db) as conn: + assert conn.execute("SELECT native_id FROM messages ORDER BY position").fetchall() == [ + ("message-0",), + ("message-1",), + ] + + +def test_raw_failure_cursor_guard_uses_root_source_tier_for_pointer_index(tmp_path: Path) -> None: + """The active index generation never owns durable raw-failure evidence.""" + archive_root = tmp_path / "archive" + archive_root.mkdir() + generation = tmp_path / "generation" + generation.mkdir() + index_db = generation / "index.db" + sqlite3.connect(index_db).close() + (archive_root / ".index-active-pointer").write_text(str(index_db), encoding="utf-8") + source_db = archive_root / "source.db" + initialize_archive_database(source_db, ArchiveTier.SOURCE) + path = archive_root / "sessions" / "terminal.jsonl" + path.parent.mkdir() + path.write_bytes(b'{"type":"session_meta"') + payload_hash = "ab" * 32 + with sqlite3.connect(source_db) as conn: + conn.execute( + """ + INSERT INTO raw_sessions ( + raw_id, origin, native_id, source_path, blob_hash, blob_size, + acquired_at_ms, parse_error, detection_warnings_json + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + "terminal-raw", + "codex-session", + "terminal", + str(path), + bytes.fromhex(payload_hash), + path.stat().st_size, + 1_770_000_000_000, + "captured JSONL payload ends before a complete record boundary", + "[]", + ), + ) + upsert_raw_artifact( + conn, + "terminal-raw", + ArchiveSourceArtifact( + artifact_id="terminal-evidence", + origin="codex-session", + source_path=str(path), + source_index=0, + artifact_kind="terminal_corrupt_input", + classification_reason="terminal_corrupt_input", + support_status=ArtifactSupportStatus.DECODE_FAILED, + ), + ) + cursor = CursorStore(index_db) + stat = path.stat() + cursor.set( + path, + stat.st_size, + byte_offset=stat.st_size, + last_complete_newline=stat.st_size, + parser_fingerprint="test-parser", + content_fingerprint=payload_hash, + tail_hash=_cursor_hash_authority(path.read_bytes()), + source_name="codex", + st_dev=stat.st_dev, + st_ino=stat.st_ino, + mtime_ns=stat.st_mtime_ns, + ) + processor = LiveBatchProcessor( + cast(Any, SimpleNamespace(archive_root=archive_root, backend=SimpleNamespace(db_path=index_db))), + (WatchSource(name="codex", root=path.parent),), + cursor=cursor, + parser_fingerprint="test-parser", + ) + record = cursor.get_record(path) + + assert record is not None + assert processor._cursor_references_raw_failure_requiring_full_replay(path, record) + + def test_captured_incomplete_jsonl_is_rejected_after_source_disappears( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, From 93515684198eb2fb09cd6df9fa7dcf3bc7aca385 Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 4 Aug 2026 09:51:47 +0200 Subject: [PATCH 7/9] fix(status): read raw failures from archive root An active index generation can live outside the archive root. Resolve raw-failure evidence from the durable root source tier and cover pointer-selected status reads. Co-Authored-By: Codex --- polylogue/daemon/status.py | 8 ++- tests/unit/daemon/test_raw_failure_sample.py | 60 ++++++++++++++++++-- 2 files changed, 60 insertions(+), 8 deletions(-) diff --git a/polylogue/daemon/status.py b/polylogue/daemon/status.py index 7471b365d6..d4236ff75c 100644 --- a/polylogue/daemon/status.py +++ b/polylogue/daemon/status.py @@ -838,11 +838,13 @@ def _raw_failure_info() -> dict[str, object]: distinguish ``"ingest"`` rows from ``"maintenance"`` rows without re-querying. """ + root = archive_root() dbf = _active_status_db_path() + source_db = root / "source.db" maintenance_samples, maintenance_count = _maintenance_failure_info() if not dbf.exists(): archive_info = _archive_raw_failure_info( - dbf.with_name("source.db"), + source_db, maintenance_samples=maintenance_samples, maintenance_count=maintenance_count, ) @@ -859,11 +861,11 @@ def _raw_failure_info() -> dict[str, object]: "samples": maintenance_samples, } archive_info = _archive_raw_failure_info( - dbf.with_name("source.db"), + source_db, maintenance_samples=maintenance_samples, maintenance_count=maintenance_count, ) - if dbf.with_name("source.db").exists() and archive_info is not None: + if source_db.exists() and archive_info is not None: return archive_info try: diff --git a/tests/unit/daemon/test_raw_failure_sample.py b/tests/unit/daemon/test_raw_failure_sample.py index 9cc832e062..f5c6523738 100644 --- a/tests/unit/daemon/test_raw_failure_sample.py +++ b/tests/unit/daemon/test_raw_failure_sample.py @@ -240,7 +240,10 @@ def test_raw_failure_info_reads_archive_file_set_from_archive_tiers(self, tmp_pa ) conn.commit() - with patch("polylogue.daemon.status._active_status_db_path", return_value=index_db): + with ( + patch("polylogue.daemon.status.archive_root", return_value=tmp_path), + patch("polylogue.daemon.status._active_status_db_path", return_value=index_db), + ): info = _raw_failure_info() assert info["parse_failures"] == 1 @@ -264,7 +267,10 @@ def test_raw_failure_info_samples_are_typed(self, tmp_path: Path) -> None: blob_size=1024, ) - with patch("polylogue.daemon.status._active_status_db_path", return_value=index_db): + with ( + patch("polylogue.daemon.status.archive_root", return_value=tmp_path), + patch("polylogue.daemon.status._active_status_db_path", return_value=index_db), + ): info = _raw_failure_info() samples = info["samples"] @@ -289,7 +295,10 @@ def test_raw_failure_info_schema_violation_kind(self, tmp_path: Path) -> None: blob_size=512, ) - with patch("polylogue.daemon.status._active_status_db_path", return_value=index_db): + with ( + patch("polylogue.daemon.status.archive_root", return_value=tmp_path), + patch("polylogue.daemon.status._active_status_db_path", return_value=index_db), + ): info = _raw_failure_info() samples = cast(list[RawFailureSample], info["samples"]) @@ -310,7 +319,10 @@ def test_raw_failure_info_generic_parse_error_kind(self, tmp_path: Path) -> None blob_size=256, ) - with patch("polylogue.daemon.status._active_status_db_path", return_value=index_db): + with ( + patch("polylogue.daemon.status.archive_root", return_value=tmp_path), + patch("polylogue.daemon.status._active_status_db_path", return_value=index_db), + ): info = _raw_failure_info() samples = cast(list[RawFailureSample], info["samples"]) @@ -378,7 +390,10 @@ def test_raw_failure_info_separates_closed_lifecycle_evidence(self, tmp_path: Pa ), ) - with patch("polylogue.daemon.status._active_status_db_path", return_value=index_db): + with ( + patch("polylogue.daemon.status.archive_root", return_value=tmp_path), + patch("polylogue.daemon.status._active_status_db_path", return_value=index_db), + ): info = _raw_failure_info() assert info["deferred_failures"] == 1 @@ -387,6 +402,41 @@ def test_raw_failure_info_separates_closed_lifecycle_evidence(self, tmp_path: Pa samples = cast(list[RawFailureSample], info["samples"]) assert {sample.lifecycle for sample in samples} == {"deferred", "terminal", "unexplained"} + def test_raw_failure_info_uses_root_source_tier_for_pointer_index(self, tmp_path: Path) -> None: + generation = tmp_path / "generation" + generation.mkdir() + active_index = generation / "index.db" + sqlite3.connect(active_index).close() + (tmp_path / ".index-active-pointer").write_text(str(active_index), encoding="utf-8") + _seed_archive_raw_session( + tmp_path, + raw_id="raw-terminal", + origin="codex-session", + native_id="terminal", + source_path="/data/terminal.jsonl", + parse_error="captured JSONL payload ends before a complete record boundary", + ) + with sqlite3.connect(tmp_path / "source.db") as conn: + upsert_raw_artifact( + conn, + "raw-terminal", + ArchiveSourceArtifact( + artifact_id="terminal-evidence", + origin="codex-session", + source_path="/data/terminal.jsonl", + source_index=0, + artifact_kind="terminal_corrupt_input", + classification_reason="terminal_corrupt_input", + support_status=ArtifactSupportStatus.DECODE_FAILED, + ), + ) + + with patch("polylogue.daemon.status.archive_root", return_value=tmp_path): + info = _raw_failure_info() + + assert info["parse_failures"] == 1 + assert info["terminal_rejections"] == 1 + def test_raw_failure_info_empty_when_no_failures(self, tmp_path: Path) -> None: db = tmp_path / "index.db" with sqlite3.connect(db) as conn: From 58993e011bd5bad1f2f75d86d95e0f4a2012925d Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 4 Aug 2026 10:08:47 +0200 Subject: [PATCH 8/9] fix(raw): preserve streamed failure lifecycle Verify streamed JSONL blobs against their captured hash, stream raw failure accounting, and retain every raw-failure state in direct operator summaries. Co-Authored-By: Codex --- polylogue/cli/commands/status.py | 6 +- polylogue/daemon/health.py | 3 +- polylogue/daemon/status.py | 2 +- polylogue/sources/live/batch.py | 16 +++- tests/unit/cli/test_status.py | 42 +++++++++ tests/unit/daemon/test_health_check_paths.py | 27 ++++++ tests/unit/daemon/test_raw_failure_sample.py | 94 +++++++++++++++++++ tests/unit/sources/test_live_batch_support.py | 62 ++++++++++++ 8 files changed, 246 insertions(+), 6 deletions(-) diff --git a/polylogue/cli/commands/status.py b/polylogue/cli/commands/status.py index 9e400c7ed7..f127ba7cf8 100644 --- a/polylogue/cli/commands/status.py +++ b/polylogue/cli/commands/status.py @@ -2271,7 +2271,11 @@ def _show_direct_status( env.ui.console.print(f" Messages: {msgs:,}") env.ui.console.print(f" Raw records: {raw:,}") raw_failure_status = _direct_raw_failure_status(active_root) - raw_total = raw_failure_status["raw_parse_failures"] + raw_failure_status["raw_validation_failures"] + raw_total = ( + raw_failure_status["raw_parse_failures"] + + raw_failure_status["raw_validation_failures"] + + raw_failure_status["raw_maintenance_failures"] + ) if raw_total: env.ui.console.print( " Raw failures: " diff --git a/polylogue/daemon/health.py b/polylogue/daemon/health.py index 1bf57aeebb..1d262d357b 100644 --- a/polylogue/daemon/health.py +++ b/polylogue/daemon/health.py @@ -705,7 +705,8 @@ def _check_raw_failures_medium() -> HealthAlert: ) elif total_failures == 0: severity = HealthSeverity.WARNING - message = f"{deferred} deferred retryable raw capture(s); daemon work remains pending" + terminal_context = f"; {terminal} terminal rejection(s) recorded" if terminal else "" + message = f"{deferred} deferred retryable raw capture(s){terminal_context}; daemon work remains pending" elif total_failures <= _RAW_FAILURE_WARN_COUNT: severity = HealthSeverity.WARNING message = ( diff --git a/polylogue/daemon/status.py b/polylogue/daemon/status.py index d4236ff75c..671c830b93 100644 --- a/polylogue/daemon/status.py +++ b/polylogue/daemon/status.py @@ -1043,7 +1043,7 @@ def _archive_raw_failure_info( WHERE parse_error IS NOT NULL OR validation_status = 'failed' ORDER BY acquired_at_ms DESC """ - ).fetchall(): + ): parse_err = str(row[2] or "") if row[2] else "" val_status = str(row[3] or "") if row[3] else "" val_err = str(row[4] or "") if row[4] else "" diff --git a/polylogue/sources/live/batch.py b/polylogue/sources/live/batch.py index 905d7fc357..2b8c6cb94e 100644 --- a/polylogue/sources/live/batch.py +++ b/polylogue/sources/live/batch.py @@ -165,14 +165,23 @@ def _file_observation(stat: os.stat_result) -> tuple[int, int, int, int, int]: return stat.st_dev, stat.st_ino, stat.st_size, stat.st_mtime_ns, stat.st_ctime_ns -def _hot_capture_prefix_is_proven(path: str, payload: bytes | None, *, blob_size: int) -> bool: +def _hot_capture_prefix_is_proven( + path: str, + payload: bytes | None, + *, + blob_hash: str, + blob_size: int, +) -> bool: """Prove a rejected JSONL capture is a live prefix, never merely assume it. A later source size alone is insufficient because a rewrite can have the same pathname. The retained bytes must still be the exact current prefix and the source must have grown beyond them. """ - if payload is None: + expected_fingerprint = sha256(payload).hexdigest() if payload is not None else blob_hash.lower() + if len(expected_fingerprint) != 64 or any( + character not in "0123456789abcdef" for character in expected_fingerprint + ): return False source = Path(path) try: @@ -182,7 +191,7 @@ def _hot_capture_prefix_is_proven(path: str, payload: bytes | None, *, blob_size fingerprint, _bytes_read = sha256_range_from_path(source, start_offset=0, end_offset=blob_size) except (EOFError, OSError): return False - return fingerprint == sha256(payload).hexdigest() + return fingerprint == expected_fingerprint def _write_codex_thread_state_evidence( @@ -2191,6 +2200,7 @@ def _ingest_full_records_archive( if _hot_capture_prefix_is_proven( record.source_path, payload, + blob_hash=blob_hash, blob_size=record.blob_size, ): archive.record_raw_failure_evidence( diff --git a/tests/unit/cli/test_status.py b/tests/unit/cli/test_status.py index 81272bb020..e246bcaab3 100644 --- a/tests/unit/cli/test_status.py +++ b/tests/unit/cli/test_status.py @@ -366,6 +366,48 @@ def test_direct_status_reads_archive_file_set_from_archive_tiers(self, tmp_path: assert "Messages: 2" in combined assert "Raw records: 1" in combined + def test_direct_status_counts_maintenance_raw_failures(self, tmp_path: Path) -> None: + env = _make_app_env() + db_anchor = tmp_path / "custom.sqlite" + index_db = tmp_path / "index.db" + source_db = tmp_path / "source.db" + with sqlite3.connect(index_db) as conn: + conn.executescript( + """ + CREATE TABLE sessions (session_id TEXT PRIMARY KEY, message_count INTEGER NOT NULL); + INSERT INTO sessions VALUES ('codex-session:one', 1); + """ + ) + with sqlite3.connect(source_db) as conn: + conn.executescript( + """ + CREATE TABLE raw_sessions (raw_id TEXT PRIMARY KEY); + INSERT INTO raw_sessions VALUES ('raw-1'); + """ + ) + + with ( + patch("polylogue.paths.db_path", return_value=db_anchor), + patch("polylogue.paths.archive_root", return_value=tmp_path), + patch( + "polylogue.cli.commands.status._direct_raw_failure_status", + return_value={ + "raw_parse_failures": 0, + "raw_validation_failures": 0, + "raw_quarantined": 0, + "raw_maintenance_failures": 3, + "raw_deferred_failures": 0, + "raw_terminal_rejections": 0, + "raw_unexplained_failures": 0, + "raw_failure_samples": [], + }, + ), + patch("polylogue.cli.commands.status_diagnostics.diagnose_first_run"), + ): + _show_direct_status(env) + + assert "Raw failures: 3 total" in _combined_calls(env) + def test_direct_status_reports_sqlite_maintenance_state(self, tmp_path: Path) -> None: env = _make_app_env() db_anchor = tmp_path / "custom.sqlite" diff --git a/tests/unit/daemon/test_health_check_paths.py b/tests/unit/daemon/test_health_check_paths.py index 2455eb5cd6..984ab27b16 100644 --- a/tests/unit/daemon/test_health_check_paths.py +++ b/tests/unit/daemon/test_health_check_paths.py @@ -522,6 +522,33 @@ def test_raw_failures_marks_deferred_work_retryable_not_unexplained( assert alert.message == "4 deferred retryable raw capture(s); daemon work remains pending" +def test_raw_failures_warning_names_deferred_and_terminal_states( + workspace_env: dict[str, Path], + monkeypatch: pytest.MonkeyPatch, +) -> None: + import polylogue.daemon.status as status_module + + monkeypatch.setattr( + status_module, + "_raw_failure_info", + lambda: { + "parse_failures": 5, + "validation_failures": 0, + "quarantined": 5, + "maintenance_failures": 0, + "deferred_failures": 3, + "terminal_rejections": 2, + "unexplained_failures": 0, + }, + ) + + alert = _check_raw_failures_medium() + + assert alert.severity == HealthSeverity.WARNING + assert "3 deferred" in alert.message + assert "2 terminal" in alert.message + + # --------------------------------------------------------------------------- # MEDIUM: stale_ingest_attempts # --------------------------------------------------------------------------- diff --git a/tests/unit/daemon/test_raw_failure_sample.py b/tests/unit/daemon/test_raw_failure_sample.py index f5c6523738..09181de61b 100644 --- a/tests/unit/daemon/test_raw_failure_sample.py +++ b/tests/unit/daemon/test_raw_failure_sample.py @@ -437,6 +437,100 @@ def test_raw_failure_info_uses_root_source_tier_for_pointer_index(self, tmp_path assert info["parse_failures"] == 1 assert info["terminal_rejections"] == 1 + def test_raw_failure_info_streams_lifecycle_counts_beyond_sample_cap(self, tmp_path: Path) -> None: + """Lifecycle counts cover every failed raw without bulk-fetching rows.""" + source_db = tmp_path / "source.db" + initialize_archive_database(source_db, ArchiveTier.SOURCE) + rows = [] + with sqlite3.connect(source_db) as conn: + for index in range(120): + raw_id = f"raw-{index}" + rows.append( + ( + raw_id, + "codex-session", + raw_id, + f"/data/{raw_id}.jsonl", + index.to_bytes(32, "big"), + 128, + 1_770_000_000_000 + index, + "captured JSONL payload ends before a complete record boundary", + "[]", + ) + ) + conn.executemany( + """ + INSERT INTO raw_sessions ( + raw_id, origin, native_id, source_path, blob_hash, blob_size, + acquired_at_ms, parse_error, detection_warnings_json + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + rows, + ) + for index, row in enumerate(rows): + deferred = index % 2 == 0 + upsert_raw_artifact( + conn, + str(row[0]), + ArchiveSourceArtifact( + artifact_id=f"artifact-{index}", + origin="codex-session", + source_path=str(row[3]), + source_index=0, + artifact_kind="deferred_hot_jsonl_capture" if deferred else "terminal_corrupt_input", + classification_reason="raw-failure", + support_status=( + ArtifactSupportStatus.PARTIAL_DECODE if deferred else ArtifactSupportStatus.DECODE_FAILED + ), + ), + ) + + class GuardedCursor: + def __init__(self, cursor: sqlite3.Cursor, sql: str) -> None: + self._cursor = cursor + self._sql = sql + + def fetchone(self) -> object: + return self._cursor.fetchone() + + def fetchall(self) -> list[object]: + normalized = " ".join(self._sql.split()).lower() + if "from raw_sessions as r" in normalized and "order by acquired_at_ms desc" in normalized: + raise AssertionError("raw-failure lifecycle rows must stream instead of fetchall") + return self._cursor.fetchall() + + def __iter__(self) -> object: + return iter(self._cursor) + + def __getattr__(self, name: str) -> object: + return getattr(self._cursor, name) + + class GuardedConnection: + def __init__(self, connection: sqlite3.Connection) -> None: + self._connection = connection + + def execute(self, sql: str) -> GuardedCursor: + return GuardedCursor(self._connection.execute(sql), sql) + + def close(self) -> None: + self._connection.close() + + with ( + patch("polylogue.daemon.status.archive_root", return_value=tmp_path), + patch("polylogue.daemon.status._active_status_db_path", return_value=tmp_path / "index.db"), + patch( + "polylogue.daemon.status.open_readonly_connection", + side_effect=lambda *_args, **_kwargs: GuardedConnection(sqlite3.connect(source_db)), + ), + ): + info = _raw_failure_info() + + assert info["parse_failures"] == 120 + assert info["deferred_failures"] == 60 + assert info["terminal_rejections"] == 60 + assert info["unexplained_failures"] == 0 + assert len(cast(list[RawFailureSample], info["samples"])) == 50 + def test_raw_failure_info_empty_when_no_failures(self, tmp_path: Path) -> None: db = tmp_path / "index.db" with sqlite3.connect(db) as conn: diff --git a/tests/unit/sources/test_live_batch_support.py b/tests/unit/sources/test_live_batch_support.py index b234fb3f36..526b083fc1 100644 --- a/tests/unit/sources/test_live_batch_support.py +++ b/tests/unit/sources/test_live_batch_support.py @@ -439,6 +439,68 @@ def grow_source_after_capture(**kwargs: object) -> bool: assert artifact == ("deferred_hot_jsonl_capture", "partial_decode", 1) +def test_streamed_incomplete_jsonl_capture_defers_then_replays_completed_source( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A streamed hot capture proves its retained blob against the live prefix.""" + from polylogue.sources.live import batch as live_batch + + root = tmp_path / "sessions" + root.mkdir() + path = root / "streaming-active.jsonl" + captured = b'{"type":"session_meta","payload":{"id":"streaming-active"}' + completed = ( + b'{"type":"session_meta","payload":{"id":"streaming-active"}}\n' + b'{"type":"response_item","payload":{"type":"message","id":"message-0","role":"user",' + b'"content":[{"type":"input_text","text":"complete"}]}}\n' + ) + path.write_bytes(captured) + index_db = tmp_path / "index.db" + cursor = CursorStore(index_db) + processor = LiveBatchProcessor( + cast(Any, SimpleNamespace(archive_root=tmp_path, backend=SimpleNamespace(db_path=index_db))), + (WatchSource(name="codex", root=root),), + cursor=cursor, + parser_fingerprint="test-parser", + ) + monkeypatch.setattr("polylogue.sources.live.batch._STREAMING_FULL_INGEST_BYTES", len(captured) - 1) + monkeypatch.setattr( + "polylogue.sources.live.batch._jsonl_provider_and_session_artifact", + lambda _path, fallback_provider: (fallback_provider, True), + ) + boundary_check = live_batch._captured_jsonl_ends_at_record_boundary + source_completed = False + + def complete_source_after_capture(**kwargs: object) -> bool: + nonlocal source_completed + if not source_completed: + path.write_bytes(completed) + source_completed = True + return boundary_check(**kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(live_batch, "_captured_jsonl_ends_at_record_boundary", complete_source_after_capture) + + deferred = asyncio.run(processor.ingest_files([path])) + + assert deferred.full_file_count == 1 + assert deferred.succeeded_file_count == 1 + with sqlite3.connect(tmp_path / "source.db") as conn: + artifact = conn.execute("SELECT artifact_kind FROM raw_artifacts ORDER BY last_observed_at_ms DESC").fetchone() + assert artifact == ("deferred_hot_jsonl_capture",) + + replayed = asyncio.run(processor.ingest_files([path])) + + assert replayed.full_file_count == 1 + assert replayed.append_file_count == 0 + assert replayed.succeeded_file_count == 1 + final_cursor = cursor.get_record(path) + assert final_cursor is not None + assert final_cursor.failure_count == 0 + with sqlite3.connect(index_db) as conn: + assert conn.execute("SELECT native_id FROM messages").fetchall() == [("message-0",)] + + def test_full_ingest_rejects_incomplete_jsonl_without_hot_prefix_proof( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, From 843ca77df831fcb2350cc34b55bdbfe32091c18f Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 4 Aug 2026 10:42:28 +0200 Subject: [PATCH 9/9] fix(raw): bind replay guards to durable failures Problem: raw-failure cursors compared their durable source-row identity to a blob digest, hot prefix proof accepted in-place rewrites, and stopped status followed the active index generation for durable evidence.\n\nWhat changed: bind failure cursors and replay queries through source raw IDs, use stable pre/post file observations for hot captures, aggregate lifecycle totals in SQLite with a bounded sample query, and read direct status evidence from the configured archive root.\n\nRef #3746.\n\nCo-Authored-By: Codex --- polylogue/cli/commands/status.py | 4 +- polylogue/daemon/status.py | 43 +++++++--- polylogue/sources/live/batch.py | 29 +++++-- tests/unit/cli/test_status.py | 32 ++++++++ tests/unit/daemon/test_raw_failure_sample.py | 7 +- tests/unit/sources/test_live_batch_support.py | 81 +++++++++++-------- 6 files changed, 141 insertions(+), 55 deletions(-) diff --git a/polylogue/cli/commands/status.py b/polylogue/cli/commands/status.py index f127ba7cf8..09a344bebd 100644 --- a/polylogue/cli/commands/status.py +++ b/polylogue/cli/commands/status.py @@ -1682,7 +1682,7 @@ def _show_direct_json( "next_action": diag.next_action, "diagnostic": diagnostic_payload(diag), } - payload.update(_direct_raw_failure_status(active_root)) + payload.update(_direct_raw_failure_status(root)) if active_db is not None and active_db.exists(): payload["active_db_path"] = str(active_db) try: @@ -2270,7 +2270,7 @@ def _show_direct_status( env.ui.console.print(f" Sessions: {convs:,}") env.ui.console.print(f" Messages: {msgs:,}") env.ui.console.print(f" Raw records: {raw:,}") - raw_failure_status = _direct_raw_failure_status(active_root) + raw_failure_status = _direct_raw_failure_status(root) raw_total = ( raw_failure_status["raw_parse_failures"] + raw_failure_status["raw_validation_failures"] diff --git a/polylogue/daemon/status.py b/polylogue/daemon/status.py index 671c830b93..9be33bd24c 100644 --- a/polylogue/daemon/status.py +++ b/polylogue/daemon/status.py @@ -1025,10 +1025,38 @@ def _archive_raw_failure_info( ).fetchone()[0] or 0 ) + deferred_kinds = tuple(sorted(RAW_FAILURE_DEFERRED_EVIDENCE_KINDS)) + terminal_kinds = tuple(sorted(RAW_FAILURE_TERMINAL_EVIDENCE_KINDS)) + known_kinds = deferred_kinds + terminal_kinds + deferred_placeholders = ", ".join("?" for _ in deferred_kinds) + terminal_placeholders = ", ".join("?" for _ in terminal_kinds) + known_placeholders = ", ".join("?" for _ in known_kinds) + lifecycle_totals = conn.execute( + f""" + WITH raw_failures AS ( + SELECT ( + SELECT a.artifact_kind + FROM raw_artifacts AS a + WHERE a.raw_id = r.raw_id + ORDER BY a.last_observed_at_ms DESC, a.artifact_id DESC + LIMIT 1 + ) AS artifact_kind + FROM raw_sessions AS r + WHERE r.parse_error IS NOT NULL OR r.validation_status = 'failed' + ) + SELECT + COALESCE(SUM(artifact_kind IN ({deferred_placeholders})), 0), + COALESCE(SUM(artifact_kind IN ({terminal_placeholders})), 0), + COALESCE(SUM(artifact_kind IS NULL OR artifact_kind NOT IN ({known_placeholders})), 0) + FROM raw_failures + """, + (*deferred_kinds, *terminal_kinds, *known_kinds), + ).fetchone() + assert lifecycle_totals is not None + deferred_failures = int(lifecycle_totals[0] or 0) + terminal_rejections = int(lifecycle_totals[1] or 0) + unexplained_failures = int(lifecycle_totals[2] or 0) samples: list[RawFailureSample] = [] - deferred_failures = 0 - terminal_rejections = 0 - unexplained_failures = 0 for row in conn.execute( """ SELECT r.raw_id, r.origin, r.parse_error, r.validation_status, r.validation_error, @@ -1042,6 +1070,7 @@ def _archive_raw_failure_info( FROM raw_sessions AS r WHERE parse_error IS NOT NULL OR validation_status = 'failed' ORDER BY acquired_at_ms DESC + LIMIT 50 """ ): parse_err = str(row[2] or "") if row[2] else "" @@ -1050,14 +1079,6 @@ def _archive_raw_failure_info( origin = str(row[1]) if row[1] else None artifact_kind = str(row[5]) if row[5] is not None else None lifecycle = _raw_failure_lifecycle(artifact_kind) - if lifecycle == "deferred": - deferred_failures += 1 - elif lifecycle == "terminal": - terminal_rejections += 1 - else: - unexplained_failures += 1 - if len(samples) >= 50: - continue if "JSONDecodeError" in parse_err or "decode error" in parse_err.lower(): kind: Literal["decode_error", "parse_error", "schema_violation", "unknown"] = "decode_error" elif val_status == "failed": diff --git a/polylogue/sources/live/batch.py b/polylogue/sources/live/batch.py index 2b8c6cb94e..d2aef0751a 100644 --- a/polylogue/sources/live/batch.py +++ b/polylogue/sources/live/batch.py @@ -185,13 +185,14 @@ def _hot_capture_prefix_is_proven( return False source = Path(path) try: - stat = source.stat() - if stat.st_size <= blob_size: + proof_start = source.stat() + if proof_start.st_size <= blob_size: return False fingerprint, _bytes_read = sha256_range_from_path(source, start_offset=0, end_offset=blob_size) + proof_end = source.stat() except (EOFError, OSError): return False - return fingerprint == expected_fingerprint + return fingerprint == expected_fingerprint and _file_observation(proof_start) == _file_observation(proof_end) def _write_codex_thread_state_evidence( @@ -1183,7 +1184,15 @@ def _record_full_cursor( assert prefix_proof.stat is not None stat = prefix_proof.stat if source_revision is not None: - fp = source_revision + # Ordinary append cursors use the blob-backed source revision as + # their byte-proof identity. A retained raw failure instead binds + # the cursor to its durable source-tier ID so the next growth can + # find typed failure evidence and force full replay. + fp = ( + raw_fingerprint + if raw_fingerprint is not None and self._raw_failure_requires_full_replay(path, raw_fingerprint) + else source_revision + ) last_nl = byte_size tail_hash = source_revision if captured_content_hash is not None: @@ -3046,8 +3055,14 @@ def _cursor_references_raw_failure_requiring_full_replay(self, path: Path, curso When the file subsequently grows, replay the complete source so the parser receives its header and preceding messages intact. """ + if cursor.content_fingerprint is None: + return False + return self._raw_failure_requires_full_replay(path, cursor.content_fingerprint) + + def _raw_failure_requires_full_replay(self, path: Path, raw_id: str) -> bool: + """Return whether one durable raw ID carries replay-blocking evidence.""" source_db = self._archive_source_db_path() - if not source_db.exists() or cursor.content_fingerprint is None: + if not source_db.exists(): return False placeholders = ", ".join("?" for _ in RAW_FAILURE_EVIDENCE_KINDS) try: @@ -3059,13 +3074,13 @@ def _cursor_references_raw_failure_requiring_full_replay(self, path: Path, curso SELECT 1 FROM raw_sessions AS r JOIN raw_artifacts AS a ON a.raw_id = r.raw_id - WHERE lower(hex(r.blob_hash)) = ? + WHERE r.raw_id = ? AND r.source_path = ? AND r.parse_error IS NOT NULL AND a.artifact_kind IN ({placeholders}) LIMIT 1 """, - (cursor.content_fingerprint, str(path), *sorted(RAW_FAILURE_EVIDENCE_KINDS)), + (raw_id, str(path), *sorted(RAW_FAILURE_EVIDENCE_KINDS)), ).fetchone() is not None ) diff --git a/tests/unit/cli/test_status.py b/tests/unit/cli/test_status.py index e246bcaab3..c5c5467fcd 100644 --- a/tests/unit/cli/test_status.py +++ b/tests/unit/cli/test_status.py @@ -32,6 +32,8 @@ ) from polylogue.cli.shared.types import AppEnv from polylogue.core.enums import ArtifactSupportStatus +from polylogue.maintenance.failure_routing import route_failure_sample +from polylogue.maintenance.planner import FailureSample from polylogue.storage.sqlite.archive_tiers import ARCHIVE_VERSION_BY_TIER from polylogue.storage.sqlite.archive_tiers.bootstrap import initialize_archive_database from polylogue.storage.sqlite.archive_tiers.source_write import ArchiveSourceArtifact, upsert_raw_artifact @@ -736,6 +738,36 @@ def test_direct_status_json_reports_active_archive_root_for_sibling_index(self, assert payload["archive_tiers"]["user"]["exists"] is False assert payload["raw_records"] == 1 + def test_direct_status_json_reads_maintenance_failure_from_configured_root_with_external_index( + self, + tmp_path: Path, + ) -> None: + env = _make_app_env() + configured_root = tmp_path / "archive" + generation = tmp_path / "generation" + configured_root.mkdir() + generation.mkdir() + index_db = generation / "index.db" + initialize_archive_database(configured_root / "source.db", ArchiveTier.SOURCE) + initialize_archive_database(index_db, ArchiveTier.INDEX) + (configured_root / ".index-active-pointer").write_text(str(index_db), encoding="utf-8") + route_failure_sample( + FailureSample(kind="RuntimeError", locator="target:session_insights", message="root maintenance failure"), + operation_id="root-maintenance", + archive_root=configured_root, + target="session_insights", + ) + + with ( + patch("polylogue.paths.db_path", return_value=configured_root / "index.db"), + patch("polylogue.paths.archive_root", return_value=configured_root), + ): + _show_direct_json(env) + + payload = json.loads(_combined_calls(env)) + assert payload["active_db_path"] == str(index_db) + assert payload["raw_failures"]["maintenance"] == 1 + def test_direct_status_reports_archive_surface_blockers(self, tmp_path: Path) -> None: env = _make_app_env() db_anchor = tmp_path / "custom.sqlite" diff --git a/tests/unit/daemon/test_raw_failure_sample.py b/tests/unit/daemon/test_raw_failure_sample.py index 09181de61b..3a4dfe3458 100644 --- a/tests/unit/daemon/test_raw_failure_sample.py +++ b/tests/unit/daemon/test_raw_failure_sample.py @@ -500,6 +500,9 @@ def fetchall(self) -> list[object]: return self._cursor.fetchall() def __iter__(self) -> object: + normalized = " ".join(self._sql.split()).lower() + if "from raw_sessions as r" in normalized and "limit 50" not in normalized: + raise AssertionError("raw-failure lifecycle totals must aggregate in SQL") return iter(self._cursor) def __getattr__(self, name: str) -> object: @@ -509,8 +512,8 @@ class GuardedConnection: def __init__(self, connection: sqlite3.Connection) -> None: self._connection = connection - def execute(self, sql: str) -> GuardedCursor: - return GuardedCursor(self._connection.execute(sql), sql) + def execute(self, sql: str, parameters: tuple[object, ...] = ()) -> GuardedCursor: + return GuardedCursor(self._connection.execute(sql, parameters), sql) def close(self) -> None: self._connection.close() diff --git a/tests/unit/sources/test_live_batch_support.py b/tests/unit/sources/test_live_batch_support.py index 526b083fc1..e71a93d0d5 100644 --- a/tests/unit/sources/test_live_batch_support.py +++ b/tests/unit/sources/test_live_batch_support.py @@ -3202,14 +3202,26 @@ def test_incomplete_full_jsonl_capture_retries_without_losing_split_record( assert first.failed_file_count == 0 captured_cursor = cursor.get_record(path) assert captured_cursor is not None + # The cursor records the acquired source snapshot, but raw-failure + # evidence below still prohibits an append from that frontier. assert captured_cursor.byte_offset == path.stat().st_size with sqlite3.connect(index_db) as conn: assert conn.execute("SELECT COUNT(*) FROM sessions").fetchone() == (0,) with sqlite3.connect(tmp_path / "source.db") as conn: + raw_id = conn.execute( + """ + SELECT r.raw_id + FROM raw_sessions AS r + JOIN raw_artifacts AS a ON a.raw_id = r.raw_id + WHERE a.artifact_kind = 'terminal_corrupt_input' + """ + ).fetchone()[0] parse_error = conn.execute("SELECT parse_error FROM raw_sessions").fetchone()[0] artifact = conn.execute("SELECT artifact_kind, support_status, parse_as_session FROM raw_artifacts").fetchone() assert "complete record boundary" in str(parse_error) assert artifact == ("terminal_corrupt_input", "decode_failed", 0) + assert captured_cursor.content_fingerprint == raw_id + assert processor._cursor_references_raw_failure_requiring_full_replay(path, captured_cursor) with path.open("ab") as handle: handle.write(split_record[split_at:]) @@ -3239,7 +3251,6 @@ def spy_record_convergence_debt(self: CursorStore, **kwargs: Any) -> None: assert second.failed_file_count == 0 final_cursor = cursor.get_record(path) assert final_cursor is not None - assert final_cursor.byte_offset == path.stat().st_size assert final_cursor.failure_count == 0 assert any( call.get("stage") == "fts" and call.get("subject_id") == "codex-session:split-record" for call in recorded_debt @@ -3297,20 +3308,11 @@ def test_deferred_full_jsonl_with_prior_session_replays_completed_snapshot( with sqlite3.connect(index_db) as conn: assert conn.execute("SELECT native_id FROM messages ORDER BY position").fetchall() == [("message-0",)] - seed_cursor = cursor.get_record(path) - assert seed_cursor is not None - cursor.set( - path, - seed_cursor.byte_size, - byte_offset=seed_cursor.byte_offset, - last_complete_newline=seed_cursor.last_complete_newline, + processor = LiveBatchProcessor( + cast(Any, SimpleNamespace(archive_root=tmp_path, backend=SimpleNamespace(db_path=index_db))), + (WatchSource(name="codex", root=root),), + cursor=cursor, parser_fingerprint="previous-parser", - content_fingerprint=seed_cursor.content_fingerprint, - tail_hash=seed_cursor.tail_hash, - source_name=seed_cursor.source_name, - st_dev=seed_cursor.st_dev, - st_ino=seed_cursor.st_ino, - mtime_ns=seed_cursor.mtime_ns, ) path.write_bytes(baseline + completed_record[:split_at]) original_boundary_check = live_batch._captured_jsonl_ends_at_record_boundary @@ -3332,30 +3334,13 @@ def complete_source_after_capture(**kwargs: object) -> bool: with sqlite3.connect(tmp_path / "source.db") as conn: artifact = conn.execute( """ - SELECT a.artifact_kind, lower(hex(r.blob_hash)) + SELECT a.artifact_kind FROM raw_artifacts AS a - JOIN raw_sessions AS r ON r.raw_id = a.raw_id WHERE a.artifact_kind = 'deferred_hot_jsonl_capture' """ ).fetchone() assert artifact is not None assert artifact[0] == "deferred_hot_jsonl_capture" - deferred_raw_fingerprint = str(artifact[1]) - completed_stat = path.stat() - cursor.set( - path, - len(baseline), - byte_offset=len(baseline), - last_complete_newline=len(baseline), - parser_fingerprint="current-parser", - content_fingerprint=deferred_raw_fingerprint, - tail_hash=_cursor_hash_authority(baseline), - source_name="codex", - st_dev=completed_stat.st_dev, - st_ino=completed_stat.st_ino, - mtime_ns=completed_stat.st_mtime_ns, - ) - replayed = asyncio.run(processor.ingest_files([path])) assert replayed.full_file_count == 1 @@ -3371,6 +3356,36 @@ def complete_source_after_capture(**kwargs: object) -> bool: ] +def test_hot_capture_prefix_proof_rejects_in_place_rewrite_race( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A hash read cannot prove a prefix that changed before its post-check.""" + from polylogue.sources.live import batch as live_batch + + path = tmp_path / "racing.jsonl" + captured = b'{"type":"session_meta","payload":{"id":"racing"}}\n' + path.write_bytes(captured + b'{"type":"message"}\n') + original_hash = sha256_range_from_path + + def rewrite_after_hash(*args: object, **kwargs: object) -> tuple[str, int]: + result = original_hash(*args, **kwargs) # type: ignore[arg-type] + path.write_bytes(b"x" * path.stat().st_size) + return result + + monkeypatch.setattr(live_batch, "sha256_range_from_path", rewrite_after_hash) + + assert ( + live_batch._hot_capture_prefix_is_proven( + str(path), + captured, + blob_hash=sha256(captured).hexdigest(), + blob_size=len(captured), + ) + is False + ) + + def test_raw_failure_cursor_guard_uses_root_source_tier_for_pointer_index(tmp_path: Path) -> None: """The active index generation never owns durable raw-failure evidence.""" archive_root = tmp_path / "archive" @@ -3427,7 +3442,7 @@ def test_raw_failure_cursor_guard_uses_root_source_tier_for_pointer_index(tmp_pa byte_offset=stat.st_size, last_complete_newline=stat.st_size, parser_fingerprint="test-parser", - content_fingerprint=payload_hash, + content_fingerprint="terminal-raw", tail_hash=_cursor_hash_authority(path.read_bytes()), source_name="codex", st_dev=stat.st_dev,