分层记录任务失败与恢复终态

script_build_host/migrations/CHECKSUMS.sha256

@@ -2,3 +2,4 @@
 c570caabd328ca4ff0c7aaa6212f44af3bae916bcdb5f2b028e40a1f16da2ff7  migrations/versions/0002_phase_two_artifact_types.py
 1e4f94fd826ca8dfe8b0b0c1dd33470f052d371221d9d47c7d4f5638cc741f80  migrations/versions/0003_phase_three_publication_fencing.py
 ee3b6c2af035616b17fe8e0c2430f50be0a6de31fed23b076837120d07ddaa0e  migrations/versions/0004_candidate_source_identity.py
+e7c16114f6c559c88d487f2c6cc3b06453b28b2cdacfb9ef710700900d35ba1d  migrations/versions/0005_mission_failure_events.py

script_build_host/migrations/versions/0005_mission_failure_events.py

@@ -0,0 +1,65 @@
+"""Persist typed mission failures without losing the primary cause.
+
+Revision ID: 0005_mission_failure_events
+Revises: 0004_candidate_source_identity
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import mysql
+
+revision: str = "0005_mission_failure_events"
+down_revision: str | None = "0004_candidate_source_identity"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    timestamp = sa.DateTime(timezone=True).with_variant(mysql.DATETIME(fsp=6), "mysql")
+    op.create_table(
+        "script_build_mission_failure_event",
+        sa.Column(
+            "id",
+            sa.BigInteger().with_variant(sa.Integer, "sqlite"),
+            primary_key=True,
+            autoincrement=True,
+        ),
+        sa.Column("event_key", sa.String(80), nullable=False),
+        sa.Column("script_build_id", sa.BigInteger(), nullable=False),
+        sa.Column("root_trace_id", sa.String(128), nullable=False),
+        sa.Column("event_type", sa.String(32), nullable=False),
+        sa.Column("code", sa.String(128), nullable=False),
+        sa.Column("summary", sa.String(1000), nullable=False),
+        sa.Column("task_id", sa.String(128)),
+        sa.Column("attempt_id", sa.String(128)),
+        sa.Column("validation_id", sa.String(128)),
+        sa.Column("depth", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("cause_event_key", sa.String(80)),
+        sa.Column("recovery_of_event_key", sa.String(80)),
+        sa.Column("successor_attempt_id", sa.String(128)),
+        sa.Column("successor_validation_id", sa.String(128)),
+        sa.Column("metadata_json", sa.JSON(), nullable=False),
+        sa.Column("occurred_at", timestamp, nullable=False),
+        sa.UniqueConstraint("event_key", name="uq_mission_failure_event_key"),
+        sa.CheckConstraint(
+            "event_type IN ('failure_observed', 'recovery_action', 'stage_terminal')",
+            name="ck_mission_failure_event_type",
+        ),
+    )
+    op.create_index(
+        "ix_mission_failure_build",
+        "script_build_mission_failure_event",
+        ["script_build_id", "occurred_at"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        "ix_mission_failure_build",
+        table_name="script_build_mission_failure_event",
+    )
+    op.drop_table("script_build_mission_failure_event")

script_build_host/src/script_build_host/application/failure_policy.py

@@ -1,67 +1,116 @@
-"""Classify Agent-visible script-build failures in one application policy."""
+"""Exhaustive typed Host failure policy; no string inference."""
 
 from __future__ import annotations
 
 from collections.abc import Mapping
+from dataclasses import dataclass
 from typing import Any
 
 from agent import FailureDetail, FailureDisposition
 
 from script_build_host.domain.errors import ScriptBuildError
 
-_REPLAN_CODES = frozenset(
-    {
-        "INPUT_SCOPE_MISMATCH",
-        "CHILD_DECISION_INVALID",
-        "PHASE_POLICY_VIOLATION",
-        "PHASE_TWO_BOUNDARY_NOT_READY",
-        "PHASE_THREE_BOUNDARY_NOT_READY",
-        "PROTOCOL_VIOLATION",
-        "TASK_CHILD_BLOCKED",
-        "TASK_BUDGET_EXCEEDED",
-        "TASK_KIND_PRESET_MISMATCH",
-        "TASK_STATE_CONFLICT",
-        "ATTEMPT_WORKSPACE_FROZEN",
-        "CONTEXT_HANDLE_UNKNOWN",
-        "WRITE_SCOPE_VIOLATION",
-        "WRITE_SCOPE_CONFLICT",
-        "SUPERSESSION_CYCLE",
-    }
-)
-_REPAIR_CODES = frozenset(
+
+@dataclass(frozen=True, slots=True)
+class FailurePolicyCatalog:
+    dispositions: Mapping[str, FailureDisposition]
+    tool_overrides: Mapping[tuple[str, str], FailureDisposition]
+
+    def disposition_for(
+        self, code: str, *, source_tool: str
+    ) -> FailureDisposition | None:
+        return self.tool_overrides.get((code, source_tool), self.dispositions.get(code))
+
+
+_REPLAN = {
+    "ACCEPTED_INPUT_CONFLICT",
+    "ADOPTION_INVALID",
+    "ATTEMPT_WORKSPACE_FROZEN",
+    "CHILD_DECISION_INVALID",
+    "GOAL_COVERAGE_INCOMPLETE",
+    "GOAL_COVERAGE_INVALID",
+    "GOAL_SCOPE_INVALID",
+    "INPUT_SCOPE_MISMATCH",
+    "PHASE_POLICY_VIOLATION",
+    "PHASE_TWO_BOUNDARY_NOT_READY",
+    "PHASE_THREE_BOUNDARY_NOT_READY",
+    "SUPERSESSION_CYCLE",
+    "TASK_BUDGET_EXCEEDED",
+    "TASK_CAPABILITY_MISMATCH",
+    "TASK_CHILD_BLOCKED",
+    "TASK_CONTRACT_DUPLICATE",
+    "TASK_CONTRACT_INVALID",
+    "TASK_CONTRACT_NOT_EXECUTABLE",
+    "TASK_DECISION_CONFLICT",
+    "TASK_KIND_PRESET_MISMATCH",
+    "TASK_PARENT_AMBIGUOUS",
+    "TASK_TARGET_AMBIGUOUS",
+    "WRITE_SCOPE_CONFLICT",
+    "WRITE_SCOPE_VIOLATION",
+    "WORKSPACE_SEED_MISSING",
+}
+_REPAIR = {
+    "ARTIFACT_NOT_FOUND",
+    "LEGACY_ARTIFACT_NOT_FOUND",
+    "LEGACY_REFERENCE_INVALID",
+    "LEGACY_WRITE_INVALID",
+}
+_RETRY = {
+    "CONTEXT_CURSOR_INVALID",
+    "CONTEXT_CURSOR_OUT_OF_SEQUENCE",
+    "CONTEXT_CURSOR_SCOPE_MISMATCH",
+    "CONTEXT_HANDLE_FORMAT_INVALID",
+    "CONTEXT_HANDLE_INVALID",
+    "CONTEXT_HANDLE_UNAUTHORIZED",
+    "INVALID_TOOL_ARGUMENT",
+    "PUBLICATION_DEADLOCK_RETRYABLE",
+    "PUBLICATION_LOCK_TIMEOUT",
+    "RATE_LIMITED",
+    "RETRIEVAL_RESERVATION_BUSY",
+    "STALE_BASE_REVISION",
+    "STALE_WORKBENCH_STATE",
+    "TRANSIENT_CONNECTION_LOST",
+    "TRANSIENT_TIMEOUT",
+    "UPSTREAM_UNAVAILABLE",
+    "VALIDATION_EVIDENCE_INVALID",
+    "VALIDATION_EVIDENCE_UNAUTHORIZED",
+}
+_ABORT = {
+    "ARTIFACT_DIGEST_MISMATCH",
+    "ARTIFACT_OWNERSHIP_MISMATCH",
+    "BUILD_NOT_FOUND",
+    "CONTEXT_BUDGET_EXCEEDED",
+    "HOST_CONTRACT_INVARIANT_BROKEN",
+    "IDEMPOTENCY_KEY_CONFLICT",
+    "INPUT_DIGEST_MISMATCH",
+    "INTERNAL_UNCLASSIFIED_FAILURE",
+    "INVALID_CONFIGURATION",
+    "MISSION_ALREADY_OWNED",
+    "MISSION_FENCING_TOKEN_STALE",
+    "MISSION_STOP_REQUESTED",
+    "PROTOCOL_VIOLATION",
+    "PUBLICATION_READBACK_MISMATCH",
+    "REQUIRED_EVIDENCE_UNSATISFIED",
+    "TASK_CONTRACT_NOT_FOUND",
+    "UNSAFE_OUTBOUND_TARGET",
+    "UPSTREAM_AUTH_REJECTED",
+    "VALIDATION_EVIDENCE_CONFLICT",
+}
+
+FAILURE_POLICY_CATALOG = FailurePolicyCatalog(
     {
-        "LEGACY_WRITE_INVALID",
-        "LEGACY_REFERENCE_INVALID",
-        "ARTIFACT_NOT_FOUND",
-    }
-)
-_RETRY_CODES = frozenset(
+        **{code: FailureDisposition.REPLAN_TASK for code in _REPLAN},
+        **{code: FailureDisposition.REPAIR_ATTEMPT for code in _REPAIR},
+        **{code: FailureDisposition.RETRY_CALL for code in _RETRY},
+        **{code: FailureDisposition.ABORT_RUN for code in _ABORT},
+        "SCRIPT_CONTENT_INCOMPLETE": FailureDisposition.REPLAN_TASK,
+    },
     {
-        "PUBLICATION_LOCK_TIMEOUT",
-        "PUBLICATION_DEADLOCK_RETRYABLE",
-        "STALE_BASE_REVISION",
-        "STALE_WORKBENCH_STATE",
-        "VALIDATION_EVIDENCE_INVALID",
-        "CONTEXT_NOT_EXHAUSTED",
-        "CONTEXT_HANDLE_INVALID",
-        "CONTEXT_HANDLE_FORMAT_INVALID",
-        "CONTEXT_HANDLE_UNAUTHORIZED",
-        "CONTEXT_CURSOR_INVALID",
-        "TRANSIENT_TIMEOUT",
-        "TRANSIENT_CONNECTION_LOST",
-        "RATE_LIMITED",
-        "UPSTREAM_UNAVAILABLE",
-    }
-)
-_ABORT_MARKERS = (
-    "STOP",
-    "FENCING",
-    "OWNERSHIP",
-    "ALREADY_OWNED",
-    "DIGEST",
-    "RECOVERY",
-    "MIGRATION",
-    "RECONCILIATION",
+        ("SCRIPT_CONTENT_INCOMPLETE", "submit_attempt"):
+            FailureDisposition.REPAIR_ATTEMPT,
+        ("SCRIPT_CONTENT_INCOMPLETE", "save_paragraph_candidate"):
+            FailureDisposition.REPAIR_ATTEMPT,
+    },
 )
 
 
@@ -71,56 +120,41 @@ def classify_script_failure(
     source_tool: str,
     context: Mapping[str, Any] | None = None,
 ) -> FailureDetail:
-    """Return the sole Host classification for an Agent-visible failure."""
-
     if isinstance(error, ValueError) and not isinstance(error, ScriptBuildError):
         return FailureDetail(
-            code="INVALID_TOOL_ARGUMENT",
-            message=str(error),
-            disposition=FailureDisposition.RETRY_CALL,
+            "INVALID_TOOL_ARGUMENT",
+            str(error),
+            FailureDisposition.RETRY_CALL,
             source_tool=source_tool,
-            details=_failure_details(error, context),
+            details=_details(error, context),
         )
-
     code = error.code
-    if code == "SCRIPT_CONTENT_INCOMPLETE":
-        disposition = (
-            FailureDisposition.REPAIR_ATTEMPT
-            if source_tool in {"submit_attempt", "save_paragraph_candidate"}
-            else FailureDisposition.REPLAN_TASK
-        )
-    elif (
-        code in _REPLAN_CODES
-        or code.startswith("GOAL_")
-        or code.startswith("TASK_CAPABILITY_")
-        or code.startswith("TASK_CONTRACT_")
-        or code.startswith("ADOPTION_")
-    ):
-        disposition = FailureDisposition.REPLAN_TASK
-    elif code in _REPAIR_CODES:
-        disposition = FailureDisposition.REPAIR_ATTEMPT
-    elif code in _RETRY_CODES or code.startswith("CONTEXT_CURSOR_"):
-        disposition = FailureDisposition.RETRY_CALL
-    elif any(marker in code for marker in _ABORT_MARKERS):
-        disposition = FailureDisposition.ABORT_RUN
-    else:
+    disposition = FAILURE_POLICY_CATALOG.disposition_for(
+        code, source_tool=source_tool
+    )
+    details = _details(error, context)
+    message = error.summary
+    if disposition is None:
+        details["observed_code"] = code
+        code = "INTERNAL_UNCLASSIFIED_FAILURE"
+        message = "an unclassified internal failure terminated the script build"
         disposition = FailureDisposition.ABORT_RUN
     return FailureDetail(
-        code=code,
-        message=error.summary,
-        disposition=disposition,
+        code,
+        message,
+        disposition,
         source_tool=source_tool,
-        details=_failure_details(error, context),
+        details=details,
     )
 
 
-def _failure_details(
+def _details(
     error: ScriptBuildError | ValueError,
     context: Mapping[str, Any] | None,
 ) -> dict[str, Any]:
     raw = getattr(error, "details", None)
     details = dict(raw) if isinstance(raw, Mapping) else {}
-    if isinstance(context, Mapping):
+    if context:
         details.update(
             {
                 key: str(context[key])
@@ -131,4 +165,8 @@ def _failure_details(
     return details
 
 
-__all__ = ["classify_script_failure"]
+__all__ = [
+    "FAILURE_POLICY_CATALOG",
+    "FailurePolicyCatalog",
+    "classify_script_failure",
+]

script_build_host/src/script_build_host/application/mission_failure_flow.py

@@ -0,0 +1,304 @@
+"""Project runtime failures into a typed journal and one terminal root cause."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from datetime import UTC, datetime
+from typing import Any
+
+from script_build_host.application.mission_failure_resolver import (
+    MissionFailureResolver,
+)
+from script_build_host.domain.failure_journal import (
+    FailureEventType,
+    FailureSubject,
+    MissionFailureEvent,
+    RecoveryAction,
+)
+from script_build_host.domain.failure_ports import MissionFailureRepository
+
+
+class MissionFailureFlow:
+    def __init__(
+        self,
+        repository: MissionFailureRepository,
+        *,
+        resolver: MissionFailureResolver | None = None,
+    ) -> None:
+        self.repository = repository
+        self.resolver = resolver or MissionFailureResolver()
+
+    async def sync_ledger(
+        self,
+        *,
+        script_build_id: int,
+        root_trace_id: str,
+        ledger: Any,
+    ) -> tuple[MissionFailureEvent, ...]:
+        """Project immutable runtime failures and their explicit successors."""
+
+        events: list[MissionFailureEvent] = []
+        for task in ledger.tasks.values():
+            attempts = [
+                ledger.attempts[item]
+                for item in task.attempt_ids
+                if item in ledger.attempts
+            ]
+            validations = [
+                ledger.validations[item]
+                for item in task.validation_ids
+                if item in ledger.validations
+            ]
+            for attempt in attempts:
+                if attempt.failure is None:
+                    continue
+                observed = self._runtime_observed(
+                    script_build_id,
+                    root_trace_id,
+                    task.task_id,
+                    attempt,
+                    None,
+                )
+                events.append(observed)
+                successor = next(
+                    (
+                        item
+                        for item in attempts
+                        if _after(item.created_at, attempt.created_at)
+                        and getattr(item.status, "value", item.status) == "submitted"
+                        and _attempt_passed(ledger, task, item)
+                    ),
+                    None,
+                )
+                recovery = self._terminal_recovery(
+                    script_build_id,
+                    root_trace_id,
+                    task,
+                    observed,
+                    successor_attempt=successor,
+                )
+                if recovery is not None:
+                    events.append(recovery)
+            for validation in validations:
+                if validation.failure is None:
+                    continue
+                attempt = ledger.attempts[validation.attempt_id]
+                observed = self._runtime_observed(
+                    script_build_id,
+                    root_trace_id,
+                    task.task_id,
+                    attempt,
+                    validation,
+                )
+                events.append(observed)
+                successor = next(
+                    (
+                        item
+                        for item in validations
+                        if _after(item.created_at, validation.created_at)
+                        and item.attempt_id == validation.attempt_id
+                        and getattr(item.status, "value", item.status) == "completed"
+                        and getattr(item.verdict, "value", item.verdict) == "passed"
+                    ),
+                    None,
+                )
+                successor_attempt = next(
+                    (
+                        item
+                        for item in attempts
+                        if _after(item.created_at, validation.created_at)
+                        and getattr(item.status, "value", item.status)
+                        == "submitted"
+                        and _attempt_passed(ledger, task, item)
+                    ),
+                    None,
+                )
+                recovery = self._terminal_recovery(
+                    script_build_id,
+                    root_trace_id,
+                    task,
+                    observed,
+                    successor_attempt=successor_attempt,
+                    successor_validation=successor,
+                )
+                if recovery is not None:
+                    events.append(recovery)
+        unique = tuple({item.event_key: item for item in events}.values())
+        await self.repository.append(unique)
+        return unique
+
+    async def terminate(
+        self,
+        *,
+        script_build_id: int,
+        root_trace_id: str,
+        completion: Mapping[str, Any],
+        error: Exception,
+        stage_code: str,
+    ) -> str:
+        existing = list(await self.repository.list_for_build(script_build_id))
+        observed = [
+            self._observed(script_build_id, root_trace_id, item)
+            for item in completion.get("blocking_failures", ())
+            if isinstance(item, Mapping)
+        ]
+        all_observed = {
+            item.event_key: item
+            for item in (*existing, *observed)
+            if item.event_type is FailureEventType.OBSERVED
+        }
+        blocking_keys = {item.event_key for item in observed}
+        blocking_existing = [
+            item for key, item in all_observed.items() if key in blocking_keys
+        ]
+        cause_key = (
+            max(
+                blocking_existing,
+                key=lambda item: (item.depth, item.occurred_at),
+            ).event_key
+            if blocking_existing
+            else None
+        )
+        wrapper_code = str(getattr(error, "code", type(error).__name__))
+        wrapper = MissionFailureEvent.observed(
+            script_build_id=script_build_id,
+            root_trace_id=root_trace_id,
+            code=wrapper_code,
+            summary=str(error) or wrapper_code,
+            subject=FailureSubject(),
+            depth=0,
+            cause_event_key=cause_key,
+        )
+        terminal = MissionFailureEvent.stage_terminal(
+            script_build_id=script_build_id,
+            root_trace_id=root_trace_id,
+            code=stage_code,
+        )
+        new_events = {
+            item.event_key: item for item in (*observed, wrapper, terminal)
+        }
+        combined = {
+            item.event_key: item for item in (*existing, *new_events.values())
+        }
+        resolution = self.resolver.resolve(tuple(combined.values()))
+        await self.repository.terminate(
+            script_build_id=script_build_id,
+            events=tuple(
+                item for key, item in new_events.items() if key not in {
+                    event.event_key for event in existing
+                }
+            ),
+            primary_code=resolution.primary_cause.code,
+        )
+        return resolution.primary_cause.code
+
+    @staticmethod
+    def _observed(
+        script_build_id: int,
+        root_trace_id: str,
+        value: Mapping[str, Any],
+    ) -> MissionFailureEvent:
+        subject = FailureSubject(
+            task_id=_optional_text(value.get("task_id")),
+            attempt_id=_optional_text(value.get("attempt_id")),
+            validation_id=_optional_text(value.get("validation_id")),
+        )
+        return MissionFailureEvent.observed(
+            script_build_id=script_build_id,
+            root_trace_id=root_trace_id,
+            code=str(value.get("code") or "INTERNAL_UNCLASSIFIED_FAILURE"),
+            summary=str(value.get("message") or value.get("code") or "failure"),
+            subject=subject,
+            depth=2 if subject.validation_id else 1,
+        )
+
+    @staticmethod
+    def _runtime_observed(
+        script_build_id: int,
+        root_trace_id: str,
+        task_id: str,
+        attempt: Any,
+        validation: Any | None,
+    ) -> MissionFailureEvent:
+        failure = validation.failure if validation is not None else attempt.failure
+        return MissionFailureEvent.observed(
+            script_build_id=script_build_id,
+            root_trace_id=root_trace_id,
+            code=str(failure.code),
+            summary=str(failure.message),
+            subject=FailureSubject(
+                task_id,
+                attempt.attempt_id,
+                validation.validation_id if validation is not None else None,
+            ),
+            depth=2 if validation is not None else 1,
+            occurred_at=_time(
+                validation.created_at if validation is not None else attempt.created_at
+            ),
+        )
+
+    @staticmethod
+    def _terminal_recovery(
+        script_build_id: int,
+        root_trace_id: str,
+        task: Any,
+        observed: MissionFailureEvent,
+        *,
+        successor_attempt: Any | None = None,
+        successor_validation: Any | None = None,
+    ) -> MissionFailureEvent | None:
+        status = getattr(task.status, "value", task.status)
+        if status in {"cancelled", "superseded"}:
+            action = (
+                RecoveryAction.CANCELLED
+                if status == "cancelled"
+                else RecoveryAction.SUPERSEDED
+            )
+        elif successor_validation is not None:
+            action = RecoveryAction.REVALIDATION_PASSED
+        elif successor_attempt is not None:
+            action = RecoveryAction.ATTEMPT_SUCCEEDED
+        else:
+            return None
+        return MissionFailureEvent.recovery(
+            script_build_id=script_build_id,
+            root_trace_id=root_trace_id,
+            action=action,
+            recovery_of_event_key=observed.event_key,
+            subject=observed.subject,
+            successor_attempt_id=(
+                successor_attempt.attempt_id if successor_attempt is not None else None
+            ),
+            successor_validation_id=(
+                successor_validation.validation_id
+                if successor_validation is not None
+                else None
+            ),
+            occurred_at=_time(task.updated_at),
+        )
+
+
+def _optional_text(value: object) -> str | None:
+    return str(value) if value else None
+
+
+def _time(value: object) -> datetime:
+    parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
+    return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=UTC)
+
+
+def _after(left: object, right: object) -> bool:
+    return _time(left) > _time(right)
+
+
+def _attempt_passed(ledger: Any, task: Any, attempt: Any) -> bool:
+    return any(
+        validation.attempt_id == attempt.attempt_id
+        and getattr(validation.status, "value", validation.status) == "completed"
+        and getattr(validation.verdict, "value", validation.verdict) == "passed"
+        for validation_id in task.validation_ids
+        if (validation := ledger.validations.get(validation_id)) is not None
+    )
+
+
+__all__ = ["MissionFailureFlow"]

