fix(storage): persist typed raw failure authority (#3897)

docs/plans/layering-surface-baseline.json

@@ -229,11 +229,6 @@
     "file": "polylogue/cli/archive_query.py",
     "import": "polylogue.storage.sqlite.archive_tiers.write"
   },
-  {
-    "target": "polylogue/cli",
-    "file": "polylogue/cli/click_app.py",
-    "import": "polylogue.storage.archive_identity"
-  },
   {
     "target": "polylogue/cli",
     "file": "polylogue/cli/click_app.py",
@@ -1444,11 +1439,6 @@
     "file": "polylogue/daemon/similarity.py",
     "import": "polylogue.storage.archive_identity"
   },
-  {
-    "target": "polylogue/daemon",
-    "file": "polylogue/daemon/similarity.py",
-    "import": "polylogue.storage.sqlite.sqlite_vec_extension"
-  },
   {
     "target": "polylogue/daemon",
     "file": "polylogue/daemon/status.py",
@@ -1554,11 +1544,6 @@
     "file": "polylogue/mcp/server_prompts.py",
     "import": "polylogue.storage.sqlite.archive_tiers.archive"
   },
-  {
-    "target": "polylogue/mcp",
-    "file": "polylogue/mcp/server_prompts.py",
-    "import": "polylogue.storage.sqlite.archive_tiers.write"
-  },
   {
     "target": "polylogue/mcp",
     "file": "polylogue/mcp/server_resources.py",

docs/plans/layering.yaml

@@ -39,6 +39,8 @@ writer_modules:
         [admit_raw_and_parsed_result, apply_raw_membership_classification, apply_raw_revision_replay,
         classify_raw_revision_cohort_for_frozen_candidate,
         classify_raw_revision_cohort_for_live_watch, classify_raw_revision_cohort_for_rebuild_repair,
+        finalize_raw_parse_state,
+        mark_raw_parse_failed,
         release_provisional_full_revisions, replace_raw_membership_census,
         write_parsed_for_retained_raw, write_parsed_for_retained_raw_result, write_raw_and_parsed,
         write_raw_and_parsed_result, write_raw_blob_and_parsed, write_raw_blob_and_parsed_result]

polylogue/core/errors.py

@@ -32,6 +32,12 @@ class PolylogueError(Exception):
     http_status_code: int = HTTPStatus.INTERNAL_SERVER_ERROR
 
 
+class RawCASFrontierError(PolylogueError):
+    """Retryable compare-and-swap conflict while advancing raw authority."""
+
+    is_transient = True
+
+
 class DatabaseError(PolylogueError):
     """Base class for database errors."""
 
@@ -93,5 +99,6 @@ def __init__(self, message: str, *, readiness_status: str) -> None:
     "DatabaseError",
     "EmbeddingRetrievalNotReadyError",
     "PolylogueError",
+    "RawCASFrontierError",
     "SchemaVersionMismatchError",
 ]

polylogue/core/raw_failure_evidence.py

@@ -7,6 +7,7 @@
 
 from __future__ import annotations
 
+import json
 from enum import StrEnum
 
 from polylogue.core.enums import ArtifactSupportStatus
@@ -16,35 +17,197 @@ class RawFailureEvidenceKind(StrEnum):
     """Durable lifecycle evidence attached to a retained raw artifact."""
 
     DEFERRED_HOT_JSONL_CAPTURE = "deferred_hot_jsonl_capture"
+    DEFERRED_CLAUDE_CODE_PARTIAL_JSONL = "deferred_claude_code_partial_jsonl"
+    DEFERRED_CAS_FRONTIER = "deferred_cas_frontier"
+    # Historical rows written before CAS evidence was made provider-neutral.
+    # Keep this token readable until a backup-gated migration or re-observation
+    # receipt has converted every retained row.
+    DEFERRED_CODEX_CAS_FRONTIER = "deferred_codex_cas_frontier"
     TERMINAL_CORRUPT_INPUT = "terminal_corrupt_input"
+    TERMINAL_SUPERSEDED_DEFERRED_CAS_FRONTIER = "terminal_superseded_deferred_cas_frontier"
+    TERMINAL_UNKNOWN_JSON_DECODE = "terminal_unknown_json_decode"
+    TERMINAL_UNKNOWN_EXPORT_NO_SESSION = "terminal_unknown_export_no_session"
     TERMINAL_UNSUPPORTED_SHAPE = "terminal_unsupported_shape"
 
     @property
     def support_status(self) -> ArtifactSupportStatus:
