Resume verified runs after a known pre-dispatch failure

src/firefly_forge/checkpoint_mode_handoff_custody.py

@@ -47,7 +47,9 @@ def _one(entries: list[Data], kind: str, key: str, handoff_id: str | None = None
     return found[0]
 
 
-def _validate(run: Data, entries: list[Data], rows: list[Any], values: dict[str, Data]) -> dict[str, str]:
+def _validate(
+    run: Data, entries: list[Data], rows: list[Any], values: dict[str, Data], *, before_dispatch: bool = False
+) -> dict[str, str]:
     if not entries or any(e["run_id"] != run["id"] for e in entries) or not Journal.verify_chain(entries):
         raise ValueError("Completed verification requires its intact native journal")
     state, request, response = (
@@ -102,7 +104,6 @@ def _validate(run: Data, entries: list[Data], rows: list[Any], values: dict[str,
         raise ValueError("Completed verification differs from its frozen request, commands or scope")
     answer = _one(entries, "worker_response", "worker-response:" + handoff_id, handoff_id)
     outcome = _one(entries, "verification_handoff_outcome", handoff_id + ":outcome", handoff_id)
-    coding = _one(entries, "verification_continuation_intent", handoff_id + ":coding", handoff_id)
     resumed = [
         e
         for e in entries
@@ -112,7 +113,17 @@ def _validate(run: Data, entries: list[Data], rows: list[Any], values: dict[str,
         raise ValueError("Completed verification requires its unique native resume")
     resume = resumed[0]
     task_key = uuid5(NAMESPACE_URL, f"forge:{run['id']}:implement:{intent['parent_round'] + 1}").hex
-    implementation = _one(entries, "intent", task_key)
+    coding = implementation = None
+    if before_dispatch:
+        if entries[-1]["digest"] != resume["digest"] or any(
+            (e["kind"] == "intent" and e["idempotency_key"] == task_key)
+            or (e["kind"] == "verification_continuation_intent" and e["payload"].get("id") == handoff_id)
+            for e in entries
+        ):
+            raise ValueError("Pre-dispatch verification cannot contain later execution or admission")
+    else:
+        coding = _one(entries, "verification_continuation_intent", handoff_id + ":coding", handoff_id)
+        implementation = _one(entries, "intent", task_key)
     responding_actor = {
         "version": 4,
         "authentication_method": "factory_verification",
@@ -141,26 +152,27 @@ def _validate(run: Data, entries: list[Data], rows: list[Any], values: dict[str,
         or resume["payload"].get("agent") != run["agent"]
         or resume["task_id"] != resume["kind"]
         or resume["payload"].get("config_revision") != run["config_revision"]
-        or implementation["payload"].get("config_revision") != run["config_revision"]
-        or digest(coding["payload"])
-        != digest(
-            {
-                "version": 1,
-                "id": handoff_id,
-                "request_id": intent["request_id"],
-                "parent_round": intent["parent_round"],
-                "next_round": intent["parent_round"] + 1,
-                "provider_digest": intent["provider_digest"],
-            }
-        )
-        or not handoff["sequence"]
-        < answer["sequence"]
-        < outcome["sequence"]
-        < resume["sequence"]
-        < implementation["sequence"]
-        < coding["sequence"]
+        or not handoff["sequence"] < answer["sequence"] < outcome["sequence"] < resume["sequence"]
     ):
         raise ValueError("Completed verification lacks its exact response and native continuation grant")
+    if not before_dispatch:
+        assert implementation is not None and coding is not None
+        if (
+            implementation["payload"].get("config_revision") != run["config_revision"]
+            or digest(coding["payload"])
+            != digest(
+                {
+                    "version": 1,
+                    "id": handoff_id,
+                    "request_id": intent["request_id"],
+                    "parent_round": intent["parent_round"],
+                    "next_round": intent["parent_round"] + 1,
+                    "provider_digest": intent["provider_digest"],
+                }
+            )
+            or not resume["sequence"] < implementation["sequence"] < coding["sequence"]
+        ):
+            raise ValueError("Completed verification lacks its exact response and native continuation grant")
     command_records = [
         e for e in entries if e["kind"] == "verification_handoff_command" and e["payload"].get("id") == handoff_id
     ]
@@ -273,18 +285,21 @@ def _validate(run: Data, entries: list[Data], rows: list[Any], values: dict[str,
         "response_digest": answer["digest"],
         "outcome_digest": outcome["digest"],
         "resume_digest": resume["digest"],
-        "implementation_intent_digest": implementation["digest"],
-        "continuation_intent_digest": coding["digest"],
+        **(
+            {"implementation_intent_digest": implementation["digest"], "continuation_intent_digest": coding["digest"]}
+            if implementation is not None and coding is not None
+            else {}
+        ),
         "command_custody_digest": digest({"commands": refs}),
     }
     return {**result, "digest": digest(result)}
 
 
 def validate_completed_handoff(
-    run: Data, entries: list[Data], rows: list[Any], values: dict[str, Data]
+    run: Data, entries: list[Data], rows: list[Any], values: dict[str, Data], *, before_dispatch: bool = False
 ) -> dict[str, str]:
     try:
-        return _validate(run, entries, rows, values)
+        return _validate(run, entries, rows, values, before_dispatch=before_dispatch)
     except (
         ValueError,
         TypeError,

src/firefly_forge/journal.py

@@ -30,6 +30,7 @@
 from firefly_forge.store import TERMINAL, Store, identifier, now, prepared_credential_rejection
 
 if TYPE_CHECKING:
+    from firefly_forge.retained_verified_source import PreparedVerifiedResume
     from firefly_forge.scm_source_edit_models import SealedSourceEditAdmission
 
 
@@ -2203,6 +2204,22 @@ async def resume(
         from firefly_forge.checkpoint_mode_runtime import mode_execution_scope
 
         async def selected() -> Data:
+            verified_resume = None
+            if current.get("status") == "failed":
+                if (
+                    auto
+                    or agent is not None
+                    or model_profile_id is not None
+                    or campaign_action_id is not None
+                    or verification_handoff_id is not None
+                    or expected_provider_digest is not None
+                    or source_human_readmission is not None
+                    or source_budget_readmission is not None
+                ):
+                    raise ValueError("Failed verification admission requires an explicit unchanged-provider resume")
+                from firefly_forge.retained_verified_source import prepare_verified_resume
+
+                verified_resume = await prepare_verified_resume(self.store, self, current)
             return await self._resume_selected(
                 current,
                 run_id,
@@ -2214,6 +2231,7 @@ async def selected() -> Data:
                 expected_provider_digest=expected_provider_digest,
                 source_human_readmission=source_human_readmission,
                 source_budget_readmission=source_budget_readmission,
+                verified_resume=verified_resume,
             )
 
         if current.get("checkpoint_mode_adoption"):
@@ -2234,6 +2252,7 @@ async def _resume_selected(
         expected_provider_digest: str | None = None,
         source_human_readmission: str | None = None,
         source_budget_readmission: str | None = None,
+        verified_resume: PreparedVerifiedResume | None = None,
     ) -> Data:
         from firefly_forge.project_access import ProjectAccess
 
@@ -2370,7 +2389,11 @@ async def _resume_selected(
                     {**copy.deepcopy(row.data), "digest": row.digest} for row in await self.records.find_for_run(run.id)
                 ]
                 await assert_mode_actors(self, run.data, mode_entries)
-            if run.status not in {"paused", "needs_reconciliation", "interrupted"}:
+            if verified_resume is not None:
+                if run.status != "failed":
+                    raise ValueError("Failed verification admission was already resumed or changed")
+                await verified_resume.admit()
+            elif run.status not in {"paused", "needs_reconciliation", "interrupted"}:
                 raise ValueError("Only a paused or interrupted run can resume")
             if run.data.get("worker_request", {}).get("status") == "pending":
                 raise ValueError("Answer the pending worker question before explicitly resuming this run")
@@ -2561,10 +2584,13 @@ async def _resume_selected(
                     **({"checkpoint_mode_adoption_digest": inherited_mode["record_digest"]} if inherited_mode else {}),
                     "automatic": auto or verification_handoff_id is not None,
                     **({"verification_handoff_id": verification_handoff_id} if verification_handoff_id else {}),
+                    **({"verified_pre_dispatch_resume": verified_resume.proof} if verified_resume else {}),
                     **({"human_clarification": human_clarification} if human_clarification else {}),
                     **({"source_budget_continuation": source_budget_link} if source_budget_link else {}),
                 },
             )
+            if verified_resume is not None:
+                await verified_resume.source_current()
             result = await self.store.get_run(run_id)
             if verification_handoff_id is not None:
                 assert admitted is not None

src/firefly_forge/retained_verified_source.py

@@ -203,3 +203,169 @@ def selected(key: str, kind: str) -> Data:
     ):
         raise ValueError("Retained source receipt changed its accepted blocker or verified continuation")
     validate_native_receipt(previous, entries)
+
+
+@dataclass(frozen=True)
+class PreparedVerifiedResume:
+    proof: Data
+    _admit: Callable[[], Awaitable[None]]
+    _source_current: Callable[[], Awaitable[None]]
+
+    async def admit(self) -> None:
+        await self._admit()
+
+    async def source_current(self) -> None:
+        await self._source_current()
+
+
+async def prepare_verified_resume(store: Any, journal: Any, run: Data) -> PreparedVerifiedResume:
+    """Requeue only the untouched native admission after passed QA, never a dispatched attempt."""
+    from pyfly.data.relational.sqlalchemy import LockMode
+
+    from firefly_forge import invalid_read_continuities
+    from firefly_forge.completed_final_epoch import validated_completed_final
+    from firefly_forge.invalid_read_lease import policy_snapshot
+    from firefly_forge.verification_dependencies import prepare_public_source_cut
+
+    _outside_unit()
+    completed = await validated_completed_final(journal, run)
+    if completed is None:
+        raise ValueError("Failed resume requires its completed native final action")
+    p = completed.record["payload"]
+    profile = invalid_read_continuities.current_profile_seal()
+    state = run.get("verification_handoff") or {}
+    if (
+        run.get("status") != "failed"
+        or p["candidate"]["outcome"] != "blocked"
+        or p["candidate"]["followup_kind"] != "verification"
+        or state.get("status") != "completed"
+        or type(state.get("failed_commands")) is not int
+        or state["failed_commands"] != 0
+        or type(run.get("attempt")) is not int
+        or run["attempt"] != p["candidate"]["attempt"]
+        or type(run.get("implementation_rounds")) is not int
+        or run["implementation_rounds"] != run["attempt"]
+        or run.get("implementation_pending") is not False
+        or not _same(run.get("usage"), p["run"].get("usage"))
+        or run.get("publication_status") == "publishing"
+    ):
+        raise ValueError("Failed resume requires unchanged spent progress after passed native verification")
+    actor = CURRENT_PRINCIPAL.get()
+    if actor is None or actor.authentication_method not in {"local_bootstrap", "oidc_session", "oidc_bearer"}:
+        raise AccessDenied("An authenticated native operator must explicitly resume this failed admission")
+    await store.assert_ownership()
+    ownership = store.execution_ownership_token
+    expected = copy.deepcopy(run)
+
+    async def cut() -> tuple[Data, Data, list[Data], Data, Data, list[Data]]:
+        requests = store.execution_requests
+        native, task, entries = await requests.invalid_read_continuities.recoveries._native(run["id"])
+        rows = await journal.sandboxes.for_run(identifier(run["id"]))
+        values = {row.id.hex: _canonical(row, entries, run["id"]) for row in rows}
+        custody = validate_completed_handoff(native, entries, rows, values, before_dispatch=True)
+        intent = next(e["payload"] for e in entries if e["digest"] == custody["handoff_intent_digest"])
+        if (
+            task["status"] != "queued"
+            or type(task["attempt"]) is not int
+            or task["attempt"] != native["attempt"] + 1
+            or task.get("idempotency_key") != p["task"]["idempotency_key"]
+            or intent["parent_attempt"] != native["attempt"]
+            or intent["parent_round"] != native["implementation_rounds"]
+            or completed.terminal["sequence"]
+            >= next(e["sequence"] for e in entries if e["digest"] == custody["handoff_intent_digest"])
+            or any(row.parent_attempt > native["attempt"] for row in rows)
+            or any(row.parent_attempt == native["attempt"] and row.ordinal > intent["ordinals"][-1] for row in rows)
+        ):
+            raise ValueError("Failed resume lost its untouched queued task or known command boundary")
+        request_row = await requests.records.find_by_id(
+            identifier(completed.ticket["id"]), lock=LockMode.PESSIMISTIC_WRITE
+        )
+        if request_row is None or request_row.status != "completed" or request_row.run_id.hex != run["id"]:
+            raise ValueError("Failed resume completed request columns changed")
+        ticket = await requests.get(completed.ticket["id"])
+        events = await read_recovery_events(requests, ticket["id"], maximum=LIMITS["journal"])
+        if not _same(ticket, completed.ticket):
+            raise ValueError("Failed resume completed ticket changed")
+        from firefly_forge.git_deliveries import GitDeliveries
+
+        GitDeliveries._remote_state(entries)
+        confirmed = {r["request_id"] for r in GitDeliveries._records(entries)}
+        if any(e["kind"] == "git_delivery_intent" and e["payload"]["request_id"] not in confirmed for e in entries):
+            raise ValueError("Failed resume cannot cross an unresolved Git effect")
+        if any(t.status == "running" for t in await journal.tasks.find_for_run(identifier(run["id"]))):
+            raise ValueError("Failed resume cannot replace an active task")
+        return native, task, entries, values, ticket, events
+
+    async with store._unit():
+        expected_cut = copy.deepcopy(await cut())
+    if not _same(expected_cut[0], expected):
+        raise ValueError("Failed resume changed before its native cut")
+    entries = expected_cut[2]
+    handoff = next(
+        e for e in entries if e["kind"] == "verification_handoff_intent" and e["payload"].get("id") == state["id"]
+    )
+    intent = handoff["payload"]
+    workspace = Path(intent["workspace"])
+
+    async def actor_current() -> None:
+        await store.assert_ownership()
+        if CURRENT_PRINCIPAL.get() != actor or store.execution_ownership_token is not ownership:
+            raise AccessDenied("Failed resume operator or native execution owner changed")
+        if not _same(invalid_read_continuities.current_profile_seal(), profile) or not _same(
+            policy_snapshot(store.configuration), p["lease_policy"]
+        ):
+            raise ValueError("Failed resume current qualified profile or continuity policy changed")
+        await ProjectAccess(store).authorize_execution(expected)
+        await ProjectAccess(store).authorize_actor(
+            ProjectAccess.actor_snapshot(actor), expected.get("project_id"), "execute"
+        )
+        await store.assert_campaign_current(expected)
+        check_verification_authority(store, expected, intent)
+
+    async def authority() -> None:
+        await actor_current()
+        async with store._unit():
+            if not _same(await cut(), expected_cut):
+                raise ValueError("Failed resume native run, task or completed command custody changed")
+
+    async def current() -> None:
+        await completed.current()
+        await authority()
+
+    await current()
+    catalog = verification_catalog_root(store.path, run["id"])
+    metadata = await prepare_public_source_cut(workspace, catalog, journal.source_catalog_budget)
+    source = await prepare_verification_source(
+        workspace, catalog, budget=journal.source_catalog_budget, assert_current=current
+    )
+    if source.source_digest != intent["source_digest"]:
+        raise ValueError("Failed resume source changed after passed native verification")
+    await source.revalidate(assert_current=current)
+
+    async def source_current() -> None:
+        source.fence()
+        await metadata.assert_current()
+        await actor_current()
+        async with store._unit():
+            if not _same(await store.execution_requests.get(completed.ticket["id"]), expected_cut[4]) or not _same(
+                await read_recovery_events(store.execution_requests, completed.ticket["id"], maximum=LIMITS["journal"]),
+                expected_cut[5],
+            ):
+                raise ValueError("Failed resume completed native action changed before commit")
+
+    async def admit() -> None:
+        await authority()
+        await source_current()
+
+    await admit()
+    proof = {
+        "version": 1,
+        "final_outcome_digest": completed.record["digest"],
+        "terminal_receipt_digest": completed.terminal["digest"],
+        "verification_handoff_digest": handoff["digest"],
+        "pre_dispatch_resume_digest": entries[-1]["digest"],
+        "failed_run_digest": digest(expected),
+        "task_digest": digest(expected_cut[1]),
+        "source_digest": source.source_digest,
+    }
+    return PreparedVerifiedResume(proof, admit, source_current)