Fix EP retry attempt readback binding

.github/product-version-operations/forge-version-2.7.21-retry-attempt-digest-20260917.json

@@ -0,0 +1,16 @@
+{
+  "baseline_version": "2.7.20",
+  "classification_rationale": "codex/retry-readback-attempt-digest",
+  "component": "product",
+  "event_lineage": "codex/retry-readback-attempt-digest",
+  "expected_head": "83cf8dd6adc64b19575414aba7ac63d489f3b032",
+  "operation_id": "forge-version-2.7.21-retry-attempt-digest-20260917",
+  "policy_revision": "forge-bootstrap-release-cadence-v2",
+  "product": "forge",
+  "projection_paths": "product-version.json",
+  "release_class": "PATCH",
+  "requested_bump": "patch",
+  "requested_version": null,
+  "schema_version": "1",
+  "target_version": "2.7.21"
+}

docs/architecture/FORGE_EP_PRODUCER_CONTRACT_V1_2_MIGRATION.md

@@ -88,6 +88,19 @@ host-proven retry resolution evidence, never as a second Forge submission or a
 Forge-owned retry; Forge also appends a redacted operational audit event.
 Every missing, cyclic or mismatched lineage field fails closed.
 
+Each EP-owned retry is a distinct accepted attempt and therefore retains its
+own canonical accepted-request digest. Forge continues to recompute and require
+the original digest for the original submission. For an explicit, exact-parent
+retry successor it instead requires a canonical attempt digest, unchanged
+Forge correlation/provenance/producer/scope identity, and exact digest parity
+between producer readback and the immutable terminal artifact. A v1.4
+successor may carry an EP-authorized `ALLOWED` baseline transition only on that
+host-proven retry path: the original requested revision remains unchanged and
+`allowed_to`, transition target and execution baseline must be the same exact
+SHA. Forge records the terminal successor's attempt digest with the retry
+resolution audit binding; it does not reinterpret that digest as a new Forge
+submission receipt or rewrite the original request.
+
 Pinned EP producer source: `f7c08872a2d334cff097ea5f28822836e59f78c3`.
 The migration is source compatibility only; it does not assert that every
 installed EP instance has the declaration or runs this contract. A real Mission

docs/architecture/execution-host-contract.md

@@ -87,6 +87,16 @@ reject evidence that does not match the exact dispatch. This makes stale,
 unrelated, generic-latest, and retry-predecessor evidence ineligible for
 Mission progression.
 
+When a Host exposes an explicit internal operator-retry chain, every successor
+must name its exact parent run and preserve the original Producer, correlation,
+Mission, Action, Runtime Prompt, project and repository identities. A successor
+is a distinct Host attempt: its accepted-request digest and an explicitly
+allowed execution baseline may differ from the original request. Forge accepts
+those attempt-specific facts only along the verified parent chain, requires
+readback/artifact digest parity and one exact baseline-transition SHA, and
+records the resolved attempt without representing it as a second Forge
+submission.
+
 ## Evidence and observability
 
 Every returned evidence envelope identifies the Host, correlation, host run,

forge/scheduler/ep_http_adapter.py