script_build_host/src/script_build_host/application/mission_failure_resolver.py

@@ -0,0 +1,79 @@
+"""Resolve a failure journal without allowing wrappers to replace causes."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from script_build_host.domain.errors import ProtocolViolation
+from script_build_host.domain.failure_journal import (
+    FailureEventType,
+    MissionFailureEvent,
+    RecoveryAction,
+)
+
+
+@dataclass(frozen=True, slots=True)
+class MissionFailureResolution:
+    primary_cause: MissionFailureEvent
+    blocking_failures: tuple[MissionFailureEvent, ...]
+    stage_terminal: MissionFailureEvent | None
+
+
+class MissionFailureResolver:
+    def resolve(
+        self, events: tuple[MissionFailureEvent, ...]
+    ) -> MissionFailureResolution:
+        by_key = {item.event_key: item for item in events}
+        if not events or len(by_key) != len(events):
+            raise ProtocolViolation("failure journal is empty or has duplicate keys")
+        observed = {
+            key: item
+            for key, item in by_key.items()
+            if item.event_type is FailureEventType.OBSERVED
+        }
+        for item in observed.values():
+            if item.cause_event_key and item.cause_event_key not in observed:
+                raise ProtocolViolation("failure cause references an unknown observed event")
+        self._require_acyclic(observed)
+        resolved: set[str] = set()
+        for item in events:
+            if item.event_type is not FailureEventType.RECOVERY:
+                continue
+            if item.recovery_of_event_key not in observed:
+                raise ProtocolViolation("recovery references an unknown observed event")
+            if RecoveryAction(item.code).resolves:
+                resolved.add(str(item.recovery_of_event_key))
+        blocking = tuple(item for key, item in observed.items() if key not in resolved)
+        if not blocking:
+            raise ProtocolViolation("mission has no unrecovered failure")
+        blocking_keys = {item.event_key for item in blocking}
+        leaves = tuple(
+            item
+            for item in blocking
+            if item.cause_event_key is None or item.cause_event_key not in blocking_keys
+        )
+        primary = max(
+            leaves, key=lambda item: (item.depth, item.occurred_at, item.event_key)
+        )
+        terminals = tuple(
+            item for item in events if item.event_type is FailureEventType.STAGE_TERMINAL
+        )
+        return MissionFailureResolution(
+            primary,
+            tuple(sorted(blocking, key=lambda item: (item.depth, item.event_key))),
+            max(terminals, key=lambda item: item.occurred_at) if terminals else None,
+        )
+
+    @staticmethod
+    def _require_acyclic(events: dict[str, MissionFailureEvent]) -> None:
+        for start in events:
+            seen: set[str] = set()
+            current: str | None = start
+            while current is not None:
+                if current in seen:
+                    raise ProtocolViolation("failure cause graph contains a cycle")
+                seen.add(current)
+                current = events[current].cause_event_key
+
+
+__all__ = ["MissionFailureResolution", "MissionFailureResolver"]

script_build_host/src/script_build_host/application/mission_service.py

@@ -49,6 +49,7 @@ from script_build_host.domain.task_refs import task_kind
 
 from .input_snapshot_service import AssembleInputRequest, ScriptInputSnapshotService
 from .mission_factory import ScriptMissionFactory
+from .mission_failure_flow import MissionFailureFlow
 
 PHASE_ONE_CAPABILITY_BOUNDARY = "PHASE_ONE_CAPABILITY_BOUNDARY"
 PHASE_TWO_CANDIDATE_PORTFOLIO_READY = "PHASE_TWO_CANDIDATE_PORTFOLIO_READY"
@@ -265,6 +266,7 @@ class ScriptMissionService:
         phase_two_model_manifest: dict[str, Any] | None = None,
         owner_lease_factory: Any | None = None,
         fenced_command_gate: Any | None = None,
+        failure_flow: MissionFailureFlow | None = None,
     ) -> None:
         self.runner = runner
         self.coordinator = coordinator
@@ -293,6 +295,7 @@ class ScriptMissionService:
         self.phase_two_model_manifest = dict(phase_two_model_manifest or {})
         self.owner_lease_factory = owner_lease_factory
         self.fenced_command_gate = fenced_command_gate
+        self.failure_flow = failure_flow
         self._runs: dict[int, asyncio.Task[None]] = {}
         self._owner_tokens: dict[int, Any] = {}
         self._phase_advance_locks: dict[int, asyncio.Lock] = {}
@@ -367,10 +370,11 @@ class ScriptMissionService:
             )
             await self.legacy_state.set_status(script_build_id, BuildStatus.RUNNING)
         except Exception as exc:
-            await self.legacy_state.set_status(
+            await self._terminate_failure(
                 script_build_id,
-                BuildStatus.FAILED,
-                error_summary=type(exc).__name__,
+                root_trace_id,
+                exc,
+                stage_code="MISSION_START_FAILED",
             )
             raise
         lease = None
@@ -439,6 +443,9 @@ class ScriptMissionService:
                 config=self.factory.build_run_config(binding, snapshot),
             )
             completion = await self.coordinator.root_completion(binding.root_trace_id)
+            await self._sync_failure_journal(
+                script_build_id, binding.root_trace_id
+            )
             if completion["status"] == TaskStatus.COMPLETED.value:
                 raise ProtocolViolation("phase-one Root must not complete")
             if (
@@ -458,10 +465,11 @@ class ScriptMissionService:
                 )
         except Exception as exc:
             if await self.legacy_state.get_status(script_build_id) != BuildStatus.STOPPING:
-                await self.legacy_state.set_status(
+                await self._terminate_failure(
                     script_build_id,
-                    BuildStatus.FAILED,
-                    error_summary=type(exc).__name__,
+                    binding.root_trace_id,
+                    exc,
+                    stage_code="PHASE_ONE_BOUNDARY_NOT_REACHED",
                 )
             raise
 
@@ -827,6 +835,9 @@ class ScriptMissionService:
                 async for _ in iterator:
                     pass
             completion = await self.coordinator.root_completion(binding.root_trace_id)
+            await self._sync_failure_journal(
+                script_build_id, binding.root_trace_id
+            )
             if (
                 completion.get("status") != TaskStatus.BLOCKED.value
                 or completion.get("blocked_reason") != PHASE_TWO_CANDIDATE_PORTFOLIO_READY
@@ -854,13 +865,54 @@ class ScriptMissionService:
             raise
         except Exception as exc:
             if await self.legacy_state.get_status(script_build_id) != BuildStatus.STOPPING:
-                await self.legacy_state.set_status(
+                await self._terminate_failure(
                     script_build_id,
-                    BuildStatus.FAILED,
-                    error_summary=getattr(exc, "code", type(exc).__name__),
+                    binding.root_trace_id,
+                    exc,
+                    stage_code="PHASE_TWO_BOUNDARY_NOT_REACHED",
                 )
             raise
 
+    async def _terminate_failure(
+        self,
+        script_build_id: int,
+        root_trace_id: str,
+        error: Exception,
+        *,
+        stage_code: str,
+    ) -> None:
+        if self.failure_flow is None:
+            await self.legacy_state.set_status(
+                script_build_id,
+                BuildStatus.FAILED,
+                error_summary=getattr(error, "code", type(error).__name__),
+            )
+            return
+        await self._sync_failure_journal(script_build_id, root_trace_id)
+        try:
+            completion = await self.coordinator.root_completion(root_trace_id)
+        except Exception:
+            completion = {}
+        await self.failure_flow.terminate(
+            script_build_id=script_build_id,
+            root_trace_id=root_trace_id,
+            completion=completion,
+            error=error,
+            stage_code=stage_code,
+        )
+
+    async def _sync_failure_journal(
+        self, script_build_id: int, root_trace_id: str
+    ) -> None:
+        if self.failure_flow is None:
+            return
+        ledger = await self.coordinator.task_store.load(root_trace_id)
+        await self.failure_flow.sync_ledger(
+            script_build_id=script_build_id,
+            root_trace_id=root_trace_id,
+            ledger=ledger,
+        )
+
     async def _verify_phase_two_preconditions(
         self, script_build_id: int, binding: Any, snapshot: Any
     ) -> int:
@@ -876,6 +928,7 @@ class ScriptMissionService:
         if status is not BuildStatus.PARTIAL:
             raise PhaseTwoBoundaryNotReady("build is not at a partial checkpoint")
         completion = await self.coordinator.root_completion(binding.root_trace_id)
+        await self._sync_failure_journal(script_build_id, binding.root_trace_id)
         if (
             completion.get("status") != TaskStatus.BLOCKED.value
             or completion.get("blocked_reason") != PHASE_ONE_CAPABILITY_BOUNDARY

script_build_host/src/script_build_host/application/recovery.py

@@ -13,8 +13,13 @@ from script_build_host.application.phase_two_planning import (
     PHASE_TWO_CANDIDATE_PORTFOLIO_READY,
 )
 from script_build_host.application.publisher import ScriptPublisher
+from script_build_host.application.recovery_planner import (
+    MissionRecoveryPlanner,
+    RecoveryCommandKind,
+)
 from script_build_host.domain.artifacts import ArtifactKind, ArtifactState
 from script_build_host.domain.errors import ProtocolViolation, PublicationStateInconsistent
+from script_build_host.domain.failure_ports import MissionFailureRepository
 from script_build_host.domain.phase_three_artifacts import RootDeliveryManifestV1
 from script_build_host.domain.ports import (
     BuildAuthorizer,
@@ -104,6 +109,8 @@ class PhaseThreeRecoveryService:
         legacy_projection: Any | None = None,
         mission_service: Any | None = None,
         contract_reader: Any | None = None,
+        failure_repository: MissionFailureRepository | None = None,
+        recovery_planner: MissionRecoveryPlanner | None = None,
     ) -> None:
         self.coordinator = coordinator
         self.bindings = bindings
@@ -116,6 +123,8 @@ class PhaseThreeRecoveryService:
         self.legacy_projection = legacy_projection
         self.mission_service = mission_service
         self.contract_reader = contract_reader
+        self.failure_repository = failure_repository
+        self.recovery_planner = recovery_planner or MissionRecoveryPlanner()
 
     async def resume(self, script_build_id: int, principal: Principal) -> ResumeResult:
         await self.authorizer.require_access(principal, script_build_id)
@@ -141,6 +150,38 @@ class PhaseThreeRecoveryService:
                 status,
                 "a publishing transaction requires explicit reconciliation",
             )
+        if status is BuildStatus.FAILED:
+            if self.failure_repository is None:
+                return ResumeResult(
+                    script_build_id,
+                    RecoveryClassification.MANUAL_RECONCILIATION,
+                    status,
+                    "failure journal is not configured",
+                )
+            ledger = await self.coordinator.task_store.load(binding.root_trace_id)
+            events = await self.failure_repository.list_for_build(script_build_id)
+            if not events:
+                return ResumeResult(
+                    script_build_id,
+                    RecoveryClassification.MANUAL_RECONCILIATION,
+                    status,
+                    "failed build has no typed failure journal",
+                )
+            command = self.recovery_planner.plan(events, ledger)
+            classification = {
+                RecoveryCommandKind.REVALIDATE_ATTEMPT:
+                    RecoveryClassification.RESUME_VALIDATION,
+                RecoveryCommandKind.REPLAN_MISSION:
+                    RecoveryClassification.REPLAN_REQUIRED,
+                RecoveryCommandKind.MANUAL_RECONCILIATION:
+                    RecoveryClassification.MANUAL_RECONCILIATION,
+            }[command.kind]
+            return ResumeResult(
+                script_build_id,
+                classification,
+                status,
+                f"{command.kind.value}: {command.reason}",
+            )
         if status is BuildStatus.SUCCESS:
             if publication is None or publication.state is not PublicationState.PUBLISHED:
                 raise PublicationStateInconsistent()

script_build_host/src/script_build_host/application/recovery_planner.py

@@ -0,0 +1,118 @@
+"""Pure recovery planning from the failure journal and current Ledger."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import StrEnum
+from typing import Any
+
+from agent import FailureDisposition
+
+from script_build_host.application.failure_policy import FAILURE_POLICY_CATALOG
+from script_build_host.application.mission_failure_resolver import (
+    MissionFailureResolver,
+)
+from script_build_host.domain.failure_journal import MissionFailureEvent
+
+
+class RecoveryCommandKind(StrEnum):
+    REVALIDATE_ATTEMPT = "revalidate_attempt"
+    REPLAN_MISSION = "replan_mission"
+    MANUAL_RECONCILIATION = "manual_reconciliation"
+
+
+@dataclass(frozen=True, slots=True)
+class MissionRecoveryCommand:
+    kind: RecoveryCommandKind
+    failure_event_key: str
+    reason: str
+    task_id: str | None = None
+    attempt_id: str | None = None
+
+
+class MissionRecoveryPlanner:
+    """Select one explicit command; never mutates or unblocks Task state."""
+
+    def plan(
+        self,
+        events: tuple[MissionFailureEvent, ...],
+        ledger: Any,
+    ) -> MissionRecoveryCommand:
+        primary = MissionFailureResolver().resolve(events).primary_cause
+        disposition = FAILURE_POLICY_CATALOG.dispositions.get(primary.code)
+        subject = primary.subject
+        task = ledger.tasks.get(subject.task_id or "")
+        attempt = ledger.attempts.get(subject.attempt_id or "")
+        prior_validations = (
+            [
+                ledger.validations[validation_id]
+                for validation_id in task.validation_ids
+                if validation_id in ledger.validations
+                and ledger.validations[validation_id].attempt_id
+                == getattr(attempt, "attempt_id", None)
+            ]
+            if task is not None and attempt is not None
+            else []
+        )
+        revalidation_ready = bool(
+            prior_validations
+            and getattr(attempt, "snapshot_id", None)
+            and getattr(prior_validations[-1].status, "value", prior_validations[-1].status)
+            in {"error", "stopped", "expired"}
+        )
+        if (
+            disposition
+            in {
+                FailureDisposition.RETRY_CALL,
+                FailureDisposition.REPAIR_ATTEMPT,
+            }
+            and task is not None
+            and attempt is not None
+            and attempt.task_id == task.task_id
+            and attempt.spec_version == task.current_spec_version
+            and getattr(task.status, "value", task.status) == "needs_replan"
+            and revalidation_ready
+        ):
+            return MissionRecoveryCommand(
+                RecoveryCommandKind.REVALIDATE_ATTEMPT,
+                primary.event_key,
+                primary.code,
+                task.task_id,
+                attempt.attempt_id,
+            )
+        active = any(
+            getattr(item.status, "value", item.status)
+            in {"pending", "running", "stop_requested"}
+            for item in ledger.operations.values()
+        )
+        if (
+            disposition is FailureDisposition.REPLAN_TASK
+            and not active
+            and getattr(
+                ledger.tasks[ledger.root_task_id].status,
+                "value",
+                ledger.tasks[ledger.root_task_id].status,
+            )
+            not in {"completed", "cancelled", "superseded"}
+        ):
+            return MissionRecoveryCommand(
+                RecoveryCommandKind.REPLAN_MISSION,
+                primary.event_key,
+                primary.code,
+                subject.task_id,
+                subject.attempt_id,
+            )
+        return MissionRecoveryCommand(
+            RecoveryCommandKind.MANUAL_RECONCILIATION,
+            primary.event_key,
+            primary.code,
+            subject.task_id,
+            subject.attempt_id,
+        )
+
+
+__all__ = [
+    "MissionRecoveryCommand",
+    "MissionRecoveryPlanner",
+    "RecoveryCommandKind",
+]

