diff --git a/contract_tests/test_claiming.py b/contract_tests/test_claiming.py index 106436f8..ade02159 100644 --- a/contract_tests/test_claiming.py +++ b/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( diff --git a/skills/atelier/references/delegation.md b/skills/atelier/references/delegation.md index e26abb43..bf881c4b 100644 --- a/skills/atelier/references/delegation.md +++ b/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. diff --git a/skills/atelier/scripts/delegation.py b/skills/atelier/scripts/delegation.py index e6a27bf8..bc471c82 100644 --- a/skills/atelier/scripts/delegation.py +++ b/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, )