@@ -530,7 +530,22 @@ def _submission_receipt_id(self, request: ExecutionRequest, binding: Mapping[str
             raise ValueError("EP persisted submission receipt audit identity is invalid")
         return receipt_id
 
-    def _validate_readback(self, request: ExecutionRequest, binding: Mapping[str, Any], readback: Mapping[str, Any]) -> None:
+    @staticmethod
+    def _readback_accepted_request_digest(submission: Mapping[str, Any]) -> str:
+        digest = submission.get("accepted_request_digest")
+        if (not isinstance(digest, str) or not digest.startswith("sha256:") or len(digest) != 71
+                or any(character not in "0123456789abcdef" for character in digest[7:])):
+            raise ValueError("EP readback accepted-request digest is invalid")
+        return digest
+
+    def _validate_readback(
+        self,
+        request: ExecutionRequest,
+        binding: Mapping[str, Any],
+        readback: Mapping[str, Any],
+        *,
+        host_retry_successor: bool = False,
+    ) -> None:
         if readback.get("contract_version") not in self.SUPPORTED_PRODUCER_READBACK_CONTRACTS:
             raise ValueError("EP_READBACK_CONTRACT_INCOMPATIBLE")
         required = {"contract_version", "submission", "producer", "correlation", "provenance", "disposition", "run", "result", "evidence"}
@@ -547,8 +562,14 @@ def _validate_readback(self, request: ExecutionRequest, binding: Mapping[str, An
             raise ValueError("EP readback correlation does not bind persisted request")
         if submission.get("repository_id") != request.repository_id or submission.get("project_id") != self.config.project_id:
             raise ValueError("EP readback submission does not bind project and repository")
-        if (request.repository_revision_binding is not None
-                and submission.get("accepted_request_digest") != self._expected_ep_accepted_request_digest(request)):
+        accepted_request_digest = self._readback_accepted_request_digest(submission)
+        # Forge can recompute only the request it submitted.  An EP-owned,
+        # explicitly parent-bound operator retry is a distinct accepted
+        # attempt and therefore has its own digest.  Its remaining request
+        # identity is still checked below and its digest must agree with the
+        # immutable terminal artifact before evidence can be returned.
+        if (request.repository_revision_binding is not None and not host_retry_successor
+                and accepted_request_digest != self._expected_ep_accepted_request_digest(request)):
             raise ValueError("EP readback accepted-request digest does not bind persisted request")
         if dict(producer) != binding["producer"]:
             raise ValueError("EP readback producer does not bind persisted request")
@@ -619,11 +640,13 @@ def _readback(self, request: ExecutionRequest, binding: Mapping[str, Any]) -> di
         return self._readback_for_submission(request, binding, submission_id)
 
     def _readback_for_submission(self, request: ExecutionRequest, binding: Mapping[str, Any],
-                                 submission_id: str) -> dict[str, Any]:
+                                 submission_id: str, *, host_retry_successor: bool = False) -> dict[str, Any]:
         readback = self._json(
             f"/v1/projects/{self._segment(self.config.project_id)}/submissions/{self._segment(submission_id)}"
         )
-        self._validate_readback(request, binding, readback)
+        self._validate_readback(
+            request, binding, readback, host_retry_successor=host_retry_successor,
+        )
         if readback.get("submission", {}).get("id") != submission_id:
             raise ValueError("EP readback submission identity changed")
         return readback
@@ -653,7 +676,9 @@ def _operator_retry_resolution(self, request: ExecutionRequest, binding: Mapping
             if not isinstance(successor_id, str) or not successor_id or successor_id in seen_submissions:
                 raise ValueError("EP retry resolution successor identity is invalid")
             seen_submissions.add(successor_id)
-            successor = self._readback_for_submission(request, binding, successor_id)
+            successor = self._readback_for_submission(
+                request, binding, successor_id, host_retry_successor=True,
+            )
             successor_disposition = successor.get("disposition")
             if (not isinstance(successor_disposition, Mapping)
                     or successor_disposition.get("retry_parent_run_id") != expected_parent_run):
@@ -685,14 +710,19 @@ def _record_operator_retry_resolution(
         submission_id = submission.get("id") if isinstance(submission, Mapping) else None
         if not isinstance(submission_id, str) or not submission_id:
             raise ValueError("EP retry resolution submission identity is invalid")
+        accepted_request_digest = self._readback_accepted_request_digest(submission)
         resolution = {
             "submission_id": submission_id,
             "retry_parent_run_id": resolved_from_host_run_id,
             "run_id": evidence.host_run_id,
+            "accepted_request_digest": accepted_request_digest,
         }
         persisted = binding.get("operator_retry_resolution")
         if persisted is not None:
-            if persisted != resolution:
+            legacy_resolution = {
+                key: value for key, value in resolution.items() if key != "accepted_request_digest"
+            }
+            if persisted not in (resolution, legacy_resolution):
                 raise ValueError("EP retry resolution conflicts with the persisted Forge audit binding")
             return
         self._bindings.save_execution_host_binding(
@@ -710,6 +740,7 @@ def _record_operator_retry_resolution(
                 "resolution_submission_id": submission_id,
                 "retry_parent_run_id": resolved_from_host_run_id,
                 "resolved_from_host_run_id": resolved_from_host_run_id,
+                "accepted_request_digest": accepted_request_digest,
                 "result_state": evidence.outcome.value,
             },
         )

forge/scheduler/ep_v12.py

@@ -192,6 +192,7 @@ def _v14_repository_binding(
     *,
     outcome: str,
     delivery_qualified: bool,
+    host_retry_resolved: bool = False,
 ) -> tuple[str | None, str | None]:
     """Validate the v1.4 terminal fields against Forge's stored request.
 
@@ -235,8 +236,21 @@ def _v14_repository_binding(
     elif requested != revision_binding.requested_revision or transition_from != requested:
         raise ValueError("EP terminal requested revision differs from persisted Forge request")
     elif revision_binding.allowed_baseline_revision is None:
-        if (transition.get("status") != "EXACT" or transition_allowed is not None
-                or (baseline is not None and baseline != requested)):
+        exact = (transition.get("status") == "EXACT" and transition_allowed is None
+                 and (baseline is None or baseline == requested))
+        # EP owns operational retries.  Only the explicit parent-bound retry
+        # resolution path may introduce an attempt-specific allowed baseline;
+        # the original requested revision remains immutable and the terminal
+        # artifact must bind allowed_to, transition target and execution
+        # baseline to the same exact SHA.
+        retry_allowed = (
+            host_retry_resolved
+            and transition.get("status") == "ALLOWED"
+            and transition_allowed is not None
+            and baseline == transition_allowed
+            and transition_to == transition_allowed
+        )
+        if not exact and not retry_allowed:
             raise ValueError("EP terminal exact repository pin is inconsistent")
     elif (transition.get("status") != "ALLOWED"
           or transition_allowed != revision_binding.allowed_baseline_revision
@@ -362,6 +376,7 @@ def terminal_evidence(readback: Mapping[str, Any], artifact: bytes, *, host_id:
         _, revision = _v14_repository_binding(
             document, repository, document.get("delivery"), repository_revision_binding,
             outcome=outcome, delivery_qualified=qualified_readback,
+            host_retry_resolved=resolved_from_host_run_id is not None,
         )
     elif repository_revision_binding is not None:
         raise ValueError("EP historical terminal artifact cannot satisfy a v1.4 Forge request binding")

product-version.json

@@ -1,5 +1,5 @@
 {
   "product": "forge",
   "schema_version": 1,
-  "version": "2.7.20"
+  "version": "2.7.21"
 }

tests/test_ep_http_adapter.py

@@ -500,6 +500,13 @@ def test_v14_revision_binding_rejects_wrong_request_baseline_candidate_or_shape(
                 "execution_baseline": "b" * 40,
                 "baseline_transition": {"status": "EXACT", "from": "a" * 40, "to": "b" * 40, "allowed_to": None},
             })),
+            ("unapproved allowed baseline", lambda a: a["repository"].update({
+                "execution_baseline": "b" * 40,
+                "baseline_transition": {
+                    "status": "ALLOWED", "from": "a" * 40,
+                    "to": "b" * 40, "allowed_to": "b" * 40,
+                },
+            })),
             ("candidate", lambda a: a["repository"].update({"candidate": "d" * 40})),
             ("sha", lambda a: a["repository"].update({"execution_baseline": "not-a-sha"})),
             ("missing", lambda a: a["repository"].pop("candidate")),
@@ -848,12 +855,70 @@ def test_host_proven_operator_retry_resolution_returns_the_successor_evidence(se
         binding = self.database.execution_host_binding(self.request.correlation_id)
         self.assertEqual(binding["operator_retry_resolution"], {
             "submission_id": "retry-submission", "retry_parent_run_id": "run-fixture", "run_id": "retry-run",
+            "accepted_request_digest": self.readback["submission"]["accepted_request_digest"],
         })
         page = self.database.operational_log_page(correlation_id=self.request.correlation_id)
         event = next(item for item in page["items"] if item["event"] == "ep_operator_retry_resolution_evidence_accepted")
         self.assertEqual(event["run_id"], "retry-run")
         self.assertEqual(event["details"]["resolution_submission_id"], "retry-submission")
 
+    def test_host_proven_operator_retry_uses_its_attempt_digest_and_baseline_transition(self) -> None:
+        self._seed_binding()
+        parent = json.loads(json.dumps(self.readback))
+        parent["run"].update({"state": "BLOCKED", "terminal": False, "operator_resolution": "RETRIED"})
+        parent["result"].update({"outcome": "BLOCKED", "terminal": False, "delivery_qualified": False})
+        parent["evidence"]["terminal_artifact"] = None
+        parent["disposition"].update({"resolution_submission_id": "retry-submission", "retry_parent_run_id": None})
+
+        attempt_digest = "sha256:" + "b" * 64
+        retry_baseline = "b" * 40
+        successor = json.loads(json.dumps(self.readback))
+        successor["submission"].update({
+            "id": "retry-submission",
+            "accepted_request_digest": attempt_digest,
+        })
+        successor["run"].update({
+            "id": "retry-run", "state": "COMPLETE", "terminal": True,
+            "operator_resolution": "NONE",
+        })
+        successor["disposition"].update({
+            "resolution_submission_id": None,
+            "retry_parent_run_id": "run-fixture",
+        })
+
+        artifact = json.loads(self.artifact)
+        artifact["submission"].update({
+            "id": "retry-submission",
+            "accepted_request_digest": attempt_digest,
+        })
+        artifact["run"]["id"] = "retry-run"
+        artifact["repository"].update({
+            "execution_baseline": retry_baseline,
+            "baseline_transition": {
+                "status": "ALLOWED", "from": "a" * 40,
+                "to": retry_baseline, "allowed_to": retry_baseline,
+            },
+        })
+        raw = json.dumps(artifact, sort_keys=True, separators=(",", ":")).encode() + b"\n"
+        successor["evidence"]["terminal_artifact"]["digest"] = "sha256:" + hashlib.sha256(raw).hexdigest()
+
+        with patch("forge.scheduler.ep_http_adapter._open", self._urlopen([
+                json.dumps(self.compatible).encode(), json.dumps(parent).encode(),
+                json.dumps(successor).encode(), raw], [])):
+            evidence = EngineeringPlatformHttpExecutionHost(self.config, self.database).retrieve_evidence(
+                ExecutionDispatch(self.request, "run-fixture")
+            )
+
+        self.assertEqual(evidence.host_run_id, "retry-run")
+        self.assertEqual(evidence.resolved_from_host_run_id, "run-fixture")
+        binding = self.database.execution_host_binding(self.request.correlation_id)
+        self.assertEqual(binding["operator_retry_resolution"], {
+            "submission_id": "retry-submission",
+            "retry_parent_run_id": "run-fixture",
+            "run_id": "retry-run",
+            "accepted_request_digest": attempt_digest,
+        })
+
     def test_operator_retry_resolution_requires_its_exact_parent_run(self) -> None:
         self._seed_binding()
         parent = json.loads(json.dumps(self.readback))
@@ -872,6 +937,29 @@ def test_operator_retry_resolution_requires_its_exact_parent_run(self) -> None:
                     ExecutionDispatch(self.request, "run-fixture")
                 )
 
+    def test_operator_retry_resolution_rejects_a_noncanonical_attempt_digest(self) -> None:
+        self._seed_binding()
+        parent = json.loads(json.dumps(self.readback))
+        parent["run"].update({"state": "BLOCKED", "terminal": False, "operator_resolution": "RETRIED"})
+        parent["result"].update({"outcome": "BLOCKED", "terminal": False, "delivery_qualified": False})
+        parent["evidence"]["terminal_artifact"] = None
+        parent["disposition"].update({"resolution_submission_id": "retry-submission", "retry_parent_run_id": None})
+        successor = json.loads(json.dumps(self.readback))
+        successor["submission"].update({
+            "id": "retry-submission", "accepted_request_digest": "sha256:not-a-digest",
+        })
+        successor["run"].update({"id": "retry-run", "operator_resolution": "NONE"})
+        successor["disposition"].update({
+            "resolution_submission_id": None, "retry_parent_run_id": "run-fixture",
+        })
+        with patch("forge.scheduler.ep_http_adapter._open", self._urlopen([
+                json.dumps(self.compatible).encode(), json.dumps(parent).encode(),
+                json.dumps(successor).encode()], [])):
+            with self.assertRaisesRegex(ValueError, "accepted-request digest is invalid"):
+                EngineeringPlatformHttpExecutionHost(self.config, self.database).retrieve_evidence(
+                    ExecutionDispatch(self.request, "run-fixture")
+                )
+
     def test_configuration_and_transport_fail_closed_without_persisted_authority(self) -> None:
         with self.assertRaisesRegex(ValueError, "configuration"):
             EngineeringPlatformHttpExecutionHost(EngineeringPlatformHttpConfiguration("", "forge", "credential"), self.database)