-        if self is RawFailureEvidenceKind.DEFERRED_HOT_JSONL_CAPTURE:
+        if self is RawFailureEvidenceKind.TERMINAL_SUPERSEDED_DEFERRED_CAS_FRONTIER:
+            return ArtifactSupportStatus.UNKNOWN
+        if self in {
+            RawFailureEvidenceKind.DEFERRED_HOT_JSONL_CAPTURE,
+            RawFailureEvidenceKind.DEFERRED_CLAUDE_CODE_PARTIAL_JSONL,
+            RawFailureEvidenceKind.DEFERRED_CAS_FRONTIER,
+            RawFailureEvidenceKind.DEFERRED_CODEX_CAS_FRONTIER,
+        }:
             return ArtifactSupportStatus.PARTIAL_DECODE
-        if self is RawFailureEvidenceKind.TERMINAL_CORRUPT_INPUT:
+        if self in {
+            RawFailureEvidenceKind.TERMINAL_CORRUPT_INPUT,
+            RawFailureEvidenceKind.TERMINAL_UNKNOWN_JSON_DECODE,
+        }:
             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"
+        if self is RawFailureEvidenceKind.TERMINAL_SUPERSEDED_DEFERRED_CAS_FRONTIER:
+            return "resolution"
+        return "deferred" if self.value in RAW_FAILURE_DEFERRED_EVIDENCE_KINDS else "terminal"
+
+
+RAW_FAILURE_TRUSTED_PROVENANCE = "worker-disposition-v1"
+RAW_FAILURE_VALIDATION_FAILURE_KINDS = frozenset(
+    {
+        RawFailureEvidenceKind.TERMINAL_CORRUPT_INPUT.value,
+        RawFailureEvidenceKind.TERMINAL_UNKNOWN_JSON_DECODE.value,
+    }
+)
+
+
+def raw_failure_classification_reason(
+    *,
+    diagnostic: str | None,
+    evidence_ref: str | None,
+    outcome_code: str,
+    remediation: str | None,
+    retryable: bool | None,
+    trusted_validation_failure: bool,
+) -> str:
+    """Encode the typed carrier, including proof for a validation failure."""
+    payload: dict[str, object] = {
+        "diagnostic": diagnostic,
+        "evidence_ref": evidence_ref,
+        "outcome_code": outcome_code,
+        "remediation": remediation,
+        "retryable": retryable,
+    }
+    if trusted_validation_failure:
+        payload["provenance"] = RAW_FAILURE_TRUSTED_PROVENANCE
+    return json.dumps(payload, sort_keys=True, separators=(",", ":"))
+
+
+def has_trusted_raw_failure_provenance(
+    classification_reason: object,
+    *,
+    artifact_kind: RawFailureEvidenceKind,
+    outcome_code: object,
+) -> bool:
+    """Check the structural worker receipt required for corrupt evidence."""
+    if artifact_kind.value not in RAW_FAILURE_VALIDATION_FAILURE_KINDS:
+        return False
+    if str(outcome_code) != "corrupt_input":
+        return False
+    if not isinstance(classification_reason, str):
+        return False
+    try:
+        payload = json.loads(classification_reason)
+    except (TypeError, ValueError):
+        return False
+    return isinstance(payload, dict) and payload.get("provenance") == RAW_FAILURE_TRUSTED_PROVENANCE
+
+
+def raw_failure_outcome_code(classification_reason: object) -> object:
+    """Read the typed outcome code from a structured failure carrier."""
+    if not isinstance(classification_reason, str):
+        return None
+    try:
+        payload = json.loads(classification_reason)
+    except (TypeError, ValueError):
+        return None
+    return payload.get("outcome_code") if isinstance(payload, dict) else None
+
+
+def validated_raw_failure_evidence_kind(
+    artifact_kind: object,
+    support_status: object,
+    *,
+    validation_failed: bool,
+    classification_reason: object = None,
+    outcome_code: object = None,
+) -> RawFailureEvidenceKind | None:
+    """Return a typed kind only for a complete, self-consistent carrier.
+
+    Decode failures are reported by the worker as validation failures because
+    the payload cannot satisfy the input contract.  A matching terminal
+    corrupt-input/decode carrier explains that state; deferred evidence still
+    requires validation to have passed or been skipped.
+    """
+    if artifact_kind is None or support_status is None:
+        return None
+    try:
+        evidence_kind = RawFailureEvidenceKind(str(artifact_kind))
+    except ValueError:
+        return None
+    if evidence_kind.support_status.value != str(support_status):
+        return None
+    if validation_failed and not has_trusted_raw_failure_provenance(
+        classification_reason,
+        artifact_kind=evidence_kind,
+        outcome_code=outcome_code,
+    ):
+        return None
+    return evidence_kind
 
 
 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_DEFERRED_EVIDENCE_KINDS = frozenset(
+    {
+        RawFailureEvidenceKind.DEFERRED_HOT_JSONL_CAPTURE.value,
+        RawFailureEvidenceKind.DEFERRED_CLAUDE_CODE_PARTIAL_JSONL.value,
+        RawFailureEvidenceKind.DEFERRED_CAS_FRONTIER.value,
+        RawFailureEvidenceKind.DEFERRED_CODEX_CAS_FRONTIER.value,
+    }
+)
+# Only frontier conflicts authorize retained-raw replay. Hot captures remain
+# deferred until a complete source observation arrives; replaying their
+# truncated blob would advance the cursor past the record that later bytes
+# complete.
+RAW_FAILURE_REPLAY_AUTHORITY_EVIDENCE_KINDS = frozenset(
+    {
+        RawFailureEvidenceKind.DEFERRED_CAS_FRONTIER.value,
+        RawFailureEvidenceKind.DEFERRED_CODEX_CAS_FRONTIER.value,
+    }
+)
+# Every deferred raw-failure carrier represents a partial decode. Consumers
+# selecting retry authority must validate this companion field as well as the
+# closed kind, or contradictory rows can authorize replay.
+RAW_FAILURE_DEFERRED_SUPPORT_STATUS = ArtifactSupportStatus.PARTIAL_DECODE.value
+RAW_FAILURE_EVIDENCE_SUPPORT_STATUS_PAIRS = tuple(
+    sorted((kind.value, kind.support_status.value) for kind in RawFailureEvidenceKind)
+)
+RAW_FAILURE_LIFECYCLE_EVIDENCE_SUPPORT_STATUS_PAIRS = tuple(
+    sorted(
+        (kind.value, kind.support_status.value)
+        for kind in RawFailureEvidenceKind
+        if kind.lifecycle in {"deferred", "terminal"}
+    )
+)
 RAW_FAILURE_TERMINAL_EVIDENCE_KINDS = frozenset(
     {
         RawFailureEvidenceKind.TERMINAL_CORRUPT_INPUT.value,
+        RawFailureEvidenceKind.TERMINAL_UNKNOWN_JSON_DECODE.value,
+        RawFailureEvidenceKind.TERMINAL_UNKNOWN_EXPORT_NO_SESSION.value,
         RawFailureEvidenceKind.TERMINAL_UNSUPPORTED_SHAPE.value,
     }
 )