script_build_host/src/script_build_host/domain/errors.py

@@ -119,6 +119,16 @@ class ProtocolViolation(ScriptBuildError):
         super().__init__("PROTOCOL_VIOLATION", summary)
 
 
+class RequiredEvidenceUnsatisfied(ScriptBuildError):
+    def __init__(self, summary: str) -> None:
+        super().__init__("REQUIRED_EVIDENCE_UNSATISFIED", summary)
+
+
+class ValidationEvidenceUnauthorized(ScriptBuildError):
+    def __init__(self, summary: str) -> None:
+        super().__init__("VALIDATION_EVIDENCE_UNAUTHORIZED", summary)
+
+
 class UpstreamFailure(ScriptBuildError):
     """Typed external dependency failure safe for retry and operator policy."""
 

script_build_host/src/script_build_host/domain/failure_journal.py

@@ -0,0 +1,177 @@
+"""Typed append-only mission failure events."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+from enum import StrEnum
+from typing import Any
+
+from .canonical_json import canonical_sha256
+from .errors import ProtocolViolation
+
+
+class FailureEventType(StrEnum):
+    OBSERVED = "failure_observed"
+    RECOVERY = "recovery_action"
+    STAGE_TERMINAL = "stage_terminal"
+
+
+class RecoveryAction(StrEnum):
+    RETRY_SCHEDULED = "retry_scheduled"
+    REPLAN_SCHEDULED = "replan_scheduled"
+    ATTEMPT_SUCCEEDED = "attempt_succeeded"
+    REVALIDATION_PASSED = "revalidation_passed"
+    CANCELLED = "cancelled"
+    SUPERSEDED = "superseded"
+
+    @property
+    def resolves(self) -> bool:
+        return self in {
+            RecoveryAction.ATTEMPT_SUCCEEDED,
+            RecoveryAction.REVALIDATION_PASSED,
+            RecoveryAction.CANCELLED,
+            RecoveryAction.SUPERSEDED,
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class FailureSubject:
+    task_id: str | None = None
+    attempt_id: str | None = None
+    validation_id: str | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class MissionFailureEvent:
+    event_key: str
+    script_build_id: int
+    root_trace_id: str
+    event_type: FailureEventType
+    code: str
+    summary: str
+    subject: FailureSubject = field(default_factory=FailureSubject)
+    depth: int = 0
+    cause_event_key: str | None = None
+    recovery_of_event_key: str | None = None
+    successor_attempt_id: str | None = None
+    successor_validation_id: str | None = None
+    occurred_at: datetime = field(default_factory=lambda: datetime.now(UTC))
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        if self.script_build_id < 1 or not self.event_key or not self.code:
+            raise ProtocolViolation("failure event identity is invalid")
+        if self.occurred_at.tzinfo is None:
+            raise ProtocolViolation("failure event time must be timezone-aware")
+        if self.event_type is FailureEventType.RECOVERY:
+            if not self.recovery_of_event_key:
+                raise ProtocolViolation("recovery must reference one observed failure")
+            RecoveryAction(self.code)
+        elif self.recovery_of_event_key is not None:
+            raise ProtocolViolation("only recovery events may resolve failures")
+        if self.event_type is not FailureEventType.OBSERVED and self.cause_event_key:
+            raise ProtocolViolation("only observed failures may wrap another cause")
+
+    @classmethod
+    def observed(
+        cls,
+        *,
+        script_build_id: int,
+        root_trace_id: str,
+        code: str,
+        summary: str,
+        subject: FailureSubject,
+        depth: int,
+        cause_event_key: str | None = None,
+        occurred_at: datetime | None = None,
+    ) -> MissionFailureEvent:
+        identity = {
+            "build": script_build_id,
+            "root": root_trace_id,
+            "type": FailureEventType.OBSERVED.value,
+            "code": code,
+            "subject": subject,
+            "cause": cause_event_key,
+        }
+        return cls(
+            canonical_sha256(identity).wire,
+            script_build_id,
+            root_trace_id,
+            FailureEventType.OBSERVED,
+            code,
+            summary,
+            subject,
+            depth,
+            cause_event_key,
+            occurred_at=occurred_at or datetime.now(UTC),
+        )
+
+    @classmethod
+    def recovery(
+        cls,
+        *,
+        script_build_id: int,
+        root_trace_id: str,
+        action: RecoveryAction,
+        recovery_of_event_key: str,
+        subject: FailureSubject,
+        successor_attempt_id: str | None = None,
+        successor_validation_id: str | None = None,
+        occurred_at: datetime | None = None,
+    ) -> MissionFailureEvent:
+        identity = {
+            "build": script_build_id,
+            "root": root_trace_id,
+            "type": FailureEventType.RECOVERY.value,
+            "action": action.value,
+            "recovery_of": recovery_of_event_key,
+            "successor_attempt": successor_attempt_id,
+            "successor_validation": successor_validation_id,
+        }
+        return cls(
+            event_key=canonical_sha256(identity).wire,
+            script_build_id=script_build_id,
+            root_trace_id=root_trace_id,
+            event_type=FailureEventType.RECOVERY,
+            code=action.value,
+            summary=action.value,
+            subject=subject,
+            recovery_of_event_key=recovery_of_event_key,
+            successor_attempt_id=successor_attempt_id,
+            successor_validation_id=successor_validation_id,
+            occurred_at=occurred_at or datetime.now(UTC),
+        )
+
+    @classmethod
+    def stage_terminal(
+        cls,
+        *,
+        script_build_id: int,
+        root_trace_id: str,
+        code: str,
+        occurred_at: datetime | None = None,
+    ) -> MissionFailureEvent:
+        identity = {
+            "build": script_build_id,
+            "root": root_trace_id,
+            "type": FailureEventType.STAGE_TERMINAL.value,
+            "code": code,
+        }
+        return cls(
+            event_key=canonical_sha256(identity).wire,
+            script_build_id=script_build_id,
+            root_trace_id=root_trace_id,
+            event_type=FailureEventType.STAGE_TERMINAL,
+            code=code,
+            summary=code,
+            occurred_at=occurred_at or datetime.now(UTC),
+        )
+
+
+__all__ = [
+    "FailureEventType",
+    "FailureSubject",
+    "MissionFailureEvent",
+    "RecoveryAction",
+]

script_build_host/src/script_build_host/domain/failure_ports.py

@@ -0,0 +1,26 @@
+"""Persistence port for append-only mission failure events."""
+
+from __future__ import annotations
+
+from typing import Protocol
+
+from .failure_journal import MissionFailureEvent
+
+
+class MissionFailureRepository(Protocol):
+    async def append(self, events: tuple[MissionFailureEvent, ...]) -> None: ...
+
+    async def list_for_build(
+        self, script_build_id: int
+    ) -> tuple[MissionFailureEvent, ...]: ...
+
+    async def terminate(
+        self,
+        *,
+        script_build_id: int,
+        events: tuple[MissionFailureEvent, ...],
+        primary_code: str,
+    ) -> None: ...
+
+
+__all__ = ["MissionFailureRepository"]

script_build_host/src/script_build_host/infrastructure/tables.py

@@ -182,3 +182,31 @@ http_command_table = Table(
     CheckConstraint("state IN ('reserved', 'completed', 'failed')", name="ck_http_command_state"),
     Index("ix_http_command_resource", "resource_id", "route_family"),
 )
+
+mission_failure_event_table = Table(
+    "script_build_mission_failure_event",
+    metadata,
+    Column("id", _primary_key_type, primary_key=True, autoincrement=True),
+    Column("event_key", String(80), nullable=False),
+    Column("script_build_id", BigInteger, nullable=False),
+    Column("root_trace_id", String(128), nullable=False),
+    Column("event_type", String(32), nullable=False),
+    Column("code", String(128), nullable=False),
+    Column("summary", String(1000), nullable=False),
+    Column("task_id", String(128)),
+    Column("attempt_id", String(128)),
+    Column("validation_id", String(128)),
+    Column("depth", Integer, nullable=False, default=0, server_default="0"),
+    Column("cause_event_key", String(80)),
+    Column("recovery_of_event_key", String(80)),
+    Column("successor_attempt_id", String(128)),
+    Column("successor_validation_id", String(128)),
+    Column("metadata_json", JSON, nullable=False),
+    Column("occurred_at", _timestamp_type, nullable=False),
+    UniqueConstraint("event_key", name="uq_mission_failure_event_key"),
+    CheckConstraint(
+        "event_type IN ('failure_observed', 'recovery_action', 'stage_terminal')",
+        name="ck_mission_failure_event_type",
+    ),
+    Index("ix_mission_failure_build", "script_build_id", "occurred_at"),
+)

script_build_host/src/script_build_host/repositories/__init__.py

@@ -1,5 +1,6 @@
 from .legacy_input import LegacySqlAlchemyInputReader
 from .legacy_state import SqlAlchemyLegacyBuildStateRepository
+from .mission_failure import SqlAlchemyMissionFailureRepository
 from .sqlalchemy import (
     SqlAlchemyInputSnapshotRepository,
     SqlAlchemyMissionBindingRepository,
@@ -14,6 +15,7 @@ __all__ = [
     "SqlAlchemyInputSnapshotRepository",
     "SqlAlchemyLegacyBuildStateRepository",
     "SqlAlchemyMissionBindingRepository",
+    "SqlAlchemyMissionFailureRepository",
     "SqlAlchemyPublicationRepository",
     "SqlAlchemyScriptBusinessArtifactRepository",
 ]

script_build_host/src/script_build_host/repositories/mission_failure.py

@@ -0,0 +1,164 @@
+"""SQLAlchemy implementation of the append-only mission failure journal."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any, cast
+
+from sqlalchemy import insert, select, update
+from sqlalchemy.engine import CursorResult
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from script_build_host.domain.errors import BuildNotFound
+from script_build_host.domain.failure_journal import (
+    FailureEventType,
+    FailureSubject,
+    MissionFailureEvent,
+)
+from script_build_host.infrastructure.legacy_tables import (
+    script_build_record,
+    script_build_runtime_record,
+)
+from script_build_host.infrastructure.redaction import redact, redact_text
+from script_build_host.infrastructure.tables import mission_failure_event_table
+
+
+class SqlAlchemyMissionFailureRepository:
+    def __init__(self, sessions: async_sessionmaker[AsyncSession]) -> None:
+        self._sessions = sessions
+        bind = sessions.kw.get("bind")
+        self._build_table = (
+            script_build_runtime_record
+            if getattr(getattr(bind, "dialect", None), "name", None) == "mysql"
+            else script_build_record
+        )
+
+    async def append(self, events: tuple[MissionFailureEvent, ...]) -> None:
+        async with self._sessions() as session, session.begin():
+            await self._append(session, events)
+
+    async def list_for_build(
+        self, script_build_id: int
+    ) -> tuple[MissionFailureEvent, ...]:
+        async with self._sessions() as session:
+            rows = (
+                (
+                    await session.execute(
+                        select(mission_failure_event_table)
+                        .where(
+                            mission_failure_event_table.c.script_build_id
+                            == script_build_id
+                        )
+                        .order_by(
+                            mission_failure_event_table.c.occurred_at,
+                            mission_failure_event_table.c.id,
+                        )
+                    )
+                )
+                .mappings()
+                .all()
+            )
+        return tuple(_from_row(row) for row in rows)
+
+    async def terminate(
+        self,
+        *,
+        script_build_id: int,
+        events: tuple[MissionFailureEvent, ...],
+        primary_code: str,
+    ) -> None:
+        """Append the journal and project the primary cause in one transaction."""
+
+        async with self._sessions() as session, session.begin():
+            await self._append(session, events)
+            result = cast(
+                CursorResult[Any],
+                await session.execute(
+                    update(self._build_table)
+                    .where(
+                        self._build_table.c.id == script_build_id,
+                        self._build_table.c.is_deleted.is_(False),
+                    )
+                    .values(
+                        status="failed",
+                        error_message=redact_text(primary_code)[:255],
+                        end_time=max(item.occurred_at for item in events),
+                    )
+                )
+            )
+            if result.rowcount != 1:
+                raise BuildNotFound()
+
+    async def _append(
+        self,
+        session: AsyncSession,
+        events: tuple[MissionFailureEvent, ...],
+    ) -> None:
+        if not events:
+            return
+        keys = tuple(item.event_key for item in events)
+        existing = set(
+            (
+                await session.scalars(
+                    select(mission_failure_event_table.c.event_key).where(
+                        mission_failure_event_table.c.event_key.in_(keys)
+                    )
+                )
+            ).all()
+        )
+        for item in events:
+            if item.event_key in existing:
+                continue
+            await session.execute(
+                insert(mission_failure_event_table).values(
+                    event_key=item.event_key,
+                    script_build_id=item.script_build_id,
+                    root_trace_id=item.root_trace_id,
+                    event_type=item.event_type.value,
+                    code=item.code,
+                    summary=redact_text(item.summary)[:1000],
+                    task_id=item.subject.task_id,
+                    attempt_id=item.subject.attempt_id,
+                    validation_id=item.subject.validation_id,
+                    depth=item.depth,
+                    cause_event_key=item.cause_event_key,
+                    recovery_of_event_key=item.recovery_of_event_key,
+                    successor_attempt_id=item.successor_attempt_id,
+                    successor_validation_id=item.successor_validation_id,
+                    metadata_json=redact(item.metadata) or {},
+                    occurred_at=item.occurred_at,
+                )
+            )
+
+
+def _from_row(row: Any) -> MissionFailureEvent:
+    occurred_at = (
+        row["occurred_at"]
+        if isinstance(row["occurred_at"], datetime)
+        else datetime.fromisoformat(str(row["occurred_at"]))
+    )
+    if occurred_at.tzinfo is None:
+        occurred_at = occurred_at.replace(tzinfo=UTC)
+    return MissionFailureEvent(
+        event_key=str(row["event_key"]),
+        script_build_id=int(row["script_build_id"]),
+        root_trace_id=str(row["root_trace_id"]),
+        event_type=FailureEventType(str(row["event_type"])),
+        code=str(row["code"]),
+        summary=str(row["summary"]),
+        subject=FailureSubject(
+            task_id=row["task_id"],
+            attempt_id=row["attempt_id"],
+            validation_id=row["validation_id"],
+        ),
+        depth=int(row["depth"]),
+        cause_event_key=row["cause_event_key"],
+        recovery_of_event_key=row["recovery_of_event_key"],
+        successor_attempt_id=row["successor_attempt_id"],
+        successor_validation_id=row["successor_validation_id"],
+        occurred_at=occurred_at,
+        metadata=dict(row["metadata_json"] or {}),
+    )
+
+
+__all__ = ["SqlAlchemyMissionFailureRepository"]

