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..09a344bebd 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", @@ -1549,6 +1557,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 +1569,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 +1682,7 @@ def _show_direct_json( "next_action": diag.next_action, "diagnostic": diagnostic_payload(diag), } + 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: @@ -2241,6 +2270,19 @@ 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(root) + 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: " + 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..1d262d357b 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") + 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,42 @@ 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 + 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 = ( - 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..9be33bd24c 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) @@ -830,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, ) @@ -845,14 +855,17 @@ 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( - 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: @@ -871,6 +884,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 +964,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 +977,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, } @@ -1003,20 +1025,60 @@ 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] = [] 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 "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 +1092,7 @@ def _archive_raw_failure_info( failure_kind=kind, provider_hint=origin, redacted_error=parse_err or val_err, + lifecycle=lifecycle, ) ) finally: @@ -1044,13 +1107,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 +1160,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 +2756,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 +2956,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 +3367,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..d2aef0751a 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 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 ( @@ -164,6 +165,36 @@ 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_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. + """ + 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: + 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 and _file_observation(proof_start) == _file_observation(proof_end) + + def _write_codex_thread_state_evidence( archive: Any, snapshot: codex_state.CodexStateSnapshot, @@ -1153,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: @@ -1417,8 +1456,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: @@ -2164,7 +2206,44 @@ 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_hash=blob_hash, + 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 +2339,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 +2354,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] = [] @@ -2863,7 +2952,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: @@ -2957,6 +3046,49 @@ def _resynthesize_cursor_from_source(self, path: Path) -> CursorRecord | None: mtime_ns=None, ) + def _cursor_references_raw_failure_requiring_full_replay(self, path: Path, cursor: CursorRecord) -> bool: + """Return whether a typed raw failure invalidates append-only replay. + + 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. + """ + 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(): + return False + placeholders = ", ".join("?" for _ in RAW_FAILURE_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 r.raw_id = ? + AND r.source_path = ? + AND r.parse_error IS NOT NULL + AND a.artifact_kind IN ({placeholders}) + LIMIT 1 + """, + (raw_id, str(path), *sorted(RAW_FAILURE_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: @@ -2994,6 +3126,8 @@ def _append_plan(self, path: Path, *, cursor: CursorRecord | None = None) -> _Ap or cursor.content_fingerprint is None ): return None + 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: return None 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..c5c5467fcd 100644 --- a/tests/unit/cli/test_status.py +++ b/tests/unit/cli/test_status.py @@ -31,8 +31,12 @@ status_command, ) 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 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 @@ -250,6 +254,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"] == [] @@ -356,6 +368,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" @@ -532,6 +586,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" @@ -616,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_health_check_paths.py b/tests/unit/daemon/test_health_check_paths.py index 1f7895be6a..984ab27b16 100644 --- a/tests/unit/daemon/test_health_check_paths.py +++ b/tests/unit/daemon/test_health_check_paths.py @@ -496,6 +496,59 @@ 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" + + +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 # --------------------------------------------------------------------------- @@ -739,6 +792,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", } diff --git a/tests/unit/daemon/test_raw_failure_sample.py b/tests/unit/daemon/test_raw_failure_sample.py index dbe9f6d4ee..3a4dfe3458 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 @@ -238,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 @@ -262,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"] @@ -287,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"]) @@ -308,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"]) @@ -318,6 +332,208 @@ 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.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 + 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_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_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: + 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: + return getattr(self._cursor, name) + + class GuardedConnection: + def __init__(self, connection: sqlite3.Connection) -> None: + self._connection = connection + + def execute(self, sql: str, parameters: tuple[object, ...] = ()) -> GuardedCursor: + return GuardedCursor(self._connection.execute(sql, parameters), 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 4ba40e5088..e71a93d0d5 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()) @@ -352,39 +360,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 +379,156 @@ 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_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, +) -> 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( @@ -3080,17 +3196,32 @@ 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 + # 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:]) @@ -3120,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 @@ -3143,6 +3273,194 @@ 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",)] + + 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", + ) + 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 + FROM raw_artifacts AS a + WHERE a.artifact_kind = 'deferred_hot_jsonl_capture' + """ + ).fetchone() + assert artifact is not None + assert artifact[0] == "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 ORDER BY position").fetchall() == [ + ("message-0",), + ("message-1",), + ] + + +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" + 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="terminal-raw", + 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, @@ -3174,16 +3492,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(