fix(delegation): recover closed blocker drift

contract_tests/test_claiming.py

@@ -10,6 +10,7 @@
 import unittest
 from datetime import timedelta
 from pathlib import Path
+from types import SimpleNamespace
 from typing import Any
 
 from contract_tests import test_mailbox as fixtures
@@ -18,6 +19,7 @@
     EVIDENCE,
     OBSERVED_AT,
     git,
+    issue_reference,
     observation,
     write_json,
 )
@@ -41,6 +43,7 @@
     RESULT_SCHEMA,
     DelegationCoordinator,
     DelegationError,
+    _additive_closed_blocker_drift,
     _require_tracker_transition,
 )
 from skills.atelier.scripts.git_mailbox import (
@@ -61,6 +64,7 @@
     InitiativeDraft,
     Planner,
     PolicyTarget,
+    _ticket_material_digest,
 )
 
 APPROVED_EVIDENCE = EVIDENCE[:-1]
@@ -2616,6 +2620,11 @@ def fresh_coordinator() -> ClaimCoordinator:
         advanced_claim = self.claim(token="head-advance-token-0")
         advanced_delegation = self.delegation()
         advanced_invocation = self.delegated_invocation(advanced_claim)
+        invocation_observation_path = self.root / "advanced-invocation-observation.json"
+        write_json(
+            invocation_observation_path,
+            json.loads(self.observation_path.read_text(encoding="utf-8")),
+        )
         advanced_base = advanced_invocation["repository"]["base_sha"]
         self.assertNotEqual(advanced_base, candidate["base_sha"])
         candidate_tree = git(self.project_checkout, "rev-parse", "HEAD^{tree}").stdout.strip()
@@ -2701,6 +2710,20 @@ def fresh_coordinator() -> ClaimCoordinator:
         git(self.project_checkout, "push", "origin", "HEAD:main")
         current_base = git(self.project_checkout, "rev-parse", "HEAD").stdout.strip()
         self.assertNotEqual(current_base, advanced_invocation["repository"]["base_sha"])
+        closed_blocker_drift = json.loads(
+            invocation_observation_path.read_text(encoding="utf-8")
+        )
+        closed_blocker_drift["issue"]["blocked_by"].extend(
+            [
+                issue_reference(798, state="CLOSED"),
+                issue_reference(800, state="CLOSED"),
+            ]
+        )
+        current_ticket_digest = _ticket_material_digest(
+            closed_blocker_drift["issue"],
+            {"ticket": {"material_fields": ["body", "state", "relationships"]}},
+        )
+        write_json(self.observation_path, closed_blocker_drift)
         stale_checkpoint = self.delegated_request(
             advanced_invocation,
             advanced_push,
@@ -2759,6 +2782,7 @@ def fresh_coordinator() -> ClaimCoordinator:
             host_target=self.delegation_host_target(),
             observation_path=self.observation_path,
             observation_not_before=self.live_at,
+            invocation_observation_path=invocation_observation_path,
             ended_at=self.live_at + timedelta(minutes=3),
             now=self.live_at + timedelta(minutes=3),
         )
@@ -2806,6 +2830,11 @@ def fresh_coordinator() -> ClaimCoordinator:
                 "Delegated invocation base "
                 f"{advanced_invocation['repository']['base_sha']} is stale against "
                 f"current canonical base {current_base}.",
+                "Delegated ticket observation "
+                f"{advanced_invocation['ticket']['observation']} is stale against "
+                "current ticket observation "
+                f"{current_ticket_digest}; drift is limited "
+                "to additive closed blockers: #798 (issue-798), #800 (issue-800).",
             ],
         )
         self.assertEqual(prior_receipt["candidate"]["pull_request"], pull_request_url)
@@ -2909,6 +2938,7 @@ def fresh_coordinator() -> ClaimCoordinator:
         )
 
         live = observation(with_pull_request=True)
+        live["issue"] = json.loads(json.dumps(closed_blocker_drift["issue"]))
         live["observed_at"] = (
             (self.live_at + timedelta(minutes=9)).isoformat().replace("+00:00", "Z")
         )
@@ -3014,6 +3044,76 @@ def fresh_coordinator() -> ClaimCoordinator:
             ],
         )
 