script_build_host/tests/test_architecture_failure_journal.py

@@ -0,0 +1,169 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from types import SimpleNamespace
+
+import pytest
+
+from script_build_host.application.failure_policy import classify_script_failure
+from script_build_host.application.mission_failure_flow import MissionFailureFlow
+from script_build_host.application.mission_failure_resolver import MissionFailureResolver
+from script_build_host.application.recovery_planner import (
+    MissionRecoveryPlanner,
+    RecoveryCommandKind,
+)
+from script_build_host.domain.errors import ProtocolViolation, ScriptBuildError
+from script_build_host.domain.failure_journal import (
+    FailureEventType,
+    FailureSubject,
+    MissionFailureEvent,
+    RecoveryAction,
+)
+
+
+def test_outer_protocol_wrapper_does_not_replace_context_root_cause() -> None:
+    inner = MissionFailureEvent.observed(
+        script_build_id=1,
+        root_trace_id="root",
+        code="CONTEXT_BUDGET_EXCEEDED",
+        summary="validator prompt is too large",
+        subject=FailureSubject("task", "attempt", "validation"),
+        depth=3,
+    )
+    outer = MissionFailureEvent(
+        "outer",
+        1,
+        "root",
+        FailureEventType.OBSERVED,
+        "PROTOCOL_VIOLATION",
+        "phase boundary not reached",
+        depth=0,
+        cause_event_key=inner.event_key,
+        occurred_at=datetime.now(UTC),
+    )
+    resolved = MissionFailureResolver().resolve((inner, outer))
+    assert resolved.primary_cause.code == "CONTEXT_BUDGET_EXCEEDED"
+
+
+def test_failure_resolver_rejects_unknown_cause_link() -> None:
+    event = MissionFailureEvent(
+        "outer",
+        1,
+        "root",
+        FailureEventType.OBSERVED,
+        "PROTOCOL_VIOLATION",
+        "wrapper",
+        cause_event_key="missing",
+    )
+    with pytest.raises(ProtocolViolation, match="unknown"):
+        MissionFailureResolver().resolve((event,))
+
+
+def test_unknown_failure_code_fails_closed_without_guessing_prefix() -> None:
+    failure = classify_script_failure(
+        ScriptBuildError("TASK_MADE_UP", "unknown"),
+        source_tool="tool",
+    )
+    assert failure.code == "INTERNAL_UNCLASSIFIED_FAILURE"
+    assert failure.details["observed_code"] == "TASK_MADE_UP"
+
+
+def test_revalidation_is_not_planned_without_snapshot_and_failed_validation() -> None:
+    failure = MissionFailureEvent.observed(
+        script_build_id=1,
+        root_trace_id="root",
+        code="VALIDATION_EVIDENCE_UNAUTHORIZED",
+        summary="failed before validation was frozen",
+        subject=FailureSubject("task", "attempt", None),
+        depth=2,
+    )
+    task = SimpleNamespace(
+        task_id="task",
+        current_spec_version=1,
+        status=SimpleNamespace(value="needs_replan"),
+        validation_ids=[],
+    )
+    attempt = SimpleNamespace(
+        attempt_id="attempt",
+        task_id="task",
+        spec_version=1,
+        snapshot_id=None,
+    )
+    ledger = SimpleNamespace(
+        root_task_id="root-task",
+        tasks={
+            "root-task": SimpleNamespace(status=SimpleNamespace(value="needs_replan")),
+            "task": task,
+        },
+        attempts={"attempt": attempt},
+        validations={},
+        operations={},
+    )
+    command = MissionRecoveryPlanner().plan((failure,), ledger)
+    assert command.kind is RecoveryCommandKind.MANUAL_RECONCILIATION
+
+
+@pytest.mark.asyncio
+async def test_new_passed_attempt_recovers_prior_validation_failure() -> None:
+    now = datetime.now(UTC)
+
+    class Repository:
+        def __init__(self) -> None:
+            self.events = ()
+
+        async def append(self, events):
+            self.events = events
+
+    repository = Repository()
+    attempts = {
+        "attempt-old": SimpleNamespace(
+            attempt_id="attempt-old",
+            created_at=now.isoformat(),
+            status=SimpleNamespace(value="submitted"),
+            failure=None,
+        ),
+        "attempt-new": SimpleNamespace(
+            attempt_id="attempt-new",
+            created_at=(now + timedelta(seconds=2)).isoformat(),
+            status=SimpleNamespace(value="submitted"),
+            failure=None,
+        ),
+    }
+    validations = {
+        "validation-old": SimpleNamespace(
+            validation_id="validation-old",
+            attempt_id="attempt-old",
+            created_at=(now + timedelta(seconds=1)).isoformat(),
+            status=SimpleNamespace(value="error"),
+            verdict=None,
+            failure=SimpleNamespace(code="CONTEXT_BUDGET_EXCEEDED", message="too large"),
+        ),
+        "validation-new": SimpleNamespace(
+            validation_id="validation-new",
+            attempt_id="attempt-new",
+            created_at=(now + timedelta(seconds=3)).isoformat(),
+            status=SimpleNamespace(value="completed"),
+            verdict=SimpleNamespace(value="passed"),
+            failure=None,
+        ),
+    }
+    task = SimpleNamespace(
+        task_id="task",
+        attempt_ids=list(attempts),
+        validation_ids=list(validations),
+        status=SimpleNamespace(value="completed"),
+        updated_at=(now + timedelta(seconds=4)).isoformat(),
+    )
+    ledger = SimpleNamespace(
+        tasks={"task": task},
+        attempts=attempts,
+        validations=validations,
+    )
+    events = await MissionFailureFlow(repository).sync_ledger(
+        script_build_id=1,
+        root_trace_id="root",
+        ledger=ledger,
+    )
+    recovery = next(item for item in events if item.event_type is FailureEventType.RECOVERY)
+    assert recovery.code == RecoveryAction.ATTEMPT_SUCCEEDED
+    assert recovery.successor_attempt_id == "attempt-new"