+RAW_FAILURE_TERMINAL_EVIDENCE_SUPPORT_STATUS_PAIRS = tuple(
+    sorted((kind.value, kind.support_status.value) for kind in RawFailureEvidenceKind if kind.lifecycle == "terminal")
+)
 
 
 __all__ = [
     "RAW_FAILURE_DEFERRED_EVIDENCE_KINDS",
+    "RAW_FAILURE_DEFERRED_SUPPORT_STATUS",
     "RAW_FAILURE_EVIDENCE_KINDS",
+    "RAW_FAILURE_EVIDENCE_SUPPORT_STATUS_PAIRS",
+    "RAW_FAILURE_LIFECYCLE_EVIDENCE_SUPPORT_STATUS_PAIRS",
+    "RAW_FAILURE_REPLAY_AUTHORITY_EVIDENCE_KINDS",
     "RAW_FAILURE_TERMINAL_EVIDENCE_KINDS",
+    "RAW_FAILURE_TERMINAL_EVIDENCE_SUPPORT_STATUS_PAIRS",
+    "RAW_FAILURE_TRUSTED_PROVENANCE",
+    "RAW_FAILURE_VALIDATION_FAILURE_KINDS",
     "RawFailureEvidenceKind",
+    "has_trusted_raw_failure_provenance",
+    "raw_failure_classification_reason",
+    "raw_failure_outcome_code",
+    "validated_raw_failure_evidence_kind",
 ]

polylogue/daemon/status.py

@@ -8,7 +8,7 @@
 import sqlite3
 import sys
 import threading