+    def test_additive_closed_blocker_drift_is_exact_and_fail_closed(self) -> None:
+        prior = self.fresh_observation()
+        prior_path = self.root / "historical-ticket-observation.json"
+        write_json(prior_path, prior)
+        policy = {
+            "ticket": {
+                "material_fields": ["body", "state", "relationships"],
+            }
+        }
+        old_digest = _ticket_material_digest(prior["issue"], policy)
+        invocation = {
+            "repository": {
+                "identity": "github:example/project-1",
+            }
+        }
+        result = {"ticket": {"observation": old_digest}}
+
+        def drift_for(current: dict[str, Any], *, recoverable: bool = True):
+            state = SimpleNamespace(
+                observation=current,
+                effective_policy=policy,
+                ticket_digest=_ticket_material_digest(current["issue"], policy),
+            )
+            return _additive_closed_blocker_drift(
+                invocation,
+                result,
+                state,
+                invocation_observation_path=prior_path,
+                stale_base_recovery=recoverable,
+            )
+
+        current = json.loads(json.dumps(prior))
+        current["issue"]["blocked_by"].append(issue_reference(798, state="CLOSED"))
+        self.assertEqual(
+            drift_for(current),
+            (issue_reference(798, state="CLOSED"),),
+        )
+        self.assertIsNone(drift_for(current, recoverable=False))
+
+        invalid = json.loads(json.dumps(current))
+        invalid["issue"]["blocked_by"][-1]["state"] = "OPEN"
+        self.assertIsNone(drift_for(invalid))
+
+        invalid = json.loads(json.dumps(current))
+        invalid["issue"]["blocked_by"] = invalid["issue"]["blocked_by"][1:]
+        self.assertIsNone(drift_for(invalid))
+
+        invalid = json.loads(json.dumps(current))
+        invalid["issue"]["blocked_by"][0]["title"] = "Changed existing dependency"
+        self.assertIsNone(drift_for(invalid))
+
+        mutations = {
+            "title": lambda issue: issue.update(title="Changed title"),
+            "body": lambda issue: issue.update(body="Changed body"),
+            "state": lambda issue: issue.update(state="CLOSED"),
+            "updated_at": lambda issue: issue.update(updated_at="2026-07-28T05:00:00Z"),
+            "parent": lambda issue: issue.update(parent=issue_reference(900)),
+            "sub_issues": lambda issue: issue["sub_issues"].append(issue_reference(901)),
+            "blocking": lambda issue: issue["blocking"].append(issue_reference(902)),
+        }
+        for label, mutate in mutations.items():
+            with self.subTest(drift=label):
+                invalid = json.loads(json.dumps(current))
+                mutate(invalid["issue"])
+                self.assertIsNone(drift_for(invalid))
+
+        invalid = json.loads(json.dumps(current))
+        invalid["repository"]["name_with_owner"] = "example/other-project"
+        self.assertIsNone(drift_for(invalid))
+
     def test_delegation_delivers_one_exact_ready_pull_request(self) -> None:
         claimed = self.claim()
         self.coordinator = ClaimCoordinator(

skills/atelier/references/delegation.md

@@ -74,6 +74,19 @@ match the still-open current native ticket before recording a receipt.
   fresh claimant may advance the recovered candidate to current main only through
   an exact push and `candidate_published` acknowledgement before restoring PR
   metadata under fresh exact authority.
+- If installing the recovery also adds native blockers after the sealed ticket
+  observation, the caller may provide `invocation_observation_path` for that same
+  terminal blocked recovery only. Atelier validates the historical observation,
+  binds its material digest to the unchanged invocation and result, and compares
+  it with the fresh terminal observation. The repository, ticket identity, title,
+  body, state, update timestamp, parent, sub-issues, blocking relationships, and
+  every pre-existing blocker must be identical. The only permitted difference is
+  a nonempty set of additive blockers that are all currently closed. The receipt
+  records both ticket-observation digests and the exact added blocker numbers and
+  identities as a durable obligation. Missing historical evidence, open or
+  reopened blockers, removed or changed dependencies, and every other ticket
+  change fail closed. This exception never applies to checkpoints, `ready_pr`,
+  delivery, audit, or acceptance.
 - `requires_epic` is recorded as a blocked attempt for planner action. Atelier
   does not invoke `implement-epic` from delegated work.
 
@@ -91,6 +104,8 @@ The CLI accepts one JSON object containing:
 - `observation_path` and `observation_not_before`;
 - for `prepare`, `checkpoint_invocation_path` and the host-owned
   `observation_command`; and
+- for the exact additive-closed-blocker terminal recovery,
+  `invocation_observation_path`; and
 - the remaining operation-specific fence, invocation, checkpoint request, result,
   and timestamps.
 

skills/atelier/scripts/delegation.py

@@ -48,7 +48,11 @@
 )
 from skills.atelier.scripts.host_boundary import HostBoundaryError, check_host, validate_observation
 from skills.atelier.scripts.identifiers import new_identifier
-from skills.atelier.scripts.planning import PolicyTarget, _render_document
+from skills.atelier.scripts.planning import (
+    PolicyTarget,
+    _render_document,
+    _ticket_material_digest,
+)
 
 CAPABILITY = "agent-scripts.implement-ticket/delegated-execution/v2"
 INVOCATION_SCHEMA = "agent-scripts.implement-ticket/delegated-invocation/v2"