script_build_host/tests/test_migrations.py

@@ -26,6 +26,7 @@ def test_alembic_upgrade_and_downgrade(tmp_path: Path, monkeypatch: object) -> N
         "script_build_mission_binding",
         "script_build_input_snapshot",
         "script_build_artifact_version",
+        "script_build_mission_failure_event",
         "script_build_publication",
     }
     assert expected <= set(inspect(sync_engine).get_table_names())
@@ -59,7 +60,7 @@ def test_mysql_offline_sql_compiles_without_credentials(monkeypatch: object) ->
     config = Config(str(root / "alembic.ini"), output_buffer=output)
     command.upgrade(config, "head", sql=True)
     sql = output.getvalue()
-    assert sql.count("CREATE TABLE script_build_") == 6
+    assert sql.count("CREATE TABLE script_build_") == 7
     assert "CREATE TABLE script_build_candidate_source_identity" in sql
     assert "CREATE TABLE script_build_paragraph" not in sql
     assert "CREATE TABLE script_build_element" not in sql
@@ -145,7 +146,7 @@ async def test_mysql_phase_three_schema_semantics() -> None:
                 "READ-COMMITTED"
             )
             assert await connection.scalar(text("SELECT version_num FROM alembic_version")) == (
-                "0004_candidate_source_identity"
+                "0005_mission_failure_events"
             )
             precision = await connection.scalar(
                 text(

script_build_host/tests/test_mission_failure_repository.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+import pytest
+from sqlalchemy import select
+
+from script_build_host.application.mission_failure_flow import MissionFailureFlow
+from script_build_host.domain.errors import ProtocolViolation
+from script_build_host.infrastructure.legacy_tables import script_build_record
+from script_build_host.repositories.legacy_state import (
+    SqlAlchemyLegacyBuildStateRepository,
+)
+from script_build_host.repositories.mission_failure import (
+    SqlAlchemyMissionFailureRepository,
+)
+
+
+@pytest.mark.asyncio
+async def test_terminal_projection_keeps_inner_context_failure(database) -> None:
+    _, sessions = database
+    legacy = SqlAlchemyLegacyBuildStateRepository(sessions)
+    build_id = await legacy.create(
+        execution_id=1,
+        topic_build_id=2,
+        topic_id=3,
+        agent_type=None,
+        agent_config=None,
+        data_source_url=None,
+        strategies_config={},
+        root_trace_id="root",
+    )
+    repository = SqlAlchemyMissionFailureRepository(sessions)
+    primary = await MissionFailureFlow(repository).terminate(
+        script_build_id=build_id,
+        root_trace_id="root",
+        completion={
+            "blocking_failures": [
+                {
+                    "task_id": "element",
+                    "attempt_id": "attempt",
+                    "validation_id": "validation",
+                    "code": "CONTEXT_BUDGET_EXCEEDED",
+                    "message": "validator envelope exceeded",
+                }
+            ]
+        },
+        error=ProtocolViolation("phase two boundary not reached"),
+        stage_code="PHASE_TWO_BOUNDARY_NOT_REACHED",
+    )
+    assert primary == "CONTEXT_BUDGET_EXCEEDED"
+    events = await repository.list_for_build(build_id)
+    assert {item.code for item in events} >= {
+        "CONTEXT_BUDGET_EXCEEDED",
+        "PROTOCOL_VIOLATION",
+        "PHASE_TWO_BOUNDARY_NOT_REACHED",
+    }
+    async with sessions() as session:
+        row = (
+            (
+                await session.execute(
+                    select(
+                        script_build_record.c.status,
+                        script_build_record.c.error_message,
+                    ).where(script_build_record.c.id == build_id)
+                )
+            )
+            .mappings()
+            .one()
+        )
+    assert row["status"] == "failed"
+    assert row["error_message"] == "CONTEXT_BUDGET_EXCEEDED"