-from collections.abc import Callable
+from collections.abc import Callable, Mapping
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any, Literal, cast
@@ -21,6 +21,7 @@
 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_outcome_code, validated_raw_failure_evidence_kind
 from polylogue.core.stats import percentile
 from polylogue.daemon.catchup_status import (
     CatchupStatus as CatchupStatus,
@@ -81,6 +82,21 @@
 
 logger = get_logger(__name__)
 
+
+def _authoritative_lifecycle_artifact_kind(sample: Mapping[str, object]) -> str | None:
+    """Project a typed kind only after validating the complete lifecycle row."""
+    evidence_kind = validated_raw_failure_evidence_kind(
+        sample.get("artifact_kind"),
+        sample.get("support_status"),
+        validation_failed=str(sample.get("validation_status") or "") == "failed",
+        classification_reason=sample.get("classification_reason"),
+        outcome_code=raw_failure_outcome_code(sample.get("classification_reason")),
+    )
+    if evidence_kind is None or sample.get("lifecycle") != evidence_kind.lifecycle:
+        return None
+    return evidence_kind.value
+
+
 # Backwards-compatible alias for the stuck threshold (#1246). The "stale"
 # rollup field stays in the typed status payload to avoid breaking
 # downstream consumers, but the threshold itself is owned by
@@ -399,7 +415,21 @@ class RawFailureSample(BaseModel):
       :attr:`operation_id` and the typed planner :attr:`locator`.
     """
 
-    failure_kind: Literal["decode_error", "parse_error", "schema_violation", "maintenance", "unknown"]
+    failure_kind: Literal[
+        "decode_error",
+        "parse_error",
+        "schema_violation",
+        "maintenance",
+        "unknown",
+        "deferred_hot_jsonl_capture",
+        "deferred_claude_code_partial_jsonl",
+        "deferred_cas_frontier",
+        "deferred_codex_cas_frontier",
+        "terminal_corrupt_input",
+        "terminal_unknown_json_decode",
+        "terminal_unknown_export_no_session",
+        "terminal_unsupported_shape",
+    ]
     provider_hint: str | None = None
     redacted_error: str = ""
     source: Literal["ingest", "maintenance"] = "ingest"
@@ -933,6 +963,12 @@ def _archive_raw_failure_info(
                 for sample in lifecycle_snapshot.samples
                 if sample.get("raw_id") is not None and sample.get("lifecycle") is not None
             }
+            validated_artifact_kind_by_raw_id = {
+                str(sample["raw_id"]): artifact_kind
+                for sample in lifecycle_snapshot.samples
+                if sample.get("raw_id") is not None
+                and (artifact_kind := _authoritative_lifecycle_artifact_kind(sample)) is not None
+            }
             samples: list[RawFailureSample] = []
             rows_by_raw_id: dict[str, sqlite3.Row | tuple[object, ...]] = {}
             if sample_ids:
@@ -955,8 +991,11 @@ def _archive_raw_failure_info(
                 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
-                if "JSONDecodeError" in parse_err or "decode error" in parse_err.lower():
-                    kind: Literal["decode_error", "parse_error", "schema_violation", "unknown"] = "decode_error"
+                artifact_kind = validated_artifact_kind_by_raw_id.get(raw_id)
+                if artifact_kind is not None:
+                    kind = cast(Any, artifact_kind)
+                elif "JSONDecodeError" in parse_err or "decode error" in parse_err.lower():
+                    kind = "decode_error"
                 elif val_status == "failed":
                     kind = "schema_violation"
                 elif parse_err:

polylogue/maintenance/raw_failure_disposition_apply.py

@@ -19,7 +19,7 @@
 from typing import cast
 
 from polylogue.config import Config
-from polylogue.core.raw_failure_evidence import RawFailureEvidenceKind
+from polylogue.core.raw_failure_evidence import RawFailureEvidenceKind, raw_failure_classification_reason
 from polylogue.maintenance.offline_guard import offline_maintenance_block_reason
 from polylogue.paths import render_root
 from polylogue.storage.sqlite.archive_tiers.types import ArchiveTier
@@ -166,7 +166,20 @@ def _apply_candidate(
         (
             candidate.disposition_kind.value,
             candidate.disposition_kind.support_status.value,
-            candidate.disposition_kind.value,
+            raw_failure_classification_reason(
+                diagnostic=candidate.detail,
+                evidence_ref=f"raw-failure-disposition:{manifest_sha256}",
+                outcome_code=(
+                    "corrupt_input"
+                    if candidate.disposition_kind is RawFailureEvidenceKind.TERMINAL_CORRUPT_INPUT
+                    else "unsupported_shape"
+                ),
+                remediation="retain the reviewed terminal disposition until a forced reparse is authorized",
+                retryable=False,
+                trusted_validation_failure=(
+                    candidate.disposition_kind is RawFailureEvidenceKind.TERMINAL_CORRUPT_INPUT
+                ),
+            ),
             disposed_at_ms,
             row["artifact_id"],
         ),