@@ -344,6 +348,7 @@ def finalize(
         host_target: HostTarget,
         observation_path: Path,
         observation_not_before: datetime,
+        invocation_observation_path: Path | None = None,
         ended_at: datetime,
         now: datetime | None = None,
     ) -> WriteResult:
@@ -427,20 +432,43 @@ def revalidate(context: TransitionContext) -> None:
                 raise MailboxTransitionRejected(
                     "delegated invocation does not match its sealed digest"
                 )
+            ticket_drift = None
             if result["ticket"]["observation"] != state.ticket_digest:
-                raise MailboxTransitionRejected("terminal ticket observation is not current")
+                ticket_drift = _additive_closed_blocker_drift(
+                    invocation,
+                    result,
+                    state,
+                    invocation_observation_path=invocation_observation_path,
+                    stale_base_recovery=stale_base_recovery,
+                )
+                if ticket_drift is None:
+                    raise MailboxTransitionRejected(
+                        "terminal ticket observation is not current"
+                    )
             _require_tracker_transition(result, state.observation)
             evidence = _attempt_evidence(result)
-            if stale_base_recovery:
+            if stale_base_recovery or ticket_drift is not None:
+                obligations = list(evidence.unresolved_obligations)
+                if stale_base_recovery:
+                    obligations.append(
+                        "Delegated invocation base "
+                        f"{invocation['repository']['base_sha']} is stale against "
+                        f"current canonical base {state.current_policy.commit}."
+                    )
+                if ticket_drift is not None:
+                    blockers = ", ".join(
+                        f"#{item['number']} ({item['id']})" for item in ticket_drift
+                    )
+                    obligations.append(
+                        "Delegated ticket observation "
+                        f"{result['ticket']['observation']} is stale against current "
+                        f"ticket observation {state.ticket_digest}; drift is limited "
+                        f"to additive closed blockers: {blockers}."
+                    )
                 evidence = AttemptEvidence(
                     validation=evidence.validation,
                     reviews=evidence.reviews,
-                    unresolved_obligations=(
-                        *evidence.unresolved_obligations,
-                        "Delegated invocation base "
-                        f"{invocation['repository']['base_sha']} is stale against "
-                        f"current canonical base {state.current_policy.commit}.",
-                    ),
+                    unresolved_obligations=tuple(obligations),
                 )
             outcome = "delivered" if terminal == "ready_pr" else "blocked"
             body = (
@@ -948,6 +976,69 @@ def _stale_base_blocked_recovery(
     )
 
 
+def _additive_closed_blocker_drift(
+    invocation: Mapping[str, Any],
+    result: Mapping[str, Any],
+    state: Any,
+    *,
+    invocation_observation_path: Path | None,
+    stale_base_recovery: bool,
+) -> tuple[dict[str, Any], ...] | None:
+    if not stale_base_recovery or invocation_observation_path is None:
+        return None
+    prior_value = _read_json(invocation_observation_path)
+    observed_at = _parse_timestamp(prior_value["observed_at"])
+    prior = validate_observation(
+        invocation_observation_path,
+        not_before=observed_at,
+        now=observed_at,
+    )
+    if (
+        prior["repository"] != state.observation["repository"]
+        or prior["repository"]["name_with_owner"]
+        != invocation["repository"]["identity"].removeprefix("github:")
+    ):
+        return None
+    prior_issue = prior["issue"]
+    current_issue = state.observation["issue"]
+    if (
+        _ticket_material_digest(prior_issue, state.effective_policy)
+        != result["ticket"]["observation"]
+    ):
+        return None
+    unchanged_fields = (
+        "id",
+        "number",
+        "title",
+        "body",
+        "state",
+        "url",
+        "updated_at",
+        "parent",
+        "sub_issues",
+        "blocking",
+    )
+    if any(prior_issue[name] != current_issue[name] for name in unchanged_fields):
+        return None
+    prior_blockers = {item["id"]: item for item in prior_issue["blocked_by"]}
+    current_blockers = {item["id"]: item for item in current_issue["blocked_by"]}
+    if any(current_blockers.get(identifier) != item for identifier, item in prior_blockers.items()):
+        return None
+    added = tuple(
+        sorted(
+            (
+                item
+                for identifier, item in current_blockers.items()
+                if identifier not in prior_blockers
+            ),
+            key=lambda item: (item["number"], item["id"]),
+        )
+    )
+    if not added or any(item["state"] != "CLOSED" for item in added):
+        return None
+    return added
+
+
 def _require_current_invocation_contract(
     invocation: Mapping[str, Any],
     state: Any,
@@ -1251,6 +1342,11 @@ def execute_request(value: Mapping[str, Any]) -> dict[str, Any]:
             value["invocation"],
             value["result"],
             _fence(value["fence"]),
+            invocation_observation_path=(
+                Path(value["invocation_observation_path"])
+                if value.get("invocation_observation_path")
+                else None
+            ),
             ended_at=_parse_timestamp(value["ended_at"]),
             **common,
         )