fix(daemon): repair raw failure lifecycle replay

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",

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. |

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.

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)

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

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] "

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",
+]

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,