From 4af5aaae5d6778a48f834f56ef476aff6006626d Mon Sep 17 00:00:00 2001 From: Sinity Date: Tue, 4 Aug 2026 08:49:16 +0200 Subject: [PATCH] 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 a918c28ea1..fd3ed1a067 100644 --- a/devtools/docs_surface.py +++ b/devtools/docs_surface.py @@ -432,6 +432,12 @@ def _entry(title: str, path: str, description: str, tier: DocsTier) -> DocsEntry _entry( "Race Window Audit", "audits/2026-07-09-race-window-audit.md", "Race-window investigation record.", "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 a7229b0cb2..d91dd2579a 100644 --- a/docs/README.md +++ b/docs/README.md @@ -134,6 +134,7 @@ Start with **Guides** for a task, **Reference** for a surface contract, and **Ar | [Daemon Loop Lock-Starvation Map](audits/2026-07-09-daemon-loop-lock-starvation-map.md) | Lock-starvation investigation record. | | [Hash Boundary Census](audits/2026-07-09-hash-boundary-census.md) | Hash-boundary investigation record. | | [Race Window Audit](audits/2026-07-09-race-window-audit.md) | Race-window investigation record. | +| [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 e64404f263..3c25c5e7a4 100644 --- a/docs/audits/README.md +++ b/docs/audits/README.md @@ -10,3 +10,4 @@ and [Developer Tools](../devtools.md) references for present-tense behavior. - [Daemon loop lock-starvation map](2026-07-09-daemon-loop-lock-starvation-map.md) - [Hash boundary census](2026-07-09-hash-boundary-census.md) - [Race window audit](2026-07-09-race-window-audit.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 25937f76f4..cd63e01f9b 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(