diff --git a/AGENTS.md b/AGENTS.md index 64361493..fa3bb43d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -15,9 +15,10 @@ The [Atelier as a Skill], [Git Mailbox Contract], [Project Policy Contract], and ## Current state -The repository is a reset scaffold. The `/atelier` skill is discoverable but -production `plan`, `work`, and `audit` modes are not implemented. The open -native issue graph beginning at #772 defines implementation order. +The repository is a reset scaffold. The `/atelier` skill implements one +GitHub-backed planning assignment and one claimed, delegated `ready_pr` work +path. Production `audit` is not implemented. The open native issue graph +beginning at #772 defines implementation order. Do not claim unavailable behavior or emulate it with ad hoc orchestration. @@ -63,9 +64,7 @@ just lint just test ``` -The expected-failure contract test is intentional. It names the missing `HOST` -capability and must become an ordinary passing test when #774 implements that -contract. +All contract tests are expected to pass. ## Commits and publication diff --git a/contract_tests/test_claiming.py b/contract_tests/test_claiming.py index ccca94fa..62693642 100644 --- a/contract_tests/test_claiming.py +++ b/contract_tests/test_claiming.py @@ -2,7 +2,9 @@ from __future__ import annotations +import json import subprocess +import sys import tempfile import threading import unittest @@ -32,6 +34,15 @@ _render_document, new_identifier, ) +from skills.atelier.scripts.delegation import ( + CAPABILITY, + INVOCATION_SCHEMA, + REQUEST_SCHEMA, + RESULT_SCHEMA, + DelegationCoordinator, + DelegationError, + _require_tracker_transition, +) from skills.atelier.scripts.git_mailbox import ( FileChange, GitMailboxWriter, @@ -48,9 +59,77 @@ PolicyTarget, ) +APPROVED_EVIDENCE = EVIDENCE[:-1] DIGEST = "sha256:" + "d" * 64 HEAD = "b" * 40 CLAIMING_SCRIPT = Path(__file__).parents[1] / "skills/atelier/scripts/claiming.py" +ROOT = Path(__file__).parents[1] +HOST_CAPABILITY = ROOT / "skills/atelier/references/host-capability.json" +INSTALLED_TICKET_SKILL = Path.home() / ".agents/skills/implement-ticket" +REQUIRED_HOST_OPERATIONS = ( + "github.issue.read", + "github.issue.relationships.read", + "github.pull-request.read", + "github.pull-request.comments.read", + "github.pull-request.reviews.read", + "github.pull-request.checks.read", + "github.pull-request.threads.read", +) + + +class ProtocolFixture: + def validate(self, kind: str, value: dict[str, Any]) -> list[str]: + expected = { + "invocation": INVOCATION_SCHEMA, + "checkpoint-request": REQUEST_SCHEMA, + "result": RESULT_SCHEMA, + } + return [] if value.get("schema") == expected[kind] else ["wrong protocol schema"] + + def validate_checkpoint_progress( + self, + last_sequence: int, + current_token: str, + request: dict[str, Any], + response: dict[str, Any], + ) -> list[str]: + errors = [] + if request["sequence"] != last_sequence + 1: + errors.append("sequence did not advance exactly once") + if request["continuation_token"] != current_token: + errors.append("continuation token is stale") + if response["request_sequence"] != request["sequence"]: + errors.append("response sequence mismatch") + return errors + + def validate_checkpoint_exchange( + self, + request: dict[str, Any], + response: dict[str, Any], + ) -> list[str]: + return [] if response["request_sequence"] == request["sequence"] else ["sequence mismatch"] + + def validate_result_checkpoint_state( + self, + invocation: dict[str, Any], + result: dict[str, Any], + last_sequence: int, + current_token: str, + observed_deployment, + observed_tracker, + consumed_authority: list[str], + ) -> list[str]: + errors = [] + if result["invocation_id"] != invocation["invocation_id"]: + errors.append("result invocation mismatch") + if result["checkpoint"] != { + "last_sequence": last_sequence, + "continuation_token": current_token, + }: + errors.append("result checkpoint mismatch") + if set(result["authority_used"]) != set(consumed_authority): + errors.append("result authority mismatch") + return errors class ClaimingContract(unittest.TestCase): @@ -98,6 +177,21 @@ def setUp(self) -> None: self.observation_path = self.root / "observation.json" write_json(self.observation_path, observation()) + observation_script = self.root / "observe.py" + observation_script.write_text( + """import json +import sys +from datetime import UTC, datetime + +path = sys.argv[1] +with open(path, encoding=\"utf-8\") as stream: + value = json.load(stream) +value[\"observed_at\"] = datetime.now(UTC).isoformat().replace(\"+00:00\", \"Z\") +print(json.dumps(value, sort_keys=True)) +""", + encoding="utf-8", + ) + self.observation_command = (sys.executable, str(observation_script), str(self.observation_path)) self.work_id = fixtures.identifier("wrk", 778) self.initiative_id = fixtures.identifier("ini", 778) planner = Planner(str(self.mailbox_remote), "main") @@ -111,7 +205,7 @@ def setUp(self) -> None: preview = planner.preview_approval( self.work_id, expected_revision=1, - envelope=ApprovalEnvelope(AUTHORITY, EVIDENCE), + envelope=ApprovalEnvelope(AUTHORITY, APPROVED_EVIDENCE), policy_target=self.policy_target(), observation_path=self.observation_path, observation_not_before=OBSERVED_AT, @@ -140,7 +234,12 @@ def setUp(self) -> None: def tearDown(self) -> None: self.temporary.cleanup() - def write_policy(self, *, authority: tuple[str, ...] = AUTHORITY) -> None: + def write_policy( + self, + *, + authority: tuple[str, ...] = AUTHORITY, + evidence: tuple[str, ...] = APPROVED_EVIDENCE, + ) -> None: fixtures.write_yaml( self.policy_path, { @@ -162,16 +261,36 @@ def write_policy(self, *, authority: tuple[str, ...] = AUTHORITY) -> None: "material_fields": ["body", "state", "relationships"], }, "execution": { - "capability": "agent-scripts.implement-ticket/delegated-execution/v1", + "capability": "agent-scripts.implement-ticket/delegated-execution/v2", "delivery_outcome": "ready_pr", "parallel_assignments": False, }, "authority": {"allow": list(authority)}, "validation": {"required_commands": ["just test", "just lint"]}, - "acceptance": {"actor": "operator", "evidence": list(EVIDENCE)}, + "acceptance": {"actor": "operator", "evidence": list(evidence)}, }, ) + def commit_policy_change( + self, + *, + authority: tuple[str, ...] = AUTHORITY, + evidence: tuple[str, ...] = APPROVED_EVIDENCE, + ) -> None: + self.write_policy(authority=authority, evidence=evidence) + git(self.project_checkout, "add", ".atelier/policy.yaml") + git( + self.project_checkout, + "-c", + "user.name=Atelier Test", + "-c", + "user.email=atelier-test@invalid", + "commit", + "-m", + "change project policy", + ) + git(self.project_checkout, "push", "origin", "HEAD:main") + def policy_target(self, checkout: Path | None = None) -> PolicyTarget: return PolicyTarget( checkout=checkout or self.project_checkout, @@ -180,7 +299,6 @@ def policy_target(self, checkout: Path | None = None) -> PolicyTarget: path=".atelier/policy.yaml", ) - def policy_remote_matches(self, target: PolicyTarget, repository: str) -> bool: configured = run_git(target.checkout, ("remote", "get-url", target.remote)) if configured.returncode != 0 or repository != self.repository: @@ -196,6 +314,37 @@ def host_target(self) -> HostTarget: operations=("read_issue",), ) + def delegation_host_target(self) -> HostTarget: + return HostTarget( + descriptor_path=self.root / "host-capability.json", + skill_name="agent-scripts:implement-ticket", + skill_root=self.root / "agent-scripts", + connector="github@openai-curated", + operations=("read_issue",), + ) + + def installed_host_target(self) -> HostTarget: + return HostTarget( + descriptor_path=HOST_CAPABILITY, + skill_name="agent-scripts:implement-ticket", + skill_root=INSTALLED_TICKET_SKILL, + connector="github@openai-curated", + operations=REQUIRED_HOST_OPERATIONS, + ) + + def assert_installed_result_valid_if_available(self, result: dict[str, Any]) -> None: + if not (INSTALLED_TICKET_SKILL / "SKILL.md").is_file(): + return + dependency = DelegationCoordinator(self.coordinator)._dependency( + self.installed_host_target() + ) + self.assertEqual(dependency.validate("result", result), []) + + def delegation(self) -> DelegationCoordinator: + coordinator = DelegationCoordinator(self.coordinator) + coordinator._dependency = lambda target: ProtocolFixture() + return coordinator + def fresh_observation(self) -> dict[str, Any]: value = observation() value["observed_at"] = self.live_at.isoformat().replace("+00:00", "Z") @@ -274,11 +423,14 @@ def checkpoint( action: str, token: str, candidate_head: str | None = None, + candidate_remote_ref: str | None = None, phase: str = "pre_external_mutation", candidate: dict[str, Any] | None = None, coordinator: ClaimCoordinator | None = None, host_target: HostTarget | None = None, ): + if candidate_head is not None and candidate_remote_ref is None: + candidate_remote_ref = (candidate or self.candidate())["remote_ref"] return (coordinator or self.coordinator).authorize( self.work_id, CheckpointRequest( @@ -287,6 +439,7 @@ def checkpoint( action=action, proposed_effect_digest=DIGEST, candidate_head=candidate_head, + candidate_remote_ref=candidate_remote_ref, acknowledged_candidate_head=( candidate_head if phase == "candidate_published" else None ), @@ -315,6 +468,307 @@ def candidate(self) -> dict[str, Any]: "published_at": "2026-07-28T04:02:00Z", } + def delegated_invocation(self, claimed): + return self.delegation().prepare( + self.work_id, + self.fence(claimed), + approved_commit=self.approved_commit, + policy_target=self.policy_target(), + host_target=self.delegation_host_target(), + observation_path=self.observation_path, + observation_not_before=self.live_at, + checkpoint_invocation_path=self.root / "delegated-invocation.json", + observation_command=self.observation_command, + now=self.live_at + timedelta(seconds=30), + ) + + def delegated_request( + self, + invocation: dict[str, Any], + prior, + *, + action: str, + phase: str = "pre_external_mutation", + candidate: dict[str, Any] | None = None, + ) -> dict[str, Any]: + return { + "schema": REQUEST_SCHEMA, + "capability": CAPABILITY, + "invocation_id": invocation["invocation_id"], + "continuation_token": prior.continuation_token, + "sequence": prior.sequence + 1, + "phase": phase, + "action": action, + "ticket_observation": invocation["ticket"]["observation"], + "candidate": candidate, + "deployment": None, + "proposed_effect": f"{action} candidate", + } + + def delegated_checkpoint( + self, + delegation: DelegationCoordinator, + invocation: dict[str, Any], + prior, + *, + action: str, + token: str, + phase: str = "pre_external_mutation", + candidate: dict[str, Any] | None = None, + ): + request = self.delegated_request( + invocation, + prior, + action=action, + phase=phase, + candidate=candidate, + ) + response = delegation.checkpoint( + self.work_id, + invocation, + request, + approved_commit=self.approved_commit, + policy_target=self.policy_target(), + host_target=self.delegation_host_target(), + observation_path=self.observation_path, + observation_not_before=self.live_at, + recorded_at=self.live_at + timedelta(minutes=2), + next_continuation_token=token, + now=self.live_at + timedelta(minutes=2), + ) + self.assertEqual(response["decision"], "allow", response) + return type(prior)( + operation="checkpoint", + work_id=self.work_id, + status="active", + claim_id=prior.claim_id, + worker_run_id=prior.worker_run_id, + sequence=request["sequence"], + continuation_token=response["continuation_token"], + commit="0" * 40, + base_revision="0" * 40, + branch="main", + attempts=1, + recovered=False, + ) + + def assert_checkpoint_denied_without_mutation( + self, + delegation: DelegationCoordinator, + invocation: dict[str, Any], + request: dict[str, Any], + ) -> dict[str, Any]: + before = git( + None, + "--git-dir", + str(self.mailbox_remote), + "rev-parse", + "main", + ).stdout.strip() + response = delegation.checkpoint( + self.work_id, + invocation, + request, + approved_commit=self.approved_commit, + policy_target=self.policy_target(), + host_target=self.delegation_host_target(), + observation_path=self.observation_path, + observation_not_before=self.live_at, + recorded_at=self.live_at + timedelta(minutes=2), + next_continuation_token="must-not-be-used", + now=self.live_at + timedelta(minutes=2), + ) + after = git( + None, + "--git-dir", + str(self.mailbox_remote), + "rev-parse", + "main", + ).stdout.strip() + self.assertEqual(response["decision"], "deny") + self.assertEqual(response["continuation_token"], request["continuation_token"]) + self.assertEqual(after, before) + return response + + def acceptance_records(self, invocation: dict[str, Any], candidate_head: str): + return [ + { + "criterion": item["criterion"], + "required": item["required"], + "evidence_category": item["evidence_category"], + "stage": item["stage"], + "candidate_sha": candidate_head, + "deployed_sha": None, + "environment": item["environment"], + "url": item["url"], + "source": item["source"], + "status": "pass", + } + for item in invocation["acceptance_requirements"] + ] + + def blocked_result( + self, + invocation: dict[str, Any], + checkpointed, + *, + tracker_mode: str = "none", + tracker_state: str = "open", + reviews: list[dict[str, Any]] | None = None, + candidate: dict[str, Any] | None = None, + authority_used: list[str] | None = None, + ) -> dict[str, Any]: + return { + "schema": RESULT_SCHEMA, + "capability": CAPABILITY, + "invocation_id": invocation["invocation_id"], + "terminal_state": "blocked", + "ticket": invocation["ticket"], + "repository": { + "identity": invocation["repository"]["identity"], + "base_ref": invocation["repository"]["base_ref"], + "base_sha": invocation["repository"]["base_sha"], + }, + "tracker_transition": { + "provider": invocation["ticket"]["provider"], + "ticket_id": invocation["ticket"]["id"], + "mode": tracker_mode, + "state": tracker_state, + "observed_at": fixtures.TIMESTAMP, + }, + "implementation_state": "published" if candidate else "local", + "candidate": candidate, + "handoff": { + "transferable": candidate is not None, + "reason": None if candidate else "No published candidate exists.", + }, + "checkpoint": { + "last_sequence": checkpointed.sequence, + "continuation_token": checkpointed.continuation_token, + }, + "validation": [], + "reviews": reviews or [], + "feedback": None, + "authority_used": ( + authority_used + if authority_used is not None + else ["repository.candidate.create"] + ), + "acceptance_evidence": [ + { + "criterion": item["criterion"], + "required": item["required"], + "evidence_category": item["evidence_category"], + "stage": item["stage"], + "candidate_sha": candidate["head_sha"] if candidate else None, + "deployed_sha": None, + "environment": item["environment"], + "url": item["url"], + "source": None, + "status": "missing", + } + for item in invocation["acceptance_requirements"] + ], + "unresolved_obligations": ["Resolve the implementation blocker."], + "blocking_reason": "The delegated worker could not publish a candidate.", + "next_action": "Planner decides whether to retry or revise the work.", + } + + def assert_blocked_ref_substitution_rejected(self, *, with_prior_candidate: bool) -> None: + self.coordinator = ClaimCoordinator( + str(self.mailbox_remote), + "main", + candidate_verifier=lambda candidate: True, + capability_verifier=lambda target: True, + policy_remote_verifier=self.policy_remote_matches, + ) + claimed = self.claim() + delegation = self.delegation() + invocation = self.delegated_invocation(claimed) + prior = claimed + if with_prior_candidate: + acknowledged = { + "repository": invocation["repository"]["identity"], + "remote_url": invocation["repository"]["remote_url"], + "remote_ref": "refs/heads/scott/acknowledged-candidate", + "base_sha": invocation["repository"]["base_sha"], + "head_sha": HEAD, + } + pushed = self.delegated_checkpoint( + delegation, + invocation, + prior, + action="repository.candidate.push", + token="token-1", + candidate=acknowledged, + ) + prior = self.delegated_checkpoint( + delegation, + invocation, + pushed, + action="repository.candidate.push", + token="token-2", + phase="candidate_published", + candidate=acknowledged, + ) + authorized = { + "repository": invocation["repository"]["identity"], + "remote_url": invocation["repository"]["remote_url"], + "remote_ref": "refs/heads/scott/authorized-push", + "base_sha": invocation["repository"]["base_sha"], + "head_sha": "c" * 40 if with_prior_candidate else HEAD, + } + pushed = self.delegated_checkpoint( + delegation, + invocation, + prior, + action="repository.candidate.push", + token="token-3" if with_prior_candidate else "token-1", + candidate=authorized, + ) + substituted = { + **authorized, + "remote_ref": "refs/heads/scott/substituted-same-head", + "publication": {"kind": "ordinary", "pull_requests": []}, + } + result = self.blocked_result( + invocation, + pushed, + candidate=substituted, + authority_used=["repository.candidate.push"], + ) + before = git( + None, + "--git-dir", + str(self.mailbox_remote), + "rev-parse", + "main", + ).stdout.strip() + + with self.assertRaisesRegex(MailboxTransitionRejected, "exact push authorization"): + delegation.finalize( + self.work_id, + invocation, + result, + self.fence(pushed), + approved_commit=self.approved_commit, + policy_target=self.policy_target(), + host_target=self.delegation_host_target(), + observation_path=self.observation_path, + observation_not_before=self.live_at, + ended_at=self.live_at + timedelta(minutes=3), + now=self.live_at + timedelta(minutes=3), + ) + + after = git( + None, + "--git-dir", + str(self.mailbox_remote), + "rev-parse", + "main", + ).stdout.strip() + self.assertEqual(after, before) + def publish_candidate_with_descendant(self, *, with_pull_request: bool = False): candidate_head = git(self.project_checkout, "rev-parse", "HEAD").stdout.strip() candidate_ref = "refs/heads/scott/example-work" @@ -416,9 +870,7 @@ def revalidate(context) -> None: "mechanism": "review-code-change", "verdict": "clean", "candidate_revision": candidate_head, - "comparison_base_revision": work["claim"]["candidate"][ - "base_revision" - ], + "comparison_base_revision": work["claim"]["candidate"]["base_revision"], "observed_at": fixtures.TIMESTAMP, }, ), @@ -559,7 +1011,6 @@ def test_candidate_publication_requires_paired_push_and_exact_remote_reachabilit ["pre_external_mutation", "candidate_published"], ) - def test_default_policy_remote_verifier_binds_github_repository_identity(self) -> None: git( self.project_checkout, @@ -631,7 +1082,6 @@ def test_policy_tightening_denies_removed_action_without_widening_claim_authorit candidate_head=HEAD, ) - def test_active_takeover_preserves_published_candidate_with_handoff_receipt(self) -> None: published, candidate_head = self.publish_candidate_with_descendant() prior_fence = self.fence(published) @@ -856,9 +1306,7 @@ def test_blocked_release_keeps_decision_historical_without_resolving_it(self) -> self.assertEqual(reblocked_snapshot["views"]["blocked"], [self.work_id]) def test_delivered_takeover_returns_active_with_historical_delivery(self) -> None: - published, candidate_head = self.publish_candidate_with_descendant( - with_pull_request=True - ) + published, candidate_head = self.publish_candidate_with_descendant(with_pull_request=True) receipt_id = self.deliver_candidate(published, candidate_head) takeover = self.coordinator.takeover( self.work_id, @@ -1052,6 +1500,979 @@ def test_missing_capability_and_wrong_approval_commit_fail_before_mailbox_write( after = git(None, "--git-dir", str(self.mailbox_remote), "rev-parse", "main").stdout.strip() self.assertEqual(after, before) + def test_delegation_uses_current_policy_acceptance_after_approval_tightens(self) -> None: + self.write_policy(evidence=EVIDENCE) + git(self.project_checkout, "add", ".atelier/policy.yaml") + git( + self.project_checkout, + "-c", + "user.name=Atelier Test", + "-c", + "user.email=atelier-test@invalid", + "commit", + "-m", + "tighten acceptance evidence", + ) + git(self.project_checkout, "push", "origin", "HEAD:main") + + claimed = self.claim() + invocation = self.delegated_invocation(claimed) + + self.assertEqual( + [item["criterion"] for item in invocation["acceptance_requirements"]], + list(EVIDENCE), + ) + + def test_delegation_rejects_tracker_mutation_for_ready_and_blocked_results(self) -> None: + current = observation() + for terminal in ("ready_pr", "blocked"): + for transition in ( + {"provider": "github", "ticket_id": "777", "mode": "manual", "state": "open"}, + { + "provider": "github", + "ticket_id": "777", + "mode": "automatic", + "state": "closed", + }, + ): + with self.subTest(terminal=terminal, transition=transition): + result = { + "terminal_state": terminal, + "ticket": {"provider": "github"}, + "tracker_transition": transition, + } + with self.assertRaisesRegex(MailboxTransitionRejected, "tracker transition"): + _require_tracker_transition(result, current) + + def test_delegation_prepares_v2_invocation_and_records_blocked_receipt(self) -> None: + claimed = self.claim() + delegation = self.delegation() + invocation = delegation.prepare( + self.work_id, + self.fence(claimed), + approved_commit=self.approved_commit, + policy_target=self.policy_target(), + host_target=self.delegation_host_target(), + observation_path=self.observation_path, + observation_not_before=self.live_at, + checkpoint_invocation_path=self.root / "delegated-invocation.json", + observation_command=self.observation_command, + now=self.live_at + timedelta(seconds=30), + ) + self.assertEqual(invocation["schema"], INVOCATION_SCHEMA) + self.assertEqual(invocation["capability"], CAPABILITY) + self.assertEqual(invocation["ticket"]["observation"], self.fresh_observation_digest()) + self.assertEqual( + invocation["accepted_terminal_states"], ["ready_pr", "blocked", "requires_epic"] + ) + checkpointed = self.delegated_checkpoint( + delegation, + invocation, + claimed, + action="repository.candidate.create", + token="token-1", + ) + result = { + "schema": RESULT_SCHEMA, + "capability": CAPABILITY, + "invocation_id": invocation["invocation_id"], + "terminal_state": "blocked", + "ticket": invocation["ticket"], + "repository": { + "identity": invocation["repository"]["identity"], + "base_ref": invocation["repository"]["base_ref"], + "base_sha": invocation["repository"]["base_sha"], + }, + "tracker_transition": { + "provider": "github", + "ticket_id": invocation["ticket"]["id"], + "mode": "none", + "state": "open", + "observed_at": fixtures.TIMESTAMP, + }, + "implementation_state": "local", + "candidate": None, + "handoff": {"transferable": False, "reason": "No published candidate exists."}, + "checkpoint": { + "last_sequence": checkpointed.sequence, + "continuation_token": checkpointed.continuation_token, + }, + "validation": [], + "reviews": [], + "feedback": None, + "authority_used": ["repository.candidate.create"], + "acceptance_evidence": [ + { + "criterion": item["criterion"], + "required": item["required"], + "evidence_category": item["evidence_category"], + "stage": item["stage"], + "candidate_sha": None, + "deployed_sha": None, + "environment": item["environment"], + "url": item["url"], + "source": None, + "status": "missing", + } + for item in invocation["acceptance_requirements"] + ], + "unresolved_obligations": ["Resolve the implementation blocker."], + "blocking_reason": "The delegated worker could not publish a candidate.", + "next_action": "Planner decides whether to retry or revise the work.", + } + delegation.finalize( + self.work_id, + invocation, + result, + self.fence(checkpointed), + approved_commit=self.approved_commit, + policy_target=self.policy_target(), + host_target=self.delegation_host_target(), + observation_path=self.observation_path, + observation_not_before=self.live_at, + ended_at=self.live_at + timedelta(minutes=3), + now=self.live_at + timedelta(minutes=3), + ) + checkout = self.mailbox_clone("delegated-blocked-read") + work, _ = _read_yaml( + checkout / f"work/{self.work_id}/work.md", + frontmatter=True, + label="work", + ) + receipt, _ = _read_yaml( + checkout / f"work/{self.work_id}/receipts/{work['attempt_receipt_id']}.md", + frontmatter=True, + label="receipt", + ) + self.assertEqual(work["status"], "blocked") + self.assertEqual(receipt["outcome"], "blocked") + self.assertEqual(receipt["unresolved_obligations"], result["unresolved_obligations"]) + reconstruct_mailbox(checkout) + + def test_delegation_accepts_v2_blocked_result_without_obligations(self) -> None: + claimed = self.claim() + fixture_delegation = self.delegation() + invocation = self.delegated_invocation(claimed) + checkpointed = self.delegated_checkpoint( + fixture_delegation, + invocation, + claimed, + action="repository.candidate.create", + token="token-1", + ) + result = self.blocked_result(invocation, checkpointed) + result["unresolved_obligations"] = [] + self.assert_installed_result_valid_if_available(result) + + fixture_delegation.finalize( + self.work_id, + invocation, + result, + self.fence(checkpointed), + approved_commit=self.approved_commit, + policy_target=self.policy_target(), + host_target=self.delegation_host_target(), + observation_path=self.observation_path, + observation_not_before=self.live_at, + ended_at=self.live_at + timedelta(minutes=3), + now=self.live_at + timedelta(minutes=3), + ) + + checkout = self.mailbox_clone("installed-v2-blocked-read") + work, _ = _read_yaml( + checkout / f"work/{self.work_id}/work.md", + frontmatter=True, + label="work", + ) + receipt, body = _read_yaml( + checkout / f"work/{self.work_id}/receipts/{work['attempt_receipt_id']}.md", + frontmatter=True, + label="receipt", + ) + self.assertEqual(work["status"], "blocked") + self.assertEqual(receipt["unresolved_obligations"], []) + self.assertEqual(body, result["blocking_reason"]) + reconstruct_mailbox(checkout) + + def test_delegation_records_v2_requires_epic_result(self) -> None: + claimed = self.claim() + fixture_delegation = self.delegation() + invocation = self.delegated_invocation(claimed) + result = self.blocked_result(invocation, claimed, authority_used=[]) + result.update( + { + "terminal_state": "requires_epic", + "implementation_state": "none", + "candidate": None, + "handoff": { + "transferable": False, + "reason": "Whole epic requires implement-epic", + }, + "unresolved_obligations": [], + "blocking_reason": None, + "next_action": "Return the work to the planner for epic decomposition.", + } + ) + self.assert_installed_result_valid_if_available(result) + + fixture_delegation.finalize( + self.work_id, + invocation, + result, + self.fence(claimed), + approved_commit=self.approved_commit, + policy_target=self.policy_target(), + host_target=self.delegation_host_target(), + observation_path=self.observation_path, + observation_not_before=self.live_at, + ended_at=self.live_at + timedelta(minutes=3), + now=self.live_at + timedelta(minutes=3), + ) + + checkout = self.mailbox_clone("installed-v2-requires-epic-read") + work, _ = _read_yaml( + checkout / f"work/{self.work_id}/work.md", + frontmatter=True, + label="work", + ) + receipt, body = _read_yaml( + checkout / f"work/{self.work_id}/receipts/{work['attempt_receipt_id']}.md", + frontmatter=True, + label="receipt", + ) + self.assertEqual(work["status"], "blocked") + self.assertEqual(receipt["outcome"], "blocked") + self.assertEqual(body, result["handoff"]["reason"]) + reconstruct_mailbox(checkout) + + def test_delegation_rejects_blocked_tracker_mutation_before_receipt(self) -> None: + claimed = self.claim() + delegation = self.delegation() + invocation = self.delegated_invocation(claimed) + checkpointed = self.delegated_checkpoint( + delegation, + invocation, + claimed, + action="repository.candidate.create", + token="token-1", + ) + result = self.blocked_result( + invocation, checkpointed, tracker_mode="manual", tracker_state="closed" + ) + before = git( + None, + "--git-dir", + str(self.mailbox_remote), + "rev-parse", + "main", + ).stdout.strip() + + with self.assertRaisesRegex(MailboxTransitionRejected, "tracker transition"): + delegation.finalize( + self.work_id, + invocation, + result, + self.fence(checkpointed), + approved_commit=self.approved_commit, + policy_target=self.policy_target(), + host_target=self.delegation_host_target(), + observation_path=self.observation_path, + observation_not_before=self.live_at, + ended_at=self.live_at + timedelta(minutes=3), + now=self.live_at + timedelta(minutes=3), + ) + + after = git( + None, + "--git-dir", + str(self.mailbox_remote), + "rev-parse", + "main", + ).stdout.strip() + self.assertEqual(after, before) + + def test_delegation_rejects_altered_invocation_at_finalization(self) -> None: + claimed = self.claim() + delegation = self.delegation() + invocation = self.delegated_invocation(claimed) + checkpointed = self.delegated_checkpoint( + delegation, + invocation, + claimed, + action="repository.candidate.create", + token="token-1", + ) + result = self.blocked_result(invocation, checkpointed) + altered = {**invocation, "validation": ["just test"]} + before = git( + None, + "--git-dir", + str(self.mailbox_remote), + "rev-parse", + "main", + ).stdout.strip() + + with self.assertRaisesRegex(DelegationError, "sealed digest"): + delegation.finalize( + self.work_id, + altered, + result, + self.fence(checkpointed), + approved_commit=self.approved_commit, + policy_target=self.policy_target(), + host_target=self.delegation_host_target(), + observation_path=self.observation_path, + observation_not_before=self.live_at, + ended_at=self.live_at + timedelta(minutes=3), + now=self.live_at + timedelta(minutes=3), + ) + + after = git( + None, + "--git-dir", + str(self.mailbox_remote), + "rev-parse", + "main", + ).stdout.strip() + self.assertEqual(after, before) + + def test_delegation_normalizes_unavailable_blocked_review(self) -> None: + claimed = self.claim() + delegation = self.delegation() + invocation = self.delegated_invocation(claimed) + candidate = { + "repository": invocation["repository"]["identity"], + "remote_url": invocation["repository"]["remote_url"], + "remote_ref": "refs/heads/scott/blocked-candidate", + "base_sha": invocation["repository"]["base_sha"], + "head_sha": HEAD, + } + created = self.delegated_checkpoint( + delegation, + invocation, + claimed, + action="repository.candidate.push", + token="token-1", + candidate=candidate, + ) + published = self.delegated_checkpoint( + delegation, + invocation, + created, + action="repository.candidate.push", + token="token-2", + phase="candidate_published", + candidate=candidate, + ) + reviews = [ + { + "name": "review-code-change", + "outcome": "unavailable", + "candidate_sha": HEAD, + "observed_at": fixtures.TIMESTAMP, + }, + ] + terminal_candidate = { + **candidate, + "publication": {"kind": "ordinary", "pull_requests": []}, + } + result = self.blocked_result( + invocation, + published, + reviews=reviews, + candidate=terminal_candidate, + authority_used=["repository.candidate.push"], + ) + delegation.finalize( + self.work_id, + invocation, + result, + self.fence(published), + approved_commit=self.approved_commit, + policy_target=self.policy_target(), + host_target=self.delegation_host_target(), + observation_path=self.observation_path, + observation_not_before=self.live_at, + ended_at=self.live_at + timedelta(minutes=3), + now=self.live_at + timedelta(minutes=3), + ) + + checkout = self.mailbox_clone("delegated-review-read") + work, _ = _read_yaml( + checkout / f"work/{self.work_id}/work.md", + frontmatter=True, + label="work", + ) + receipt, _ = _read_yaml( + checkout / f"work/{self.work_id}/receipts/{work['attempt_receipt_id']}.md", + frontmatter=True, + label="receipt", + ) + self.assertEqual( + [review["verdict"] for review in receipt["reviews"]], + ["blocked"], + ) + self.assertEqual(receipt["reviews"][0]["mechanism"], "review-code-change") + reconstruct_mailbox(checkout) + + def test_delegation_denies_stale_checkpoint_without_mailbox_mutation(self) -> None: + claimed = self.claim() + delegation = self.delegation() + invocation = self.delegated_invocation(claimed) + request = self.delegated_request(invocation, claimed, action="repository.candidate.create") + request["continuation_token"] = "stale-token" + + self.assert_checkpoint_denied_without_mutation(delegation, invocation, request) + + def test_delegation_denies_wrong_sequence_with_current_token_without_mutation(self) -> None: + claimed = self.claim() + delegation = self.delegation() + invocation = self.delegated_invocation(claimed) + request = self.delegated_request(invocation, claimed, action="repository.candidate.create") + request["sequence"] += 1 + + self.assert_checkpoint_denied_without_mutation(delegation, invocation, request) + + def test_delegation_denies_altered_invocation_without_mutation(self) -> None: + claimed = self.claim() + delegation = self.delegation() + invocation = self.delegated_invocation(claimed) + altered = {**invocation, "validation": ["just test"]} + request = self.delegated_request(altered, claimed, action="repository.candidate.create") + + response = self.assert_checkpoint_denied_without_mutation(delegation, altered, request) + self.assertIn("sealed digest", response["reason"]) + + def test_delegation_denies_foreign_checkpoint_candidate_without_mutation(self) -> None: + claimed = self.claim() + delegation = self.delegation() + invocation = self.delegated_invocation(claimed) + candidate = { + "repository": invocation["repository"]["identity"], + "remote_url": "https://github.com/foreign/project.git", + "remote_ref": "refs/heads/scott/foreign-candidate", + "base_sha": invocation["repository"]["base_sha"], + "head_sha": HEAD, + } + request = self.delegated_request( + invocation, + claimed, + action="repository.candidate.push", + candidate=candidate, + ) + + response = self.assert_checkpoint_denied_without_mutation(delegation, invocation, request) + self.assertIn("foreign", response["reason"]) + + @unittest.skipUnless( + (INSTALLED_TICKET_SKILL / "references/delegated-execution/validate.py").is_file(), + "exact installed Agent Scripts v2 bundle is unavailable", + ) + def test_checkpoint_command_services_one_request_with_installed_v2_bundle(self) -> None: + claimed = self.claim() + github_remote = "git@github.com:example/project-1.git" + git( + self.project_checkout, + "config", + f"url.{self.project_remote}.insteadOf", + github_remote, + ) + installed_policy_target = PolicyTarget( + checkout=self.project_checkout, + remote=github_remote, + canonical_ref="refs/heads/main", + path=".atelier/policy.yaml", + ) + delegation = DelegationCoordinator( + ClaimCoordinator(str(self.mailbox_remote), "main") + ) + invocation = delegation.prepare( + self.work_id, + self.fence(claimed), + approved_commit=self.approved_commit, + policy_target=installed_policy_target, + host_target=self.installed_host_target(), + observation_path=self.observation_path, + observation_not_before=self.live_at, + checkpoint_invocation_path=self.root / "installed-delegated-invocation.json", + observation_command=self.observation_command, + now=self.live_at + timedelta(seconds=30), + ) + request = self.delegated_request( + invocation, + claimed, + action="repository.candidate.create", + ) + + completed = subprocess.run( + invocation["checkpoint"]["command"], + input=json.dumps(request), + text=True, + capture_output=True, + check=False, + ) + + self.assertEqual(completed.returncode, 0, completed.stderr) + response = json.loads(completed.stdout) + self.assertEqual(response["decision"], "allow", response) + self.assertEqual(response["request_sequence"], 1) + checkout = self.mailbox_clone("checkpoint-command-read") + work, _ = _read_yaml( + checkout / f"work/{self.work_id}/work.md", + frontmatter=True, + label="work", + ) + self.assertEqual(work["claim"]["checkpoint"]["sequence"], 1) + + def test_delegation_denies_checkpoint_after_current_policy_tightens(self) -> None: + claimed = self.claim() + delegation = self.delegation() + invocation = self.delegated_invocation(claimed) + request = self.delegated_request( + invocation, + claimed, + action="repository.candidate.create", + ) + self.commit_policy_change(evidence=EVIDENCE) + + response = self.assert_checkpoint_denied_without_mutation(delegation, invocation, request) + + self.assertIn("stale", response["reason"]) + + def test_delegation_rejects_terminal_result_after_current_policy_tightens(self) -> None: + claimed = self.claim() + delegation = self.delegation() + invocation = self.delegated_invocation(claimed) + checkpointed = self.delegated_checkpoint( + delegation, + invocation, + claimed, + action="repository.candidate.create", + token="token-1", + ) + result = self.blocked_result(invocation, checkpointed) + self.commit_policy_change(evidence=EVIDENCE) + before = git( + None, + "--git-dir", + str(self.mailbox_remote), + "rev-parse", + "main", + ).stdout.strip() + + with self.assertRaisesRegex(MailboxTransitionRejected, "stale"): + delegation.finalize( + self.work_id, + invocation, + result, + self.fence(checkpointed), + approved_commit=self.approved_commit, + policy_target=self.policy_target(), + host_target=self.delegation_host_target(), + observation_path=self.observation_path, + observation_not_before=self.live_at, + ended_at=self.live_at + timedelta(minutes=3), + now=self.live_at + timedelta(minutes=3), + ) + + after = git( + None, + "--git-dir", + str(self.mailbox_remote), + "rev-parse", + "main", + ).stdout.strip() + self.assertEqual(after, before) + + def test_blocked_result_rejects_same_head_on_unauthorized_ref_without_prior(self) -> None: + self.assert_blocked_ref_substitution_rejected(with_prior_candidate=False) + + def test_blocked_result_rejects_same_head_on_unauthorized_ref_after_prior(self) -> None: + self.assert_blocked_ref_substitution_rejected(with_prior_candidate=True) + + def test_blocked_result_recovers_unacknowledged_pushed_candidate(self) -> None: + claimed = self.claim() + delegation = self.delegation() + invocation = self.delegated_invocation(claimed) + candidate = { + "repository": invocation["repository"]["identity"], + "remote_url": invocation["repository"]["remote_url"], + "remote_ref": "refs/heads/scott/recover-pushed-candidate", + "base_sha": invocation["repository"]["base_sha"], + "head_sha": HEAD, + } + pushed = self.delegated_checkpoint( + delegation, + invocation, + claimed, + action="repository.candidate.push", + token="token-1", + candidate=candidate, + ) + terminal_candidate = { + **candidate, + "publication": {"kind": "ordinary", "pull_requests": []}, + } + result = self.blocked_result( + invocation, + pushed, + candidate=terminal_candidate, + authority_used=["repository.candidate.push"], + ) + + delegation.finalize( + self.work_id, + invocation, + result, + self.fence(pushed), + approved_commit=self.approved_commit, + policy_target=self.policy_target(), + host_target=self.delegation_host_target(), + observation_path=self.observation_path, + observation_not_before=self.live_at, + ended_at=self.live_at + timedelta(minutes=3), + now=self.live_at + timedelta(minutes=3), + ) + + checkout = self.mailbox_clone("recovered-pushed-candidate-read") + work, _ = _read_yaml( + checkout / f"work/{self.work_id}/work.md", + frontmatter=True, + label="work", + ) + receipt, _ = _read_yaml( + checkout / f"work/{self.work_id}/receipts/{work['attempt_receipt_id']}.md", + frontmatter=True, + label="receipt", + ) + self.assertEqual(work["claim"]["candidate"]["head_revision"], HEAD) + self.assertEqual(receipt["candidate"]["head_revision"], HEAD) + reconstruct_mailbox(checkout) + + def test_blocked_result_replaces_older_candidate_with_latest_verified_push(self) -> None: + self.coordinator = ClaimCoordinator( + str(self.mailbox_remote), + "main", + candidate_verifier=lambda candidate: True, + capability_verifier=lambda target: True, + policy_remote_verifier=self.policy_remote_matches, + ) + claimed = self.claim() + delegation = self.delegation() + invocation = self.delegated_invocation(claimed) + old_candidate = { + "repository": invocation["repository"]["identity"], + "remote_url": invocation["repository"]["remote_url"], + "remote_ref": "refs/heads/scott/older-candidate", + "base_sha": invocation["repository"]["base_sha"], + "head_sha": HEAD, + } + old_push = self.delegated_checkpoint( + delegation, + invocation, + claimed, + action="repository.candidate.push", + token="token-1", + candidate=old_candidate, + ) + old_published = self.delegated_checkpoint( + delegation, + invocation, + old_push, + action="repository.candidate.push", + token="token-2", + phase="candidate_published", + candidate=old_candidate, + ) + latest_head = "c" * 40 + latest_candidate = { + **old_candidate, + "remote_ref": "refs/heads/scott/latest-candidate", + "head_sha": latest_head, + } + latest_push = self.delegated_checkpoint( + delegation, + invocation, + old_published, + action="repository.candidate.push", + token="token-3", + candidate=latest_candidate, + ) + result = self.blocked_result( + invocation, + latest_push, + candidate={ + **latest_candidate, + "publication": {"kind": "ordinary", "pull_requests": []}, + }, + authority_used=["repository.candidate.push"], + ) + + delegation.finalize( + self.work_id, + invocation, + result, + self.fence(latest_push), + approved_commit=self.approved_commit, + policy_target=self.policy_target(), + host_target=self.delegation_host_target(), + observation_path=self.observation_path, + observation_not_before=self.live_at, + ended_at=self.live_at + timedelta(minutes=3), + now=self.live_at + timedelta(minutes=3), + ) + + checkout = self.mailbox_clone("latest-pushed-candidate-read") + work, _ = _read_yaml( + checkout / f"work/{self.work_id}/work.md", + frontmatter=True, + label="work", + ) + self.assertEqual(work["claim"]["candidate"]["head_revision"], latest_head) + reconstruct_mailbox(checkout) + + def test_delegation_delivers_one_exact_ready_pull_request(self) -> None: + claimed = self.claim() + self.coordinator = ClaimCoordinator( + str(self.mailbox_remote), + "main", + candidate_verifier=lambda value: _candidate_remote_reachable( + {**value, "remote_url": str(self.project_remote)} + ), + capability_verifier=lambda target: True, + policy_remote_verifier=self.policy_remote_matches, + ) + delegation = self.delegation() + invocation = self.delegated_invocation(claimed) + candidate_head = git(self.project_checkout, "rev-parse", "HEAD").stdout.strip() + candidate_ref = "refs/heads/scott/delegated-candidate" + git(self.project_checkout, "push", "origin", f"{candidate_head}:{candidate_ref}") + candidate = { + "repository": self.repository, + "remote_url": invocation["repository"]["remote_url"], + "remote_ref": candidate_ref, + "base_sha": invocation["repository"]["base_sha"], + "head_sha": candidate_head, + } + push = self.delegated_checkpoint( + delegation, + invocation, + claimed, + action="repository.candidate.push", + token="token-1", + candidate=candidate, + ) + published = self.delegated_checkpoint( + delegation, + invocation, + push, + action="repository.candidate.push", + token="token-2", + phase="candidate_published", + candidate=candidate, + ) + pull_request = self.delegated_checkpoint( + delegation, + invocation, + published, + action="pull_request.create", + token="token-3", + candidate=candidate, + ) + pull_request_url = "https://github.com/example/project-1/pull/900" + live = observation(with_pull_request=True) + live["observed_at"] = ( + (self.live_at + timedelta(minutes=3)).isoformat().replace("+00:00", "Z") + ) + live["pull_request"]["url"] = pull_request_url + live["pull_request"]["base"].update( + ref=invocation["repository"]["base_ref"], + sha=invocation["repository"]["base_sha"], + ) + live["pull_request"]["head"].update(ref=candidate_ref, sha=candidate_head) + write_json(self.observation_path, live) + result = { + "schema": RESULT_SCHEMA, + "capability": CAPABILITY, + "invocation_id": invocation["invocation_id"], + "terminal_state": "ready_pr", + "ticket": invocation["ticket"], + "repository": { + "identity": invocation["repository"]["identity"], + "base_ref": invocation["repository"]["base_ref"], + "base_sha": invocation["repository"]["base_sha"], + }, + "tracker_transition": { + "provider": "github", + "ticket_id": invocation["ticket"]["id"], + "mode": "none", + "state": "open", + "observed_at": fixtures.TIMESTAMP, + }, + "implementation_state": "published", + "candidate": { + **candidate, + "publication": { + "kind": "ordinary", + "pull_requests": [ + { + "id": "900", + "url": pull_request_url, + "base_ref": invocation["repository"]["base_ref"], + "base_sha": invocation["repository"]["base_sha"], + "head_ref": candidate_ref, + "head_sha": candidate_head, + "state": "open", + } + ], + }, + }, + "handoff": {"transferable": True, "reason": None}, + "checkpoint": { + "last_sequence": pull_request.sequence, + "continuation_token": pull_request.continuation_token, + }, + "validation": [ + { + "name": command, + "outcome": "passed", + "candidate_sha": candidate_head, + "observed_at": fixtures.TIMESTAMP, + } + for command in invocation["validation"] + ], + "reviews": [ + { + "name": "review-code-change", + "outcome": "passed", + "candidate_sha": candidate_head, + "observed_at": fixtures.TIMESTAMP, + } + ], + "feedback": { + "unresolved_material_count": 0, + "candidate_sha": candidate_head, + "observed_at": fixtures.TIMESTAMP, + }, + "authority_used": ["repository.candidate.push", "pull_request.create"], + "acceptance_evidence": self.acceptance_records(invocation, candidate_head), + "unresolved_obligations": [], + "blocking_reason": None, + "next_action": "Atelier validates and presents the candidate for operator acceptance.", + } + def finalize(reported_result: dict[str, Any]) -> None: + delegation.finalize( + self.work_id, + invocation, + reported_result, + self.fence(pull_request), + approved_commit=self.approved_commit, + policy_target=self.policy_target(), + host_target=self.delegation_host_target(), + observation_path=self.observation_path, + observation_not_before=self.live_at + timedelta(minutes=3), + ended_at=self.live_at + timedelta(minutes=4), + now=self.live_at + timedelta(minutes=4), + ) + + before_rejection = git( + None, + "--git-dir", + str(self.mailbox_remote), + "rev-parse", + "main", + ).stdout.strip() + for pr_field, live_section, live_field, substituted in ( + ("head_ref", "head", "ref", "refs/heads/substituted"), + ("head_sha", "head", "sha", "c" * 40), + ("base_sha", "base", "sha", "d" * 40), + ): + substituted_result = json.loads(json.dumps(result)) + substituted_live = json.loads(json.dumps(live)) + reported_pr = substituted_result["candidate"]["publication"]["pull_requests"][0] + reported_pr[pr_field] = substituted + substituted_live["pull_request"][live_section][live_field] = substituted + write_json(self.observation_path, substituted_live) + with self.assertRaisesRegex( + MailboxTransitionRejected, + "terminal pull request does not identify the acknowledged candidate", + ): + finalize(substituted_result) + self.assertEqual( + git( + None, + "--git-dir", + str(self.mailbox_remote), + "rev-parse", + "main", + ).stdout.strip(), + before_rejection, + ) + foreign_head = json.loads(json.dumps(live)) + foreign_head["pull_request"]["head"]["repository"] = "foreign/fork" + write_json(self.observation_path, foreign_head) + with self.assertRaisesRegex( + MailboxTransitionRejected, + "live pull request identity is not the terminal candidate", + ): + finalize(result) + self.assertEqual( + git( + None, + "--git-dir", + str(self.mailbox_remote), + "rev-parse", + "main", + ).stdout.strip(), + before_rejection, + ) + write_json(self.observation_path, live) + + foreign_review = json.loads(json.dumps(result)) + foreign_review["reviews"][0]["name"] = "generic-review" + self.assert_installed_result_valid_if_available(foreign_review) + with self.assertRaisesRegex( + MailboxTransitionRejected, + "review evidence uses a mechanism Atelier cannot represent", + ): + finalize(foreign_review) + self.assertEqual( + git( + None, + "--git-dir", + str(self.mailbox_remote), + "rev-parse", + "main", + ).stdout.strip(), + before_rejection, + ) + + finalize(result) + checkout = self.mailbox_clone("delegated-delivered-read") + work, _ = _read_yaml( + checkout / f"work/{self.work_id}/work.md", + frontmatter=True, + label="work", + ) + receipt, _ = _read_yaml( + checkout / f"work/{self.work_id}/receipts/{work['delivery_receipt_id']}.md", + frontmatter=True, + label="receipt", + ) + self.assertEqual(work["status"], "delivered") + self.assertEqual(work["claim"]["candidate"]["pull_request"], pull_request_url) + self.assertEqual(receipt["candidate"], work["claim"]["candidate"]) + reconstruct_mailbox(checkout) + + def fresh_observation_digest(self) -> str: + checkout = self.mailbox_clone("delegated-digest-read") + work, _ = _read_yaml( + checkout / f"work/{self.work_id}/work.md", + frontmatter=True, + label="work", + ) + return work["claim"]["ticket_observation_digest"] if work["claim"] else "" + def test_cli_generates_a_strict_durable_identifier(self) -> None: completed = subprocess.run( ["python3", str(CLAIMING_SCRIPT), "new-id", "clm"], diff --git a/contract_tests/test_git_mailbox.py b/contract_tests/test_git_mailbox.py index 32e114f3..2f1516dd 100644 --- a/contract_tests/test_git_mailbox.py +++ b/contract_tests/test_git_mailbox.py @@ -188,9 +188,7 @@ def setUp(self) -> None: git(None, "clone", str(self.remote), str(seed)) mailbox = fixtures.MailboxFixture(seed) self.work_id = mailbox.add_work(1, "approved") - self.repository = mailbox.projects[mailbox.works[self.work_id]["project_id"]][ - "repository" - ] + self.repository = mailbox.projects[mailbox.works[self.work_id]["project_id"]]["repository"] git(seed, "add", "-A") git( seed, @@ -769,6 +767,7 @@ def authorize(context: TransitionContext) -> TransitionPlan: "action": "repository.candidate.create", "proposed_effect_digest": fixtures.DIGEST, "candidate_head": None, + "candidate_remote_ref": None, "acknowledged_candidate_head": None, "recorded_at": fixtures.TIMESTAMP, } @@ -870,9 +869,7 @@ def take_over(context: TransitionContext) -> TransitionPlan: fresh = self.root / "fresh-handoff" git(None, "clone", str(self.remote), str(fresh)) current = read_markdown(fresh / work_path) - released = read_markdown( - fresh / f"work/{self.work_id}/receipts/{receipt_id}.md" - ) + released = read_markdown(fresh / f"work/{self.work_id}/receipts/{receipt_id}.md") self.assertEqual(released["candidate"], candidate) self.assertEqual(current["attempt_receipt_id"], receipt_id) self.assertEqual(current["claim"]["candidate"], candidate) @@ -1115,9 +1112,7 @@ def plan(context: TransitionContext) -> TransitionPlan: hook = context.checkout / ".git" / "hooks" / "pre-commit" hook.parent.mkdir(parents=True) hook.write_text( - "#!/bin/sh\n" - f"printf 'corrupted\\n' > {path}\n" - f"git add -- {path}\n", + f"#!/bin/sh\nprintf 'corrupted\\n' > {path}\ngit add -- {path}\n", encoding="utf-8", ) hook.chmod(0o700) @@ -1304,9 +1299,7 @@ def mutate_checkpoint( checkpoint["authorizations"] = checkpoint["authorizations"][:-1] checkpoint["sequence"] -= 1 else: - checkpoint["authorizations"][-1]["proposed_effect_digest"] = ( - "sha256:" + ("f" * 64) - ) + checkpoint["authorizations"][-1]["proposed_effect_digest"] = "sha256:" + ("f" * 64) return TransitionPlan( f"{mutation} checkpoint ledger", (FileChange(path, markdown(work)),), @@ -1365,6 +1358,35 @@ def mutate_claim( ) self.assertEqual(self.remote_head(), before) + def test_delegated_invocation_digest_can_be_sealed_once(self) -> None: + self._claim(with_candidate=False) + path = f"work/{self.work_id}/work.md" + + def seal(context: TransitionContext, digest: str) -> TransitionPlan: + work = read_markdown(context.checkout / path) + work["claim"]["invocation_digest"] = digest + return TransitionPlan( + "seal delegated invocation", + (FileChange(path, markdown(work)),), + ) + + self.writer().publish( + "seal delegated invocation", + revalidate=lambda context: None, + plan=lambda context: seal(context, fixtures.DIGEST), + ) + sealed_head = self.remote_head() + + with self.assertRaisesRegex( + MailboxTransitionRejected, "delegated invocation binding is immutable" + ): + self.writer().publish( + "reseal delegated invocation", + revalidate=lambda context: None, + plan=lambda context: seal(context, "sha256:" + ("f" * 64)), + ) + self.assertEqual(self.remote_head(), sealed_head) + def test_takeover_must_preserve_verified_candidate(self) -> None: original_claim = self._claim(with_candidate=True) self.assertIsNotNone(original_claim["candidate"]) @@ -1459,9 +1481,7 @@ def test_claim_and_run_identities_are_fenced_across_work_history(self) -> None: second_path = f"work/{second_work_id}/work.md" def add_second_work(context: TransitionContext) -> TransitionPlan: - current = read_markdown( - context.checkout / f"work/{self.work_id}/work.md" - ) + current = read_markdown(context.checkout / f"work/{self.work_id}/work.md") second = fixtures.base_work( second_work_id, current["project_id"], @@ -1540,17 +1560,15 @@ def test_same_claim_candidate_rebinding_requires_publication_checkpoint(self) -> def rebind_candidate(context: TransitionContext) -> TransitionPlan: work = read_markdown(context.checkout / path) - work["claim"]["candidate"]["remote_ref"] = ( - "refs/heads/scott/rebound-candidate" - ) + work["claim"]["candidate"]["remote_ref"] = "refs/heads/scott/rebound-candidate" return TransitionPlan( "rebind candidate without checkpoint", (FileChange(path, markdown(work)),), ) with self.assertRaisesRegex( - MailboxTransitionRejected, - "candidate changes require one publication checkpoint", + (MailboxTransitionRejected, MailboxValidationError), + "candidate changes require one publication checkpoint|candidate-acknowledgement", ): self.writer().publish( "rebind candidate without checkpoint", @@ -1616,9 +1634,7 @@ def plan(context: TransitionContext) -> TransitionPlan: for authorization in published_claim["checkpoint"]["authorizations"]: claim_value["checkpoint"]["authorizations"].append(copy.deepcopy(authorization)) claim_value["checkpoint"]["sequence"] = authorization["sequence"] - claim_value["checkpoint"]["continuation_token"] = ( - f"token-1-{authorization['sequence']}" - ) + claim_value["checkpoint"]["continuation_token"] = f"token-1-{authorization['sequence']}" if authorization["phase"] == "candidate_published": claim_value["candidate"] = copy.deepcopy(published_claim["candidate"]) @@ -1630,9 +1646,7 @@ def checkpoint( checkpoint_claim: dict[str, Any] = checkpoint_claim, checkpoint_number: int = checkpoint_number, ) -> TransitionPlan: - work = read_markdown( - context.checkout / f"work/{self.work_id}/work.md" - ) + work = read_markdown(context.checkout / f"work/{self.work_id}/work.md") work["claim"] = copy.deepcopy(checkpoint_claim) return TransitionPlan( f"advance checkpoint {checkpoint_number}", diff --git a/contract_tests/test_host_boundary.py b/contract_tests/test_host_boundary.py index 3bc0e7cf..33dc8c73 100644 --- a/contract_tests/test_host_boundary.py +++ b/contract_tests/test_host_boundary.py @@ -220,20 +220,24 @@ def setUp(self) -> None: "import json, sys\n" "value = json.load(open(sys.argv[2], encoding='utf-8'))\n" "raise SystemExit(0 if value.get('id') == " - "'agent-scripts.implement-ticket/delegated-execution/v1' else 1)\n", + "'agent-scripts.implement-ticket/delegated-execution/v2' else 1)\n", encoding="utf-8", ) manifest = { - "schema": "agent-scripts.implement-ticket/capability-manifest/v1", - "id": "agent-scripts.implement-ticket/delegated-execution/v1", + "schema": "agent-scripts.implement-ticket/capability-manifest/v2", + "id": "agent-scripts.implement-ticket/delegated-execution/v2", "contract": "CONTRACT.md", "invocation_schema": "invocation.schema.json", "checkpoint_request_schema": "checkpoint-request.schema.json", "checkpoint_response_schema": "checkpoint-response.schema.json", "result_schema": "result.schema.json", "validator": "validate.py", - "terminal_states": ["ready_pr", "blocked", "requires_epic"], - "checkpoint_phases": ["pre_external_mutation", "candidate_published"], + "terminal_states": ["ready_pr", "ready_prs", "merged", "blocked", "requires_epic"], + "checkpoint_phases": [ + "pre_external_mutation", + "candidate_published", + "deployment_observed", + ], } self.manifest_path = capability_root / "capability.json" write_json(self.manifest_path, manifest) @@ -246,7 +250,7 @@ def setUp(self) -> None: descriptor = { "schema": "atelier.host-capability/v1", "reference_host": "codex", - "delegated_capability": ("agent-scripts.implement-ticket/delegated-execution/v1"), + "delegated_capability": ("agent-scripts.implement-ticket/delegated-execution/v2"), "delegated_authority_actions": AUTHORITY_ACTIONS, "accepted_terminal_states": ["ready_pr", "blocked", "requires_epic"], "delegated_skill": { @@ -320,7 +324,7 @@ def test_host_capability_is_published(self) -> None: self.assertEqual(payload["reference_host"], "codex") self.assertEqual( payload["delegated_capability"], - "agent-scripts.implement-ticket/delegated-execution/v1", + "agent-scripts.implement-ticket/delegated-execution/v2", ) self.assertEqual(payload["native_state_access"], "read-only") self.assertFalse(payload["fallback_to_copied_workflows"]) diff --git a/contract_tests/test_mailbox.py b/contract_tests/test_mailbox.py index 678ce4de..6f7c1b46 100644 --- a/contract_tests/test_mailbox.py +++ b/contract_tests/test_mailbox.py @@ -153,9 +153,7 @@ def candidate(repository: str, number: int) -> dict[str, Any]: "remote_ref": f"refs/heads/scott/work-{number}", "base_revision": SHA_A, "head_revision": SHA_B, - "pull_request": ( - f"https://github.com/{repository.removeprefix('github:')}/pull/{number}" - ), + "pull_request": (f"https://github.com/{repository.removeprefix('github:')}/pull/{number}"), "workspace_id": None, "published_at": TIMESTAMP, } @@ -172,6 +170,7 @@ def claim(repository: str, number: int, *, with_candidate: bool) -> dict[str, An "action": "repository.candidate.push", "proposed_effect_digest": DIGEST, "candidate_head": SHA_B, + "candidate_remote_ref": f"refs/heads/scott/work-{number}", "acknowledged_candidate_head": None, "recorded_at": TIMESTAMP, }, @@ -182,6 +181,7 @@ def claim(repository: str, number: int, *, with_candidate: bool) -> dict[str, An "action": "repository.candidate.push", "proposed_effect_digest": DIGEST, "candidate_head": SHA_B, + "candidate_remote_ref": f"refs/heads/scott/work-{number}", "acknowledged_candidate_head": SHA_B, "recorded_at": TIMESTAMP, }, @@ -192,6 +192,7 @@ def claim(repository: str, number: int, *, with_candidate: bool) -> dict[str, An "action": "pull_request.create", "proposed_effect_digest": DIGEST, "candidate_head": SHA_B, + "candidate_remote_ref": f"refs/heads/scott/work-{number}", "acknowledged_candidate_head": None, "recorded_at": TIMESTAMP, }, @@ -206,6 +207,7 @@ def claim(repository: str, number: int, *, with_candidate: bool) -> dict[str, An "approved_commit": SHA_A, "policy_commit": SHA_A, "ticket_observation_digest": DIGEST, + "invocation_digest": None, "claimed_at": TIMESTAMP, "host": "codex", "checkpoint": { @@ -370,9 +372,7 @@ def add_work( if status != "draft": work["approval"] = approval(repository) if status in {"active", "blocked", "delivered"}: - work["claim"] = claim( - repository, number, with_candidate=status == "delivered" - ) + work["claim"] = claim(repository, number, with_candidate=status == "delivered") self.messages[work_id] = {} self.receipts[work_id] = {} if status == "blocked": @@ -515,7 +515,7 @@ def test_project_policy_is_strict_and_read_only(self) -> None: "material_fields": ["body", "state", "relationships"], }, "execution": { - "capability": "agent-scripts.implement-ticket/delegated-execution/v1", + "capability": "agent-scripts.implement-ticket/delegated-execution/v2", "delivery_outcome": "ready_pr", "parallel_assignments": False, }, @@ -530,9 +530,7 @@ def test_project_policy_is_strict_and_read_only(self) -> None: self.assertEqual(path.read_bytes(), before) policy["authority"]["allow"].append("merge") write_yaml(path, policy) - with self.assertRaisesRegex( - MAILBOX.MailboxValidationError, "unsupported value" - ): + with self.assertRaisesRegex(MAILBOX.MailboxValidationError, "unsupported value"): MAILBOX.validate_project_policy(path) policy["authority"]["allow"] = list(AUTHORITY) policy["execution"]["parallel_assignments"] = 0 @@ -555,26 +553,18 @@ def test_project_policy_is_strict_and_read_only(self) -> None: policy["mailbox"]["remote"] = "https://user:secret@github.com/example/mailbox.git" write_yaml(path, policy) before = path.read_bytes() - with self.assertRaisesRegex( - MAILBOX.MailboxValidationError, "remote-credentials" - ): + with self.assertRaisesRegex(MAILBOX.MailboxValidationError, "remote-credentials"): MAILBOX.validate_project_policy(path) self.assertEqual(path.read_bytes(), before) policy["mailbox"]["remote"] = "git@github.com:example/mailbox.git" policy["repository"]["identity"] = "github:./.." write_yaml(path, policy) - with self.assertRaisesRegex( - MAILBOX.MailboxValidationError, "repository-identity" - ): + with self.assertRaisesRegex(MAILBOX.MailboxValidationError, "repository-identity"): MAILBOX.validate_project_policy(path) policy["repository"]["identity"] = repository - policy["mailbox"]["remote"] = ( - "https://github.com/example/mailbox.git?access_token=secret" - ) + policy["mailbox"]["remote"] = "https://github.com/example/mailbox.git?access_token=secret" write_yaml(path, policy) - with self.assertRaisesRegex( - MAILBOX.MailboxValidationError, "remote-credentials" - ): + with self.assertRaisesRegex(MAILBOX.MailboxValidationError, "remote-credentials"): MAILBOX.validate_project_policy(path) def test_fresh_clones_reconstruct_all_views_identically(self) -> None: @@ -597,9 +587,7 @@ def test_fresh_clones_reconstruct_all_views_identically(self) -> None: active = self.fixture.add_work(3, "active") blocked = self.fixture.add_work(4, "blocked") delivered = self.fixture.add_work(5, "delivered") - readiness = { - ready: {"policy": True, "ticket": True, "capability": True} - } + readiness = {ready: {"policy": True, "ticket": True, "capability": True}} git_run(self.root, "init", "-b", "main") git_run(self.root, "config", "user.name", "Atelier Contract") git_run(self.root, "config", "user.email", "atelier@example.invalid") @@ -871,9 +859,9 @@ def test_message_authorship_and_candidate_observations_are_bound(self) -> None: self.fixture = MailboxFixture(self.root) delivered = self.fixture.add_work(2, "delivered") receipt_id = self.fixture.works[delivered]["delivery_receipt_id"] - self.fixture.receipts[delivered][receipt_id]["validation"][0][ - "candidate_revision" - ] = "d" * 40 + self.fixture.receipts[delivered][receipt_id]["validation"][0]["candidate_revision"] = ( + "d" * 40 + ) self.fixture.write_work(delivered) self.assert_invalid("validation-candidate") @@ -921,6 +909,7 @@ def test_checkpoint_ledger_tracks_acknowledged_candidate_history(self) -> None: "action": "pull_request.create", "proposed_effect_digest": DIGEST, "candidate_head": "d" * 40, + "candidate_remote_ref": work["claim"]["candidate"]["remote_ref"], "acknowledged_candidate_head": None, "recorded_at": TIMESTAMP, } @@ -974,6 +963,7 @@ def test_append_only_candidate_publication_history_reconstructs(self) -> None: "action": "repository.candidate.push", "proposed_effect_digest": DIGEST, "candidate_head": next_head, + "candidate_remote_ref": work["claim"]["candidate"]["remote_ref"], "acknowledged_candidate_head": None, "recorded_at": TIMESTAMP, }, @@ -984,6 +974,7 @@ def test_append_only_candidate_publication_history_reconstructs(self) -> None: "action": "repository.candidate.push", "proposed_effect_digest": DIGEST, "candidate_head": next_head, + "candidate_remote_ref": work["claim"]["candidate"]["remote_ref"], "acknowledged_candidate_head": next_head, "recorded_at": TIMESTAMP, }, @@ -1153,9 +1144,7 @@ def test_historical_blocked_receipt_retains_mutation_ownership(self) -> None: work["claim"] = None work["blocking_message_id"] = None work["attempt_receipt_id"] = None - self.fixture.receipts[work_id][receipt_id]["mutation_ownership"] = ( - "relinquished" - ) + self.fixture.receipts[work_id][receipt_id]["mutation_ownership"] = "relinquished" self.fixture.write_work(work_id) self.assert_invalid("blocked-ownership") @@ -1163,9 +1152,7 @@ def test_historical_blocked_receipt_retains_mutation_ownership(self) -> None: def test_delivered_receipt_rejects_failed_validation(self) -> None: work_id = self.fixture.add_work(1, "delivered") receipt_id = self.fixture.works[work_id]["delivery_receipt_id"] - self.fixture.receipts[work_id][receipt_id]["validation"][0]["outcome"] = ( - "failed" - ) + self.fixture.receipts[work_id][receipt_id]["validation"][0]["outcome"] = "failed" self.fixture.write_work(work_id) self.assert_invalid("delivered-receipt") @@ -1273,9 +1260,7 @@ def test_timestamps_and_candidate_remote_urls_fail_closed(self) -> None: receipt_id = work["delivery_receipt_id"] invalid_timestamp = "2026-07-27 12:00:00+00:00" work["claim"]["candidate"]["published_at"] = invalid_timestamp - self.fixture.receipts[work_id][receipt_id]["candidate"][ - "published_at" - ] = invalid_timestamp + self.fixture.receipts[work_id][receipt_id]["candidate"]["published_at"] = invalid_timestamp self.fixture.write_work(work_id) self.assert_invalid("schema-one-of") @@ -1287,9 +1272,7 @@ def test_timestamps_and_candidate_remote_urls_fail_closed(self) -> None: receipt_id = work["delivery_receipt_id"] mismatched_remote = "git@github.com:another/project.git" work["claim"]["candidate"]["remote_url"] = mismatched_remote - self.fixture.receipts[work_id][receipt_id]["candidate"][ - "remote_url" - ] = mismatched_remote + self.fixture.receipts[work_id][receipt_id]["candidate"]["remote_url"] = mismatched_remote self.fixture.write_work(work_id) self.assert_invalid("candidate-remote-url") @@ -1316,9 +1299,9 @@ def test_workspace_id_is_portable_and_not_a_machine_local_path(self) -> None: work = self.fixture.works[work_id] receipt_id = work["delivery_receipt_id"] work["claim"]["candidate"]["workspace_id"] = machine_path - self.fixture.receipts[work_id][receipt_id]["candidate"][ - "workspace_id" - ] = machine_path + self.fixture.receipts[work_id][receipt_id]["candidate"]["workspace_id"] = ( + machine_path + ) self.fixture.write_work(work_id) self.assert_invalid("schema-one-of") @@ -1354,18 +1337,14 @@ def test_github_repository_case_aliases_share_one_identity(self) -> None: second_project["policy"]["repository"] = repository_alias second_work = self.fixture.works[second] second_work["approval"]["policy"]["repository"] = repository_alias - second_work["native_ticket"]["url"] = ( - "https://github.com/EXAMPLE/PROJECT-1/issues/2" - ) + second_work["native_ticket"]["url"] = "https://github.com/EXAMPLE/PROJECT-1/issues/2" self.fixture.write_project(second_project_id) self.fixture.write_work(second) with self.assertRaises(MAILBOX.MailboxValidationError) as caught: MAILBOX.reconstruct_mailbox(self.root) codes = {item.code for item in caught.exception.diagnostics} - self.assertTrue( - {"duplicate-repository", "parallel-assignments"}.issubset(codes) - ) + self.assertTrue({"duplicate-repository", "parallel-assignments"}.issubset(codes)) def test_duplicate_yaml_keys_and_unexpected_layout_fail_closed(self) -> None: manifest = self.root / "atelier.yaml" @@ -1451,9 +1430,7 @@ def test_mailbox_collection_file_has_a_relative_layout_diagnostic(self) -> None: with self.assertRaises(MAILBOX.MailboxValidationError) as caught: MAILBOX.reconstruct_mailbox(self.root) - diagnostic = next( - item for item in caught.exception.diagnostics if item.code == "layout" - ) + diagnostic = next(item for item in caught.exception.diagnostics if item.code == "layout") self.assertEqual(diagnostic.path, "projects") def test_invalid_utf8_document_has_a_relative_encoding_diagnostic(self) -> None: @@ -1461,9 +1438,7 @@ def test_invalid_utf8_document_has_a_relative_encoding_diagnostic(self) -> None: with self.assertRaises(MAILBOX.MailboxValidationError) as caught: MAILBOX.reconstruct_mailbox(self.root) - diagnostic = next( - item for item in caught.exception.diagnostics if item.code == "encoding" - ) + diagnostic = next(item for item in caught.exception.diagnostics if item.code == "encoding") self.assertEqual(diagnostic.path, "atelier.yaml") def test_malformed_single_quoted_yaml_fails_closed(self) -> None: @@ -1502,9 +1477,7 @@ def test_readiness_input_rejects_unknown_work_and_gates(self) -> None: readiness_path = Path(self.temporary.name) / "readiness.json" readiness_path.write_bytes(b"\xff") - with self.assertRaisesRegex( - MAILBOX.MailboxValidationError, "readiness-encoding" - ): + with self.assertRaisesRegex(MAILBOX.MailboxValidationError, "readiness-encoding"): MAILBOX._read_readiness(readiness_path) diff --git a/contract_tests/test_planning.py b/contract_tests/test_planning.py index 3d711dda..f4e81419 100644 --- a/contract_tests/test_planning.py +++ b/contract_tests/test_planning.py @@ -218,7 +218,7 @@ def write_policy( "material_fields": ["body", "state", "relationships"], }, "execution": { - "capability": "agent-scripts.implement-ticket/delegated-execution/v1", + "capability": "agent-scripts.implement-ticket/delegated-execution/v2", "delivery_outcome": "ready_pr", "parallel_assignments": False, }, @@ -285,10 +285,7 @@ def mailbox_clone(self) -> Path: def initiative_digest(self) -> str: checkout = self.mailbox_clone() - content = ( - checkout - / f"initiatives/{self.initiative_id}/initiative.md" - ).read_bytes() + content = (checkout / f"initiatives/{self.initiative_id}/initiative.md").read_bytes() return f"sha256:{hashlib.sha256(content).hexdigest()}" def create(self) -> None: @@ -783,10 +780,9 @@ def test_shared_initiative_revision_rejects_stale_digest(self) -> None: ) checkout = self.mailbox_clone() - initiative = ( - checkout - / f"initiatives/{self.initiative_id}/initiative.md" - ).read_text(encoding="utf-8") + initiative = (checkout / f"initiatives/{self.initiative_id}/initiative.md").read_text( + encoding="utf-8" + ) self.assertIn("The newer shared outcome.", initiative) self.assertNotIn("The stale shared outcome.", initiative) diff --git a/docs/atelier-skill-design.md b/docs/atelier-skill-design.md index d64007e7..a5af5ce0 100644 --- a/docs/atelier-skill-design.md +++ b/docs/atelier-skill-design.md @@ -198,7 +198,7 @@ prerequisites preserve the platform/application boundary more honestly. ### Agent Scripts capability contract Before claiming work, Atelier must establish that the host can invoke a -compatible `agent-scripts.implement-ticket/delegated-execution/v1` capability + compatible `agent-scripts.implement-ticket/delegated-execution/v2` capability and validate its `capability.json` discovery manifest plus versioned invocation, checkpoint, and result schemas. Finding a skill with the expected name is not sufficient. diff --git a/docs/git-mailbox-contract.md b/docs/git-mailbox-contract.md index 4fe1fb9e..fc08fec4 100644 --- a/docs/git-mailbox-contract.md +++ b/docs/git-mailbox-contract.md @@ -416,6 +416,7 @@ claim: approved_commit: 0123456789abcdef0123456789abcdef01234567 policy_commit: 0123456789abcdef0123456789abcdef01234567 ticket_observation_digest: sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef + invocation_digest: null claimed_at: 2026-07-25T12:05:00Z host: codex checkpoint: @@ -425,6 +426,11 @@ claim: candidate: null ``` +Delegation preparation changes `invocation_digest` exactly once from `null` to +the SHA-256 digest of the complete canonical delegated invocation. The digest is +immutable thereafter and every checkpoint and terminal result must reproduce +the sealed invocation before Atelier may consume authority or record a receipt. + Every entry in `checkpoint.authorizations` has this complete shape: ```yaml @@ -434,12 +440,13 @@ phase: pre_external_mutation action: repository.candidate.create proposed_effect_digest: sha256:0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef candidate_head: null +candidate_remote_ref: null acknowledged_candidate_head: null recorded_at: 2026-07-25T12:06:00Z ``` `phase` is `pre_external_mutation` or `candidate_published`; `action` uses the -Agent Scripts v1 vocabulary. SHA fields are exact candidate revisions or `null` +Agent Scripts v2 vocabulary. SHA fields are exact candidate revisions or `null` when the action has no candidate. All other fields are required. Entries are in strictly increasing sequence order and are append-only. When transferable implementation state first exists, the worker updates the claim only after @@ -464,9 +471,10 @@ reachable from `remote_ref` on the verified declared remote. A local-only SHA, an unverified push, or a mutable branch name without the exact remote ref and head is not a candidate. -After every candidate push, the worker verifies remote reachability, publishes -the new exact head in the claim, and verifies that mailbox transition before -relying on that candidate for review, delivery, or another external mutation. +After every candidate push, the worker normally verifies remote reachability, +publishes the new exact head in the claim, and verifies that mailbox transition +before relying on that candidate for review, delivery, or another external +mutation. Claims do not expire automatically. @@ -475,19 +483,27 @@ pre-mutation request must name the current claim, sequence, and continuation token. Atelier fetches and reevaluates current claim, revision, policy, ticket, candidate, and authority. On allowance, one atomic commit increments `sequence`, rotates `continuation_token`, and appends an authorization entry containing the -invocation ID, phase, action, proposed-effect digest, exact candidate head, -acknowledged candidate head, and recorded time. Atelier pushes and reads that +invocation ID, phase, action, proposed-effect digest, exact candidate head and +remote ref, acknowledged candidate head, and recorded time. Atelier pushes and exact checkpoint back before returning `allow`. Denial echoes the prior token and does not advance the checkpoint or ledger. A consumed sequence or token cannot be replayed, and the ledger is never truncated or rewritten. `candidate_published` uses the same compare-and-swap transition while also recording the exact remotely reachable candidate. An unavailable or ambiguous -mailbox outcome does not acknowledge the candidate. Before accepting any -terminal result, Atelier requires its sequence and continuation token to equal -the current claim-ledger tail. It also requires the terminal `authority_used` -set to equal the allowed `pre_external_mutation` actions in that ledger; either -under-reporting or an unrecorded action blocks the result. +mailbox outcome does not acknowledge the candidate for further mutation. If the +push succeeded but that acknowledgement transition failed, a terminal blocked +result may recover the exact candidate only when its identity matches the sealed +invocation, the ledger tail names its exact pre-mutation push head and remote ref, +remote reachability is reverified, and the blocked receipt binds the candidate in the +same atomic mailbox transition. That receipt becomes the verified transferable +handoff for a later claim. + +Before accepting any terminal result, Atelier requires its sequence and +continuation token to equal the current claim-ledger tail. It also requires the +terminal `authority_used` set to equal the allowed `pre_external_mutation` +actions in that ledger; either under-reporting or an unrecorded action blocks the +result. A worker releases a claim explicitly when it stops without delivering the work. An operator or planner may perform an explicit takeover when the prior worker diff --git a/docs/implementation-plan.md b/docs/implementation-plan.md index 761b89ca..8a20eabb 100644 --- a/docs/implementation-plan.md +++ b/docs/implementation-plan.md @@ -48,7 +48,7 @@ gate. ### Exit criteria - Agent Scripts publishes and validates - `agent-scripts.implement-ticket/delegated-execution/v1`. + `agent-scripts.implement-ticket/delegated-execution/v2`. - The design, [Git Mailbox Contract], [Atelier Project Policy Contract], and implementation plan agree on lifecycle, authority, delivery, and acceptance. - No legacy ticket, tag, or implementation has been mutated as part of this diff --git a/docs/project-policy-contract.md b/docs/project-policy-contract.md index 2b1c6939..26427628 100644 --- a/docs/project-policy-contract.md +++ b/docs/project-policy-contract.md @@ -45,7 +45,7 @@ ticket: - relationships execution: - capability: agent-scripts.implement-ticket/delegated-execution/v1 + capability: agent-scripts.implement-ticket/delegated-execution/v2 delivery_outcome: ready_pr parallel_assignments: false @@ -146,7 +146,7 @@ sorted by stable provider identifier before encoding. ### Execution `execution.capability` is exactly -`agent-scripts.implement-ticket/delegated-execution/v1`. + `agent-scripts.implement-ticket/delegated-execution/v2`. `execution.delivery_outcome` is exactly `ready_pr`. diff --git a/docs/reset-gate-proposal.md b/docs/reset-gate-proposal.md index 5f438bfa..72d82e2c 100644 --- a/docs/reset-gate-proposal.md +++ b/docs/reset-gate-proposal.md @@ -40,7 +40,7 @@ The validated dependency is: - Agent Scripts commit: `861dd04c526d7e2ab7f33d112a00a370db17aae9`; - installed plugin: `agent-scripts@agent-scripts` version `0.1.0`; -- delegated capability: `agent-scripts.implement-ticket/delegated-execution/v1`; + - delegated capability: `agent-scripts.implement-ticket/delegated-execution/v2`; - installed `implement-ticket` skill SHA-256: `307b660864b15a167e755d0f47840acb56d20e1e8ec940d110d84548aec85243`; and - capability manifest SHA-256: diff --git a/experiments/mailbox_protocol_v0.py b/experiments/mailbox_protocol_v0.py index ded239e1..3c208193 100644 --- a/experiments/mailbox_protocol_v0.py +++ b/experiments/mailbox_protocol_v0.py @@ -312,6 +312,7 @@ def authorize_checkpoint( "action": action, "proposed_effect_digest": ticket_digest({"action": action, "claim_id": claim_id}), "candidate_head": None, + "candidate_remote_ref": None, "acknowledged_candidate_head": None, "recorded_at": "2026-07-25T12:25:00Z", } diff --git a/scripts/validate_repository.py b/scripts/validate_repository.py index 4c627f0f..c5848036 100755 --- a/scripts/validate_repository.py +++ b/scripts/validate_repository.py @@ -82,8 +82,11 @@ def validate_skill(errors: list[str]) -> None: errors.append("Atelier skill frontmatter must include a description") if "[TODO:" in text: errors.append("Atelier skill contains a TODO placeholder") - if "Delegated implementation and audit are not implemented yet" not in text: - errors.append("skill must state that delegated work and audit are unavailable") + if "Production `audit` behavior is not implemented yet" not in text: + errors.append("skill must state that audit is unavailable") + for relative in ("references/delegation.md", "scripts/delegation.py"): + if not (path.parent / relative).is_file(): + errors.append(f"delegated work boundary missing: {relative}") def validate_reset_boundary(errors: list[str]) -> None: diff --git a/skills/atelier/SKILL.md b/skills/atelier/SKILL.md index c2d9a0eb..3ad76cd3 100644 --- a/skills/atelier/SKILL.md +++ b/skills/atelier/SKILL.md @@ -23,7 +23,7 @@ fail-closed startup preflight. The preflight must prove: - the exact plugin-qualified `agent-scripts:implement-ticket` skill identity; -- the compatible `agent-scripts.implement-ticket/delegated-execution/v1` +- the compatible `agent-scripts.implement-ticket/delegated-execution/v2` manifest, schemas, and dependency-owned validator; - the six v0 candidate, pull-request, and review authority actions declared by the host capability descriptor; @@ -52,24 +52,24 @@ the fail-closed host preflight, and use `scripts/claiming.py`. Never mint a replacement fence for a stale worker or acknowledge a candidate that is not reachable at its exact declared remote ref and head. -Delegated implementation and audit are not implemented yet. Work mode stops -after durable claim/checkpoint coordination; it cannot launch Agent Scripts, -validate a terminal result, or deliver the assignment. Do not emulate missing -behavior with ad hoc orchestration, hidden local state, copied Agent Scripts -workflows, or tracker mutations. +Production delegated implementation is implemented for one active claimed +assignment. Before delegation, read `references/delegation.md`, repeat the +fail-closed host preflight, and use `scripts/delegation.py` to prepare the exact +v2 invocation, service fresh-observation checkpoints, and validate one terminal +result. Launch one fresh worker through the host with the installed +`agent-scripts:implement-ticket` skill. Do not copy its workflow, spawn a +substitute CLI process, cache provider observations, or widen Atelier's v0 +authority ceiling. -When either unavailable behavior is requested: +Production `audit` behavior is not implemented yet. When it is requested: -1. Identify the requested behavior: delegated work or audit. -1. Report that the mode is unavailable in the reset scaffold. -1. Link the owning issue: - - delegated work: shaug/atelier#779 - - `audit`: shaug/atelier#780 +1. Report that audit is unavailable in the reset scaffold. +1. Link the owning issue: shaug/atelier#780. 1. Make no mailbox, repository, ticket, pull-request, or acceptance mutation. An explicit host-readiness request may complete the read-only host preflight and -return its exact compatibility or failure result. That does not make any -production mode available. +return its exact compatibility or failure result. That does not make audit +available. ## Mailbox boundary diff --git a/skills/atelier/references/claiming.md b/skills/atelier/references/claiming.md index 9cbc8256..a60f7076 100644 --- a/skills/atelier/references/claiming.md +++ b/skills/atelier/references/claiming.md @@ -109,7 +109,8 @@ Add a `checkpoint` object containing: - the current `fence` (`claim_id`, `worker_run_id`, `sequence`, and `continuation_token`); - `phase`, `action`, and `proposed_effect_digest`; -- `candidate_head` and `acknowledged_candidate_head`, including explicit nulls; +- `candidate_head`, `candidate_remote_ref`, and `acknowledged_candidate_head`, + including explicit nulls; - a new `next_continuation_token`; - `recorded_at`; and - `candidate`, explicitly null except for `candidate_published`. diff --git a/skills/atelier/references/delegation.md b/skills/atelier/references/delegation.md new file mode 100644 index 00000000..b71f489a --- /dev/null +++ b/skills/atelier/references/delegation.md @@ -0,0 +1,85 @@ +# Delegated implementation + +Atelier delegates exactly one active, approved, claimed assignment to the +independently installed `agent-scripts:implement-ticket` skill. Agent Scripts +owns implementation and its transitive workflow. Atelier owns the approved +intent, invocation boundary, authority checkpoints, terminal validation, and +durable receipt. + +## Preconditions + +1. Complete `host-boundary.md` against the exact installed delegated-execution + v2 bundle. +2. Read `claiming.md` and obtain the current claim fence and approval commit. +3. Capture a complete GitHub observation after a new read boundary. +4. Use `scripts/delegation.py` with `operation: prepare`. Provide an immutable local + invocation path and a host-owned observation command that emits one fresh complete + GitHub observation. Preparation builds and probes the one-shot stdin/stdout + checkpoint adapter, writes the exact invocation locally, and atomically seals its + canonical digest into the current claim. Pass that invocation unchanged to one + fresh host worker. + +Never copy `implement-ticket`, launch a substitute workflow, or persist +host-local task state in the mailbox. Starting the fresh worker is a host +operation; the script does not spawn Codex, a daemon, or a service. + +## Checkpoints + +The fresh worker must call the prepared checkpoint command before every +external mutation and after candidate publication, following the +dependency-owned v2 contract. For every request: + +1. Run the exact prepared checkpoint command. The adapter establishes a new read + boundary, executes the sealed host observation command, validates the resulting + complete observation, and reads exactly one worker request from stdin. +2. The adapter atomically rereads the current claim, current policy and canonical + base, ticket, repository, sequence, and authority before deciding. +3. Read exactly one checkpoint response from stdout. +4. Return the response unchanged. An `allow` rotates the continuation token and + advances exactly one durable mailbox sequence. A `deny` consumes neither. +5. Treat `candidate_published` as an acknowledgement only when the exact remote + ref contains the declared head. + +Do not cache observations, mint replacement fences, widen the six v0 authority +actions, or acknowledge deployment. Atelier v0 accepts only `ready_pr`, +`blocked`, and `requires_epic`. + +## Terminal result + +Invoke `scripts/delegation.py` with `operation: finalize`, the unchanged sealed +invocation, the exact current fence, the worker result, and another fresh complete +observation. Atelier requires the result to report no tracker mutation and to +match the still-open current native ticket before recording a receipt. + +- `ready_pr` requires one ordinary open, non-draft, mergeable pull request whose + base, head, remote candidate, checks, validation, independent review, + feedback, and required acceptance evidence all bind to the exact candidate. +- `blocked` preserves any acknowledged candidate. If the exact push succeeded but + its `candidate_published` acknowledgement failed, the blocked result may recover + that remotely reachable candidate only when it matches the sealed invocation and + the ledger's final authorized push head and remote ref. Atelier binds the candidate + and immutable blocked receipt atomically, records one unresolved planner decision, + and retains + mutation ownership. +- `requires_epic` is recorded as a blocked attempt for planner action. Atelier + does not invoke `implement-epic` from delegated work. + +Every terminal result records exactly one immutable attempt receipt. +`ready_pr` also records that receipt as the delivery receipt. Delivery is not +operator acceptance, merge, deployment, or ticket completion. + +## JSON request boundary + +The CLI accepts one JSON object containing: + +- `operation`: `prepare`, `checkpoint`, or `finalize`; +- `mailbox`: canonical `remote` and `branch`; +- `work_id`, `approved_commit`, `policy_target`, `host_target`; +- `observation_path` and `observation_not_before`; +- for `prepare`, `checkpoint_invocation_path` and the host-owned + `observation_command`; and +- the remaining operation-specific fence, invocation, checkpoint request, result, + and timestamps. + +Unknown, missing, stale, contradictory, or unverifiable state fails closed +before a mailbox success is reported. diff --git a/skills/atelier/references/host-capability.json b/skills/atelier/references/host-capability.json index e224ddf5..b0c5a103 100644 --- a/skills/atelier/references/host-capability.json +++ b/skills/atelier/references/host-capability.json @@ -1,7 +1,7 @@ { "schema": "atelier.host-capability/v1", "reference_host": "codex", - "delegated_capability": "agent-scripts.implement-ticket/delegated-execution/v1", + "delegated_capability": "agent-scripts.implement-ticket/delegated-execution/v2", "delegated_authority_actions": [ "repository.candidate.create", "repository.candidate.push", @@ -18,17 +18,17 @@ "delegated_skill": { "stable_name": "agent-scripts:implement-ticket", "frontmatter_name": "implement-ticket", - "skill_sha256": "307b660864b15a167e755d0f47840acb56d20e1e8ec940d110d84548aec85243", + "skill_sha256": "3fabc1d030bd936c6a40caf10aa92ecbe1b32295825940e58342c7f99ea6acc4", "capability_manifest": "references/delegated-execution/capability.json", - "capability_manifest_sha256": "551d2e883d226d2a1e7e39eae66eb2bd3d9d88ec18c56cfba190253677e34156", + "capability_manifest_sha256": "60f259e7d4fba7cd4f7ad0b91dee8fd809ea81b7d4e9faa7e7e4867c8c2a3150", "bundle_sha256": { - "CONTRACT.md": "760601f774a461e7f15b03bb52d284c8fdac0a119a88c63274b8e97fc8c9f016", - "capability.schema.json": "16188433bcbf979776a8e56c6905caaf5bede1b3eca8ab51d5798dd4841fb5f0", - "invocation.schema.json": "43ae1f337500d5955c9ed9c4c057a2105bc17f951533b095b63c6aea198c71af", - "checkpoint-request.schema.json": "d94d38243c411f98b7f9c0169684595b69b3de09655f0c258f7d3522e5354aa9", - "checkpoint-response.schema.json": "7e54a12959eaac937d62a97ad736f2bc924cb085df113b87d1272d6b35ed3abb", - "result.schema.json": "bc62b055faed7fce8bef4694e8bb9e714ed032092d687befebd549b4cffe21ce", - "validate.py": "a9e49308e7165eaa0abd18952f15185dfb463c35c39b38a2933b43147be560c1" + "CONTRACT.md": "8c94c1c4bc7a379602bd802a987492fa1c5cd7a9c9c05788fe52f27be1fe9a6a", + "capability.schema.json": "586038f1cedce0b282ab5f291a319efbe5a7ffa78edd6670f46f7ff00e06f833", + "invocation.schema.json": "43416a186655623a9f35a83d739319ff1f83c5a6b0084e00db8470f45d4d0e3c", + "checkpoint-request.schema.json": "c0018468879ef356295695ba4e25c676cd77f8bed563a880c018874890a8157f", + "checkpoint-response.schema.json": "29e9e9c50fcf583532d9f88dc862c5f06422823056ef3ebf8682434c3e9b5a83", + "result.schema.json": "0128e54180cea4b9334c5dd87559ab5c2826a12355f7bbbaa485b80bc599daa1", + "validate.py": "01e017b8b6349d43a71ab5145f6bf9126b8811b64a13898d1aa16b65b3b48bde" } }, "native_state": { diff --git a/skills/atelier/references/mailbox-v1.schema.json b/skills/atelier/references/mailbox-v1.schema.json index ac3c7641..af8f7c53 100644 --- a/skills/atelier/references/mailbox-v1.schema.json +++ b/skills/atelier/references/mailbox-v1.schema.json @@ -108,6 +108,7 @@ "action", "proposed_effect_digest", "candidate_head", + "candidate_remote_ref", "acknowledged_candidate_head", "recorded_at" ], @@ -134,6 +135,16 @@ "candidate_head": { "$ref": "#/$defs/nullable_sha" }, + "candidate_remote_ref": { + "oneOf": [ + { + "type": "null" + }, + { + "$ref": "#/$defs/branch_ref" + } + ] + }, "acknowledged_candidate_head": { "$ref": "#/$defs/nullable_sha" }, @@ -228,6 +239,7 @@ "approved_commit", "policy_commit", "ticket_observation_digest", + "invocation_digest", "claimed_at", "host", "checkpoint", @@ -253,6 +265,16 @@ "ticket_observation_digest": { "$ref": "#/$defs/digest" }, + "invocation_digest": { + "oneOf": [ + { + "$ref": "#/$defs/digest" + }, + { + "type": "null" + } + ] + }, "claimed_at": { "$ref": "#/$defs/timestamp" }, @@ -748,7 +770,7 @@ ], "properties": { "capability": { - "const": "agent-scripts.implement-ticket/delegated-execution/v1" + "const": "agent-scripts.implement-ticket/delegated-execution/v2" }, "delivery_outcome": { "const": "ready_pr" diff --git a/skills/atelier/scripts/claiming.py b/skills/atelier/scripts/claiming.py index ac7cb195..cae2720f 100644 --- a/skills/atelier/scripts/claiming.py +++ b/skills/atelier/scripts/claiming.py @@ -33,6 +33,7 @@ from skills.atelier.scripts.mailbox import ( MailboxValidationError, _github_remote_url, + _valid_branch_ref, validate_project_policy, ) from skills.atelier.scripts.planning import ( @@ -111,10 +112,12 @@ class CheckpointRequest: action: str proposed_effect_digest: str candidate_head: str | None + candidate_remote_ref: str | None acknowledged_candidate_head: str | None next_continuation_token: str recorded_at: datetime candidate: Mapping[str, Any] | None = None + ticket_observation_digest: str | None = None @dataclass(frozen=True) @@ -255,6 +258,7 @@ def revalidate(context: TransitionContext) -> None: "approved_commit": approved_commit, "policy_commit": state.current_policy.commit, "ticket_observation_digest": state.ticket_digest, + "invocation_digest": None, "claimed_at": _timestamp(claimed_at), "host": "codex", "checkpoint": { @@ -293,6 +297,7 @@ def authorize( observation_path: Path, observation_not_before: datetime, now: datetime | None = None, + state_revalidator: Callable[[_ExecutionState], None] | None = None, ) -> ClaimResult: """Allow one exact external action or acknowledge one exact published candidate.""" _require_identifier(work_id, "wrk") @@ -310,6 +315,8 @@ def revalidate(context: TransitionContext) -> None: observation_not_before=observation_not_before, now=now, ) + if state_revalidator is not None: + state_revalidator(state) if state.work["status"] != "active": raise MailboxTransitionRejected(f"{work_id}: checkpoints require active work") claim = _require_fence(state.work, request.fence) @@ -319,6 +326,13 @@ def revalidate(context: TransitionContext) -> None: raise MailboxTransitionRejected( f"{work_id}: material ticket observation changed during the invocation" ) + if ( + request.ticket_observation_digest is not None + and request.ticket_observation_digest != state.ticket_digest + ): + raise MailboxTransitionRejected( + f"{work_id}: checkpoint ticket observation is not current" + ) effective_authority = set(state.effective_policy["authority"]["allow"]) approved_authority = set(state.work["approval"]["authority_ceiling"]) if request.action not in effective_authority & approved_authority: @@ -336,8 +350,7 @@ def plan(context: TransitionContext) -> TransitionPlan: work["claim"] = copy.deepcopy(planned["claim"]) return TransitionPlan( commit_message=( - f"record {request.phase} checkpoint {request.fence.sequence + 1} " - f"for {work_id}" + f"record {request.phase} checkpoint {request.fence.sequence + 1} for {work_id}" ), changes=(FileChange(_work_path(work_id), _render_document(work, state.body)),), ) @@ -537,6 +550,7 @@ def revalidate(context: TransitionContext) -> None: "approved_commit": approved_commit, "policy_commit": state.current_policy.commit, "ticket_observation_digest": state.ticket_digest, + "invocation_digest": None, "claimed_at": _timestamp(taken_over_at), "host": "codex", "checkpoint": { @@ -768,9 +782,11 @@ def _apply_checkpoint( if ( request.action != "repository.candidate.push" or request.candidate_head is None + or request.candidate_remote_ref is None or request.acknowledged_candidate_head != request.candidate_head or candidate is None or candidate.get("head_revision") != request.candidate_head + or candidate.get("remote_ref") != request.candidate_remote_ref ): raise MailboxTransitionRejected( "candidate publication must acknowledge one exact repository candidate push" @@ -784,6 +800,7 @@ def _apply_checkpoint( prior["phase"] != "pre_external_mutation" or prior["action"] != "repository.candidate.push" or prior["candidate_head"] != request.candidate_head + or prior["candidate_remote_ref"] != request.candidate_remote_ref or prior["proposed_effect_digest"] != request.proposed_effect_digest ): raise MailboxTransitionRejected( @@ -796,17 +813,25 @@ def _apply_checkpoint( raise MailboxTransitionRejected( "pre-mutation checkpoint cannot acknowledge a candidate" ) - if request.action in CANDIDATE_REQUIRED_ACTIONS and request.candidate_head is None: + if request.action in CANDIDATE_REQUIRED_ACTIONS and ( + request.candidate_head is None or request.candidate_remote_ref is None + ): raise MailboxTransitionRejected( - f"action {request.action!r} requires an exact candidate head" + f"action {request.action!r} requires an exact candidate head and remote ref" ) current_head = ( claim["candidate"]["head_revision"] if claim["candidate"] is not None else None ) + current_ref = ( + claim["candidate"]["remote_ref"] if claim["candidate"] is not None else None + ) if ( request.action in CANDIDATE_REQUIRED_ACTIONS and request.action != "repository.candidate.push" - and request.candidate_head != current_head + and ( + request.candidate_head != current_head + or request.candidate_remote_ref != current_ref + ) ): raise MailboxTransitionRejected( f"action {request.action!r} does not name the acknowledged candidate" @@ -822,6 +847,7 @@ def _apply_checkpoint( "action": request.action, "proposed_effect_digest": request.proposed_effect_digest, "candidate_head": request.candidate_head, + "candidate_remote_ref": request.candidate_remote_ref, "acknowledged_candidate_head": request.acknowledged_candidate_head, "recorded_at": _timestamp(request.recorded_at), } @@ -831,9 +857,7 @@ def _apply_checkpoint( def _verify_capability(self, target: HostTarget) -> None: if not self.capability_verifier(target): - raise ClaimingError( - "required delegated implement-ticket capability is unavailable" - ) + raise ClaimingError("required delegated implement-ticket capability is unavailable") def _verify_candidate(self, candidate: Mapping[str, Any] | None) -> None: if candidate is not None and not self.candidate_verifier(candidate): @@ -882,8 +906,7 @@ def _effective_policy( approved["ticket"]["material_fields"], current["ticket"]["material_fields"] ) effective["ticket"]["require_no_blockers"] = ( - approved["ticket"]["require_no_blockers"] - or current["ticket"]["require_no_blockers"] + approved["ticket"]["require_no_blockers"] or current["ticket"]["require_no_blockers"] ) return effective @@ -918,7 +941,7 @@ def _require_policy_identity( if policy["repository"]["identity"] != expected_repository: raise MailboxTransitionRejected("project policy repository identity is incompatible") if policy["execution"] != { - "capability": "agent-scripts.implement-ticket/delegated-execution/v1", + "capability": "agent-scripts.implement-ticket/delegated-execution/v2", "delivery_outcome": "ready_pr", "parallel_assignments": False, }: @@ -954,6 +977,7 @@ def _policy_remote_matches_repository(target: PolicyTarget, repository: str) -> remote_url = target.remote return _github_remote_url(remote_url, repository=repository) + def _candidate_remote_reachable(candidate: Mapping[str, Any]) -> bool: remote_url = candidate.get("remote_url") remote_ref = candidate.get("remote_ref") @@ -1008,7 +1032,7 @@ def _capability_compatible(target: HostTarget) -> bool: return ( result["status"] == "compatible" and result["delegated_capability"] - == "agent-scripts.implement-ticket/delegated-execution/v1" + == "agent-scripts.implement-ticket/delegated-execution/v2" ) @@ -1055,6 +1079,12 @@ def _validate_checkpoint_request(request: CheckpointRequest) -> None: ): if value is not None: _require_sha(value, label) + if (request.candidate_head is None) != (request.candidate_remote_ref is None): + raise ClaimingError("candidate_head and candidate_remote_ref must be present together") + if request.candidate_remote_ref is not None and not _valid_branch_ref( + request.candidate_remote_ref + ): + raise ClaimingError("candidate_remote_ref must be a valid full Git branch ref") _require_token(request.next_continuation_token, "next_continuation_token") if request.next_continuation_token == request.fence.continuation_token: raise ClaimingError("next continuation token must rotate") @@ -1336,6 +1366,7 @@ def main() -> int: action=checkpoint["action"], proposed_effect_digest=checkpoint["proposed_effect_digest"], candidate_head=checkpoint["candidate_head"], + candidate_remote_ref=checkpoint["candidate_remote_ref"], acknowledged_candidate_head=checkpoint["acknowledged_candidate_head"], next_continuation_token=checkpoint["next_continuation_token"], recorded_at=_parse_timestamp(checkpoint["recorded_at"]), diff --git a/skills/atelier/scripts/delegation.py b/skills/atelier/scripts/delegation.py new file mode 100644 index 00000000..e9d698b6 --- /dev/null +++ b/skills/atelier/scripts/delegation.py @@ -0,0 +1,1151 @@ +#!/usr/bin/env python3 +"""Prepare, checkpoint, and finalize one delegated Agent Scripts execution.""" + +from __future__ import annotations + +import argparse +import copy +import hashlib +import importlib.util +import json +import os +import secrets +import re +import subprocess +import sys +import tempfile +from collections.abc import Mapping, Sequence +from dataclasses import asdict +from datetime import UTC, datetime +from pathlib import Path +from types import ModuleType +from typing import Any + +if __package__ in {None, ""}: + sys.path.insert(0, str(Path(__file__).resolve().parents[3])) + +from skills.atelier.scripts.claiming import ( + AUTHORITY_ACTIONS, + AttemptEvidence, + CheckpointRequest, + ClaimCoordinator, + ClaimFence, + HostTarget, + _attempt_receipt, + _message_path, + _read_work, + _receipt_path, + _timestamp, + _work_path, +) +from skills.atelier.scripts.git_mailbox import ( + FileChange, + MailboxTransitionRejected, + TransitionContext, + TransitionPlan, + WriteResult, +) +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 + +CAPABILITY = "agent-scripts.implement-ticket/delegated-execution/v2" +INVOCATION_SCHEMA = "agent-scripts.implement-ticket/delegated-invocation/v2" +REQUEST_SCHEMA = "agent-scripts.implement-ticket/checkpoint-request/v2" +RESPONSE_SCHEMA = "agent-scripts.implement-ticket/checkpoint-response/v2" +RESULT_SCHEMA = "agent-scripts.implement-ticket/delegated-result/v2" +ACCEPTED_TERMINALS = ("ready_pr", "blocked", "requires_epic") +SECTION_PATTERN = re.compile( + r"^## (?P[^\n]+)\n\n(?P<body>.*?)(?=^## |\Z)", + re.MULTILINE | re.DOTALL, +) + + +class DelegationError(RuntimeError): + """The delegated execution boundary cannot safely continue.""" + + +class DelegationCoordinator: + """Host-owned adapter around the dependency-owned delegated protocol.""" + + def __init__(self, claims: ClaimCoordinator): + self.claims = claims + + def prepare( + self, + work_id: str, + fence: ClaimFence, + *, + approved_commit: str, + policy_target: PolicyTarget, + host_target: HostTarget, + observation_path: Path, + observation_not_before: datetime, + checkpoint_invocation_path: Path, + observation_command: Sequence[str], + now: datetime | None = None, + ) -> dict[str, Any]: + """Build one exact invocation from fresh approved mailbox and provider state.""" + + dependency = self._dependency(host_target) + if ( + not observation_command + or any(not isinstance(item, str) or not item for item in observation_command) + ): + raise DelegationError("observation command must contain nonempty string arguments") + checkpoint_command = _checkpoint_command( + invocation_path=checkpoint_invocation_path, + mailbox_remote=self.claims.writer.remote, + mailbox_branch=self.claims.writer.branch, + work_id=work_id, + approved_commit=approved_commit, + policy_target=policy_target, + host_target=host_target, + observation_command=observation_command, + ) + planned: dict[str, Any] = {} + + def revalidate(context: TransitionContext) -> None: + state = self.claims._execution_state( + context, + work_id, + approved_commit=approved_commit, + policy_target=policy_target, + observation_path=observation_path, + observation_not_before=observation_not_before, + now=now, + ) + if state.work["status"] != "active": + raise MailboxTransitionRejected(f"{work_id}: delegation requires active work") + claim = _require_fence(state.work, fence) + if claim["invocation_digest"] is not None: + raise MailboxTransitionRejected( + f"{work_id}: delegated invocation is already sealed" + ) + sections = _sections(state.body) + policy = state.effective_policy + base_ref = policy["repository"]["canonical_ref"] + base_sha = state.current_policy.commit + invocation = { + "schema": INVOCATION_SCHEMA, + "capability": CAPABILITY, + "invocation_id": claim["worker_run_id"], + "ticket": { + "provider": state.work["native_ticket"]["provider"], + "id": state.work["native_ticket"]["id"], + "url": state.work["native_ticket"]["url"], + "observation": state.ticket_digest, + }, + "repository": { + "identity": state.project["repository"], + "remote_url": ( + "https://github.com/" + + state.project["repository"].removeprefix("github:") + + ".git" + ), + "base_ref": base_ref, + "base_sha": base_sha, + }, + "work": { + "id": work_id, + "revision": state.work["revision"], + "approval_evidence": approved_commit, + "intent": sections["intent"], + "scope": [sections["scope"]], + "non_goals": [sections["non goals"]], + "constraints": [sections["constraints"]], + "done_definition": [sections["done definition"]], + }, + "validation": policy["validation"]["required_commands"], + "review": { + "independent": True, + "unresolved_feedback_required": True, + }, + "authority": { + "allow": [ + action + for action in state.work["approval"]["authority_ceiling"] + if action in policy["authority"]["allow"] and action in AUTHORITY_ACTIONS + ] + }, + "desired_outcome": "ready_pr", + "accepted_terminal_states": list(ACCEPTED_TERMINALS), + "acceptance_requirements": [ + { + "criterion": evidence, + "required": True, + "evidence_category": evidence, + "stage": "pre_merge", + "identity": "candidate", + "environment": None, + "url": None, + "source": "atelier.approval.acceptance.required_evidence", + } + for evidence in policy["acceptance"]["evidence"] + ], + "starting_deployment": None, + "checkpoint": { + "command": list(checkpoint_command), + "last_sequence": claim["checkpoint"]["sequence"], + "continuation_token": claim["checkpoint"]["continuation_token"], + }, + } + _require_valid(dependency, "invocation", invocation) + _write_json_atomic(checkpoint_invocation_path, invocation) + _probe_checkpoint_command(checkpoint_command) + updated_claim = copy.deepcopy(claim) + updated_claim["invocation_digest"] = _canonical_digest(invocation) + planned.clear() + planned.update(state=state, claim=updated_claim, invocation=invocation) + + def plan(context: TransitionContext) -> TransitionPlan: + state = planned["state"] + work = copy.deepcopy(state.work) + work["claim"] = planned["claim"] + return TransitionPlan( + commit_message=f"seal delegated invocation for {work_id}", + changes=(FileChange(_work_path(work_id), _render_document(work, state.body)),), + ) + + self.claims.writer.publish("prepare delegation", revalidate=revalidate, plan=plan) + return copy.deepcopy(planned["invocation"]) + + def checkpoint( + self, + work_id: str, + invocation: Mapping[str, Any], + request: Mapping[str, Any], + *, + approved_commit: str, + policy_target: PolicyTarget, + host_target: HostTarget, + observation_path: Path, + observation_not_before: datetime, + recorded_at: datetime, + next_continuation_token: str, + now: datetime | None = None, + ) -> dict[str, Any]: + """Service one request, denying drift without consuming the claim fence.""" + + dependency = self._dependency(host_target) + _require_valid(dependency, "invocation", invocation) + _require_valid(dependency, "checkpoint-request", request) + if request["invocation_id"] != invocation["invocation_id"]: + raise DelegationError("checkpoint invocation identity mismatch") + if request["capability"] != CAPABILITY: + raise DelegationError("checkpoint capability mismatch") + current_claim = self.claims.writer.observe( + "read delegated checkpoint fence", + lambda context: copy.deepcopy(_read_work(context.checkout, work_id)[0]["claim"]), + ) + if current_claim is None or current_claim["worker_run_id"] != invocation["invocation_id"]: + raise DelegationError("delegated invocation no longer owns the current claim") + prior_token = request["continuation_token"] + response = { + "schema": RESPONSE_SCHEMA, + "invocation_id": request["invocation_id"], + "request_sequence": request["sequence"], + "prior_continuation_token": prior_token, + "continuation_token": prior_token, + "decision": "deny", + "reason": None, + "acknowledged_candidate_sha": None, + "observed_deployment": None, + } + try: + if current_claim["invocation_digest"] != _canonical_digest(invocation): + raise MailboxTransitionRejected( + "delegated invocation does not match its sealed digest" + ) + if request["sequence"] != current_claim["checkpoint"]["sequence"] + 1: + raise MailboxTransitionRejected("checkpoint sequence does not advance exactly once") + if prior_token != current_claim["checkpoint"]["continuation_token"]: + raise MailboxTransitionRejected("checkpoint continuation token is stale") + _require_checkpoint_candidate(invocation, request.get("candidate")) + candidate = ( + _mailbox_candidate(request.get("candidate"), recorded_at) + if request["phase"] == "candidate_published" + else None + ) + result = self.claims.authorize( + work_id, + CheckpointRequest( + fence=ClaimFence( + claim_id=current_claim["id"], + worker_run_id=invocation["invocation_id"], + sequence=request["sequence"] - 1, + continuation_token=prior_token, + ), + phase=request["phase"], + action=request["action"], + proposed_effect_digest=_digest(request["proposed_effect"]), + candidate_head=request["candidate"]["head_sha"] + if request["candidate"] + else None, + candidate_remote_ref=request["candidate"]["remote_ref"] + if request["candidate"] + else None, + acknowledged_candidate_head=( + request["candidate"]["head_sha"] + if request["phase"] == "candidate_published" and request["candidate"] + else None + ), + next_continuation_token=next_continuation_token, + recorded_at=recorded_at, + candidate=candidate, + ticket_observation_digest=request["ticket_observation"], + ), + approved_commit=approved_commit, + policy_target=policy_target, + host_target=host_target, + observation_path=observation_path, + observation_not_before=observation_not_before, + now=now, + state_revalidator=lambda state: _require_current_invocation_contract( + invocation, state + ), + ) + except (DelegationError, MailboxTransitionRejected) as error: + response["reason"] = str(error) + else: + response.update( + decision="allow", + reason=None, + continuation_token=result.continuation_token, + acknowledged_candidate_sha=( + request["candidate"]["head_sha"] + if request["phase"] == "candidate_published" and request["candidate"] + else None + ), + ) + if response["decision"] == "allow": + errors = dependency.validate_checkpoint_progress( + current_claim["checkpoint"]["sequence"], + current_claim["checkpoint"]["continuation_token"], + dict(request), + response, + ) + else: + errors = dependency.validate_checkpoint_exchange(dict(request), response) + if errors: + raise DelegationError("; ".join(errors)) + return response + + def finalize( + self, + work_id: str, + invocation: Mapping[str, Any], + result: Mapping[str, Any], + fence: ClaimFence, + *, + approved_commit: str, + policy_target: PolicyTarget, + host_target: HostTarget, + observation_path: Path, + observation_not_before: datetime, + ended_at: datetime, + now: datetime | None = None, + ) -> WriteResult: + """Validate one terminal result and record one immutable attempt receipt.""" + + dependency = self._dependency(host_target) + _require_valid(dependency, "invocation", invocation) + current_claim = self.claims.writer.observe( + "read delegated result fence", + lambda context: copy.deepcopy(_read_work(context.checkout, work_id)[0]["claim"]), + ) + if current_claim is None or ( + current_claim["id"], + current_claim["worker_run_id"], + current_claim["checkpoint"]["sequence"], + current_claim["checkpoint"]["continuation_token"], + ) != ( + fence.claim_id, + fence.worker_run_id, + fence.sequence, + fence.continuation_token, + ): + raise DelegationError("delegated terminal result claim fence is stale") + if current_claim["invocation_digest"] != _canonical_digest(invocation): + raise DelegationError("delegated invocation does not match its sealed digest") + consumed = list( + dict.fromkeys( + entry["action"] + for entry in current_claim["checkpoint"]["authorizations"] + if entry["phase"] == "pre_external_mutation" + ) + ) + errors = dependency.validate_result_checkpoint_state( + dict(invocation), + dict(result), + fence.sequence, + fence.continuation_token, + None, + None, + consumed, + ) + if set(result.get("authority_used", [])) != set(consumed): + errors.append("$.authority_used: does not match Atelier's checkpoint ledger") + if errors: + raise DelegationError("; ".join(errors)) + terminal = result["terminal_state"] + if terminal not in ACCEPTED_TERMINALS: + raise DelegationError( + f"terminal state {terminal!r} exceeds Atelier's accepted boundary" + ) + receipt_id = new_identifier("rcp") + message_id = new_identifier("msg") if terminal != "ready_pr" else None + planned: dict[str, Any] = {} + + def revalidate(context: TransitionContext) -> None: + self.claims._verify_capability(host_target) + state = self.claims._execution_state( + context, + work_id, + approved_commit=approved_commit, + policy_target=policy_target, + observation_path=observation_path, + observation_not_before=observation_not_before, + now=now, + ) + _require_current_invocation_contract(invocation, state) + if state.work["status"] != "active": + raise MailboxTransitionRejected(f"{work_id}: terminal result requires active work") + claim = copy.deepcopy(_require_fence(state.work, fence)) + if claim["invocation_digest"] != _canonical_digest(invocation): + raise MailboxTransitionRejected( + "delegated invocation does not match its sealed digest" + ) + if result["ticket"]["observation"] != state.ticket_digest: + raise MailboxTransitionRejected("terminal ticket observation is not current") + _require_tracker_transition(result, state.observation) + evidence = _attempt_evidence(result) + outcome = "delivered" if terminal == "ready_pr" else "blocked" + body = ( + "Delegated candidate delivered." + if outcome == "delivered" + else ( + result["blocking_reason"] + if terminal == "blocked" + else result["handoff"]["reason"] or result["next_action"] + ) + ) + if outcome == "delivered": + self.claims._verify_candidate(claim["candidate"]) + _require_ready_pr(result, state.observation, claim) + claim["candidate"] = _delivered_candidate(result, claim["candidate"]) + else: + blocked_candidate = _blocked_candidate(result, claim, ended_at) + if blocked_candidate is not None: + self.claims._verify_candidate(blocked_candidate) + claim["candidate"] = blocked_candidate + message = None + if outcome == "blocked": + message = { + "schema": "atelier.message/v1", + "id": message_id, + "work_id": work_id, + "kind": "needs-decision", + "author_role": "worker", + "worker_run_id": claim["worker_run_id"], + "audience": "planner", + "in_reply_to": None, + "resolves": None, + "blocks": "worker", + "created_at": _timestamp(ended_at), + "subject": ( + "Delegated work requires epic planning" + if terminal == "requires_epic" + else "Delegated work is blocked" + ), + } + work_for_receipt = copy.deepcopy(state.work) + work_for_receipt["claim"] = claim + receipt = _attempt_receipt( + work_for_receipt, + claim, + receipt_id=receipt_id, + outcome=outcome, + mutation_ownership="retained", + ended_at=ended_at, + evidence=evidence, + ) + planned.clear() + planned.update( + state=state, + claim=claim, + receipt=receipt, + outcome=outcome, + body=body, + message=message, + ) + + def plan(context: TransitionContext) -> TransitionPlan: + state = planned["state"] + work = copy.deepcopy(state.work) + work["claim"] = planned["claim"] + work["status"] = planned["outcome"] + work["attempt_receipt_id"] = receipt_id + if planned["outcome"] == "delivered": + work["delivery_receipt_id"] = receipt_id + else: + work["blocking_message_id"] = message_id + changes = [ + FileChange( + _receipt_path(work_id, receipt_id), + _render_document(planned["receipt"], planned["body"]), + ), + FileChange(_work_path(work_id), _render_document(work, state.body)), + ] + if planned["message"] is not None: + changes.insert( + 0, + FileChange( + _message_path(work_id, message_id), + _render_document(planned["message"], planned["body"]), + ), + ) + return TransitionPlan( + commit_message=f"record delegated {planned['outcome']} result for {work_id}", + changes=tuple(changes), + ) + + return self.claims.writer.publish("finalize delegation", revalidate=revalidate, plan=plan) + + def _dependency(self, target: HostTarget) -> ModuleType: + check_host( + descriptor_path=target.descriptor_path, + skill_name=target.skill_name, + skill_root=target.skill_root, + connector=target.connector, + operations=list(target.operations), + ) + path = target.skill_root / "references" / "delegated-execution" / "validate.py" + spec = importlib.util.spec_from_file_location("_atelier_delegated_validator", path) + if spec is None or spec.loader is None: + raise DelegationError("dependency-owned delegated validator cannot be loaded") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _require_fence(work: Mapping[str, Any], fence: ClaimFence) -> dict[str, Any]: + claim = work["claim"] + if claim is None or ( + claim["id"], + claim["worker_run_id"], + claim["checkpoint"]["sequence"], + claim["checkpoint"]["continuation_token"], + ) != ( + fence.claim_id, + fence.worker_run_id, + fence.sequence, + fence.continuation_token, + ): + raise MailboxTransitionRejected(f"{work['id']}: delegated claim fence is stale") + return claim + + +def _sections(body: str) -> dict[str, str]: + sections = { + match.group("title").strip().lower(): match.group("body").strip() + for match in SECTION_PATTERN.finditer(body) + } + required = {"intent", "scope", "non goals", "constraints", "done definition"} + missing = sorted(required - sections.keys()) + if missing or any(not sections[name] for name in required): + raise DelegationError("approved work body is missing sections: " + ", ".join(missing)) + return sections + + +def _require_valid(dependency: ModuleType, kind: str, value: Mapping[str, Any]) -> None: + errors = dependency.validate(kind, dict(value)) + if errors: + raise DelegationError("; ".join(errors)) + + +def _digest(value: str) -> str: + return f"sha256:{hashlib.sha256(value.encode('utf-8')).hexdigest()}" + + +def _canonical_digest(value: Mapping[str, Any]) -> str: + encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + return _digest(encoded) + + +def _require_checkpoint_candidate( + invocation: Mapping[str, Any], + candidate: Mapping[str, Any] | None, +) -> None: + if candidate is None: + return + repository = invocation["repository"] + if ( + candidate["repository"] != repository["identity"] + or candidate["remote_url"] != repository["remote_url"] + or candidate["base_sha"] != repository["base_sha"] + ): + raise MailboxTransitionRejected("checkpoint candidate is foreign to the sealed invocation") + if candidate["remote_ref"] == repository["base_ref"]: + raise MailboxTransitionRejected( + "checkpoint candidate cannot publish to the canonical base ref" + ) + + +def _mailbox_candidate( + value: Mapping[str, Any] | None, recorded_at: datetime +) -> dict[str, Any] | None: + if value is None: + return None + candidate = { + "repository": value["repository"], + "remote": "origin", + "remote_url": value["remote_url"], + "remote_ref": value["remote_ref"], + "base_revision": value["base_sha"], + "head_revision": value["head_sha"], + "pull_request": None, + "workspace_id": None, + "published_at": _timestamp(recorded_at), + } + return candidate + + +def _attempt_evidence(result: Mapping[str, Any]) -> AttemptEvidence: + if any(item["name"] != "review-code-change" for item in result["reviews"]): + raise MailboxTransitionRejected( + "review evidence uses a mechanism Atelier cannot represent" + ) + verdicts = { + "passed": "clean", + "failed": "changes_required", + "unavailable": "blocked", + } + return AttemptEvidence( + validation=tuple( + { + "command": item["name"], + "outcome": item["outcome"], + "candidate_revision": item["candidate_sha"], + "observed_at": item["observed_at"], + } + for item in result["validation"] + ), + reviews=tuple( + { + "mechanism": "review-code-change", + "verdict": verdicts[item["outcome"]], + "candidate_revision": item["candidate_sha"], + "comparison_base_revision": result["repository"]["base_sha"], + "observed_at": item["observed_at"], + } + for item in result["reviews"] + ), + unresolved_obligations=tuple(result["unresolved_obligations"]), + ) + + +def _require_ready_pr( + result: Mapping[str, Any], + observation: Mapping[str, Any], + claim: Mapping[str, Any], +) -> None: + candidate = result["candidate"] + if result["implementation_state"] != "published" or candidate is None: + raise MailboxTransitionRejected("ready_pr requires one published candidate") + publication = candidate["publication"] + if publication["kind"] != "ordinary" or len(publication["pull_requests"]) != 1: + raise MailboxTransitionRejected("Atelier v0 accepts one ordinary pull request") + pull_request = publication["pull_requests"][0] + if ( + pull_request["head_ref"], + pull_request["head_sha"], + pull_request["base_sha"], + ) != ( + candidate["remote_ref"], + candidate["head_sha"], + candidate["base_sha"], + ): + raise MailboxTransitionRejected( + "terminal pull request does not identify the acknowledged candidate" + ) + current = claim["candidate"] + if current is None or ( + candidate["repository"], + candidate["remote_url"], + candidate["remote_ref"], + candidate["base_sha"], + candidate["head_sha"], + ) != ( + current["repository"], + current["remote_url"], + current["remote_ref"], + current["base_revision"], + current["head_revision"], + ): + raise MailboxTransitionRejected( + "terminal candidate does not match the acknowledged claim candidate" + ) + live = observation["pull_request"] + if live is None or ( + live["url"], + live["state"], + live["is_draft"], + live["mergeable"], + live["head"]["repository"], + live["head"]["ref"], + live["head"]["sha"], + live["base"]["ref"], + live["base"]["sha"], + ) != ( + pull_request["url"], + "OPEN", + False, + "MERGEABLE", + candidate["repository"].removeprefix("github:"), + pull_request["head_ref"], + pull_request["head_sha"], + pull_request["base_ref"], + pull_request["base_sha"], + ): + raise MailboxTransitionRejected("live pull request identity is not the terminal candidate") + if any( + check["status"].upper() != "COMPLETED" + or (check["conclusion"] or "").upper() not in {"SUCCESS", "NEUTRAL", "SKIPPED"} + for check in observation["checks"] + ): + raise MailboxTransitionRejected("live candidate checks are incomplete or failing") + if any( + not thread["is_resolved"] and not thread["is_outdated"] for thread in observation["threads"] + ): + raise MailboxTransitionRejected("live pull request has unresolved review threads") + feedback = result["feedback"] + if feedback is None or feedback["unresolved_material_count"] != 0: + raise MailboxTransitionRejected("ready_pr requires zero unresolved material feedback") + if any( + item["outcome"] != "passed" or item["candidate_sha"] != candidate["head_sha"] + for item in result["validation"] + ): + raise MailboxTransitionRejected("validation evidence is not passing on the candidate") + if not result["reviews"] or any( + item["outcome"] != "passed" or item["candidate_sha"] != candidate["head_sha"] + for item in result["reviews"] + ): + raise MailboxTransitionRejected("independent review evidence is not clean on the candidate") + required = [item for item in result["acceptance_evidence"] if item["required"]] + if any( + item["status"] != "pass" or item["candidate_sha"] != candidate["head_sha"] + for item in required + ): + raise MailboxTransitionRejected( + "required acceptance evidence is not candidate-bound and passing" + ) + + +def _delivered_candidate( + result: Mapping[str, Any], + current: Mapping[str, Any], +) -> dict[str, Any]: + value = copy.deepcopy(current) + value["pull_request"] = result["candidate"]["publication"]["pull_requests"][0]["url"] + return value + + +def _blocked_candidate( + result: Mapping[str, Any], claim: Mapping[str, Any], recorded_at: datetime +) -> dict[str, Any] | None: + if result["terminal_state"] == "requires_epic": + return copy.deepcopy(claim["candidate"]) + candidate = result["candidate"] + current = claim["candidate"] + if candidate is None: + if current is not None: + raise MailboxTransitionRejected("blocked result omitted an acknowledged candidate") + return None + candidate_identity = ( + candidate["repository"], + candidate["remote_url"], + candidate["remote_ref"], + candidate["base_sha"], + candidate["head_sha"], + ) + if current is not None and candidate_identity == ( + current["repository"], + current["remote_url"], + current["remote_ref"], + current["base_revision"], + current["head_revision"], + ): + return copy.deepcopy(current) + ledger = claim["checkpoint"]["authorizations"] + if ( + result["implementation_state"] != "published" + or not ledger + or ledger[-1]["phase"] != "pre_external_mutation" + or ledger[-1]["action"] != "repository.candidate.push" + or ledger[-1]["candidate_head"] != candidate["head_sha"] + or ledger[-1]["candidate_remote_ref"] != candidate["remote_ref"] + ): + raise MailboxTransitionRejected( + "blocked published candidate lacks its exact push authorization" + ) + return _mailbox_candidate(candidate, recorded_at) + + +def _require_tracker_transition(result: Mapping[str, Any], observation: Mapping[str, Any]) -> None: + transition = result["tracker_transition"] + issue = observation["issue"] + expected = ( + result["ticket"]["provider"], + str(issue["number"]), + "none", + issue["state"].lower(), + ) + actual = ( + transition["provider"], + str(transition["ticket_id"]), + transition["mode"], + transition["state"].lower(), + ) + if actual != expected or expected[3] != "open": + raise MailboxTransitionRejected( + "delegated result tracker transition does not match the current open ticket" + ) + + +def _require_current_invocation_contract( + invocation: Mapping[str, Any], state: Any +) -> None: + policy = state.effective_policy + current_acceptance = list(policy["acceptance"]["evidence"]) + invocation_acceptance = [ + item["criterion"] for item in invocation["acceptance_requirements"] + ] + if invocation["repository"]["identity"] != state.project["repository"]: + raise MailboxTransitionRejected( + "delegated invocation repository does not match the current project" + ) + if invocation["repository"]["base_sha"] != state.current_policy.commit: + raise MailboxTransitionRejected( + "delegated invocation base is stale against the current repository" + ) + if list(invocation["validation"]) != list(policy["validation"]["required_commands"]): + raise MailboxTransitionRejected( + "delegated invocation validation does not match current policy" + ) + if invocation_acceptance != current_acceptance: + raise MailboxTransitionRejected( + "delegated invocation acceptance does not match current policy" + ) + + +def _read_json(path: Path) -> dict[str, Any]: + value = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(value, dict): + raise DelegationError(f"{path}: expected a JSON object") + return value + + +def _parse_timestamp(value: str) -> datetime: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + if parsed.utcoffset() is None: + raise DelegationError("timestamps must include a UTC offset") + return parsed + + +def _policy_target(value: Mapping[str, Any]) -> PolicyTarget: + return PolicyTarget( + checkout=Path(value["checkout"]), + remote=value["remote"], + canonical_ref=value["canonical_ref"], + path=value["path"], + ) + + +def _host_target(value: Mapping[str, Any]) -> HostTarget: + return HostTarget( + descriptor_path=Path(value["descriptor_path"]), + skill_name=value["skill_name"], + skill_root=Path(value["skill_root"]), + connector=value["connector"], + operations=tuple(value["operations"]), + ) + + +def _fence(value: Mapping[str, Any]) -> ClaimFence: + return ClaimFence( + claim_id=value["claim_id"], + worker_run_id=value["worker_run_id"], + sequence=value["sequence"], + continuation_token=value["continuation_token"], + ) + + +def _policy_target_value(target: PolicyTarget) -> dict[str, Any]: + return { + "checkout": str(target.checkout.resolve()), + "remote": target.remote, + "canonical_ref": target.canonical_ref, + "path": target.path, + } + + +def _host_target_value(target: HostTarget) -> dict[str, Any]: + return { + "descriptor_path": str(target.descriptor_path.resolve()), + "skill_name": target.skill_name, + "skill_root": str(target.skill_root.resolve()), + "connector": target.connector, + "operations": list(target.operations), + } + + +def _compact_json(value: Any) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":")) + + +def _checkpoint_command( + *, + invocation_path: Path, + mailbox_remote: str, + mailbox_branch: str, + work_id: str, + approved_commit: str, + policy_target: PolicyTarget, + host_target: HostTarget, + observation_command: Sequence[str], +) -> list[str]: + return [ + sys.executable, + str(Path(__file__).resolve()), + "checkpoint-stdio", + "--invocation", + str(invocation_path.resolve()), + "--mailbox-remote", + mailbox_remote, + "--mailbox-branch", + mailbox_branch, + "--work-id", + work_id, + "--approved-commit", + approved_commit, + "--policy-target", + _compact_json(_policy_target_value(policy_target)), + "--host-target", + _compact_json(_host_target_value(host_target)), + "--observation-command", + _compact_json(list(observation_command)), + ] + + +def _write_json_atomic(path: Path, value: Mapping[str, Any]) -> None: + target = path.resolve() + target.parent.mkdir(parents=True, exist_ok=True) + descriptor, temporary_name = tempfile.mkstemp( + dir=target.parent, + prefix=f".{target.name}.", + ) + temporary = Path(temporary_name) + try: + with os.fdopen(descriptor, "w", encoding="utf-8") as handle: + json.dump(value, handle, indent=2, sort_keys=True) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary, target) + finally: + temporary.unlink(missing_ok=True) + + +def _probe_checkpoint_command(command: Sequence[str]) -> None: + completed = subprocess.run( + [*command, "--probe"], + text=True, + capture_output=True, + check=False, + ) + if completed.returncode != 0: + detail = completed.stderr.strip() or f"exit status {completed.returncode}" + raise DelegationError(f"checkpoint command probe failed: {detail}") + try: + response = json.loads(completed.stdout) + except json.JSONDecodeError as error: + raise DelegationError("checkpoint command probe returned malformed JSON") from error + if response != {"status": "ready"}: + raise DelegationError("checkpoint command probe did not report ready") + + +def _checkpoint_stdio_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Service one delegated checkpoint request") + parser.add_argument("--invocation", required=True, type=Path) + parser.add_argument("--mailbox-remote", required=True) + parser.add_argument("--mailbox-branch", required=True) + parser.add_argument("--work-id", required=True) + parser.add_argument("--approved-commit", required=True) + parser.add_argument("--policy-target", required=True) + parser.add_argument("--host-target", required=True) + parser.add_argument("--observation-command", required=True) + parser.add_argument("--probe", action="store_true") + return parser + + +def _json_object_argument(value: str, label: str) -> dict[str, Any]: + parsed = json.loads(value) + if not isinstance(parsed, dict): + raise DelegationError(f"{label} must be a JSON object") + return parsed + + +def _json_command_argument(value: str) -> list[str]: + parsed = json.loads(value) + if ( + not isinstance(parsed, list) + or not parsed + or any(not isinstance(item, str) or not item for item in parsed) + ): + raise DelegationError("observation command must be a nonempty JSON string array") + return parsed + + +def _checkpoint_stdio(arguments: Sequence[str]) -> dict[str, Any]: + args = _checkpoint_stdio_parser().parse_args(arguments) + policy_target = _policy_target(_json_object_argument(args.policy_target, "policy target")) + host_target = _host_target(_json_object_argument(args.host_target, "host target")) + observation_command = _json_command_argument(args.observation_command) + invocation = _read_json(args.invocation) + expected_command = _checkpoint_command( + invocation_path=args.invocation, + mailbox_remote=args.mailbox_remote, + mailbox_branch=args.mailbox_branch, + work_id=args.work_id, + approved_commit=args.approved_commit, + policy_target=policy_target, + host_target=host_target, + observation_command=observation_command, + ) + if invocation.get("checkpoint", {}).get("command") != expected_command: + raise DelegationError("checkpoint adapter arguments do not match the sealed invocation") + + read_boundary = datetime.now(UTC) + completed = subprocess.run( + observation_command, + text=True, + capture_output=True, + check=False, + ) + if completed.returncode != 0: + detail = completed.stderr.strip() or f"exit status {completed.returncode}" + raise DelegationError(f"native observation command failed: {detail}") + try: + observation = json.loads(completed.stdout) + except json.JSONDecodeError as error: + raise DelegationError("native observation command returned malformed JSON") from error + if not isinstance(observation, dict): + raise DelegationError("native observation command must return one JSON object") + observed_at = datetime.now(UTC) + with tempfile.TemporaryDirectory(prefix="atelier-checkpoint-observation-") as temporary: + observation_path = Path(temporary) / "observation.json" + _write_json_atomic(observation_path, observation) + validate_observation( + observation_path, + not_before=read_boundary, + now=observed_at, + ) + if args.probe: + return {"status": "ready"} + request = json.load(sys.stdin) + if not isinstance(request, dict): + raise DelegationError("checkpoint request must be one JSON object") + claims = ClaimCoordinator(args.mailbox_remote, args.mailbox_branch) + response = DelegationCoordinator(claims).checkpoint( + args.work_id, + invocation, + request, + approved_commit=args.approved_commit, + policy_target=policy_target, + host_target=host_target, + observation_path=observation_path, + observation_not_before=read_boundary, + recorded_at=observed_at, + next_continuation_token=secrets.token_urlsafe(32), + now=observed_at, + ) + return response + + +def execute_request(value: Mapping[str, Any]) -> dict[str, Any]: + """Execute one explicit host-owned delegation operation from a JSON request.""" + + claims = ClaimCoordinator(value["mailbox"]["remote"], value["mailbox"]["branch"]) + delegation = DelegationCoordinator(claims) + common = { + "approved_commit": value["approved_commit"], + "policy_target": _policy_target(value["policy_target"]), + "host_target": _host_target(value["host_target"]), + "observation_path": Path(value["observation_path"]), + "observation_not_before": _parse_timestamp(value["observation_not_before"]), + "now": _parse_timestamp(value["now"]) if value.get("now") else None, + } + operation = value["operation"] + if operation == "prepare": + return delegation.prepare( + value["work_id"], + _fence(value["fence"]), + checkpoint_invocation_path=Path(value["checkpoint_invocation_path"]), + observation_command=value["observation_command"], + **common, + ) + if operation == "checkpoint": + return delegation.checkpoint( + value["work_id"], + value["invocation"], + value["checkpoint_request"], + recorded_at=_parse_timestamp(value["recorded_at"]), + next_continuation_token=value["next_continuation_token"], + **common, + ) + if operation == "finalize": + result = delegation.finalize( + value["work_id"], + value["invocation"], + value["result"], + _fence(value["fence"]), + ended_at=_parse_timestamp(value["ended_at"]), + **common, + ) + return asdict(result) + raise DelegationError(f"unsupported delegation operation: {operation!r}") + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "request", type=Path, help="JSON request describing the host-owned operation" + ) + return parser + + +def main() -> int: + try: + if len(sys.argv) > 1 and sys.argv[1] == "checkpoint-stdio": + result = _checkpoint_stdio(sys.argv[2:]) + print(json.dumps(result, sort_keys=True)) + else: + result = execute_request(_read_json(_parser().parse_args().request)) + print(json.dumps(result, indent=2, sort_keys=True)) + except ( + DelegationError, + HostBoundaryError, + MailboxTransitionRejected, + KeyError, + OSError, + TypeError, + ValueError, + ) as error: + print(f"ERROR delegation: {error}", file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/skills/atelier/scripts/git_mailbox.py b/skills/atelier/scripts/git_mailbox.py index b2c7e67f..c42923aa 100644 --- a/skills/atelier/scripts/git_mailbox.py +++ b/skills/atelier/scripts/git_mailbox.py @@ -110,6 +110,7 @@ def __call__( Revalidate = Callable[[TransitionContext], None] PlanTransition = Callable[[TransitionContext], TransitionPlan] +ObserveMailbox = Callable[[TransitionContext], Any] def _valid_mailbox_document_path(path: PurePosixPath) -> bool: @@ -119,29 +120,23 @@ def _valid_mailbox_document_path(path: PurePosixPath) -> bool: if len(parts) == 3: collection, identifier, document = parts return ( - collection == "projects" - and identifier.startswith("prj_") - and document == "project.md" - ) or ( - collection == "initiatives" - and identifier.startswith("ini_") - and document == "initiative.md" - ) or ( - collection == "work" - and identifier.startswith("wrk_") - and document == "work.md" + ( + collection == "projects" + and identifier.startswith("prj_") + and document == "project.md" + ) + or ( + collection == "initiatives" + and identifier.startswith("ini_") + and document == "initiative.md" + ) + or (collection == "work" and identifier.startswith("wrk_") and document == "work.md") ) if len(parts) == 4 and parts[0] == "work" and parts[1].startswith("wrk_"): collection, document = parts[2:] return ( - collection == "messages" - and document.startswith("msg_") - and document.endswith(".md") - ) or ( - collection == "receipts" - and document.startswith("rcp_") - and document.endswith(".md") - ) + collection == "messages" and document.startswith("msg_") and document.endswith(".md") + ) or (collection == "receipts" and document.startswith("rcp_") and document.endswith(".md")) return False @@ -339,6 +334,30 @@ def publish( ) raise AssertionError("bounded mailbox write loop terminated unexpectedly") + def observe(self, operation: str, inspect: ObserveMailbox) -> Any: + """Inspect one fresh, isolated canonical snapshot without mutating it.""" + + if not operation.strip(): + raise ValueError("operation must not be empty") + with tempfile.TemporaryDirectory(prefix="atelier-mailbox-observe-") as temporary: + checkout = Path(temporary) / "mailbox" + self._initialize(checkout) + base_revision = self._fetch_current(checkout) + self._reset(checkout) + snapshot = reconstruct_mailbox(checkout) + self._require_canonical_branch(snapshot, operation=operation) + context_fingerprint = self._context_fingerprint(checkout) + result = inspect( + TransitionContext( + checkout=checkout, + base_revision=base_revision, + snapshot=snapshot, + attempt=1, + ) + ) + self._require_context_unchanged(checkout, expected=context_fingerprint) + return result + def recover(self, pending: PendingWrite) -> WriteResult | None: """Resolve an ambiguous push through ancestry and exact historical content.""" @@ -490,9 +509,7 @@ def _require_context_unchanged( "transition callback mutated its read-only context" ) from error if current != expected: - raise MailboxTransitionRejected( - "transition callback mutated its read-only context" - ) + raise MailboxTransitionRejected("transition callback mutated its read-only context") def _read_prior_claims( self, @@ -602,6 +619,17 @@ def _verify_claim_history( ) prior_checkpoint = prior_claim["checkpoint"] current_checkpoint = current_claim["checkpoint"] + if current_claim["invocation_digest"] != prior_claim["invocation_digest"]: + if not ( + prior_claim["invocation_digest"] is None + and isinstance(current_claim["invocation_digest"], str) + and document["status"] == "active" + and current_checkpoint == prior_checkpoint + and current_claim["candidate"] == prior_claim["candidate"] + ): + raise MailboxTransitionRejected( + f"{path}: delegated invocation binding is immutable" + ) prior_ledger = prior_checkpoint["authorizations"] current_ledger = current_checkpoint["authorizations"] sequence_delta = current_checkpoint["sequence"] - prior_checkpoint["sequence"] @@ -619,29 +647,152 @@ def _verify_claim_history( f"{path}: one checkpoint sequence requires one authorization entry" ) token_rotated = ( - current_checkpoint["continuation_token"] - != prior_checkpoint["continuation_token"] + current_checkpoint["continuation_token"] != prior_checkpoint["continuation_token"] ) if (sequence_delta == 1) != token_rotated: raise MailboxTransitionRejected( f"{path}: checkpoint sequence and continuation token must advance together" ) if current_claim["candidate"] != prior_claim["candidate"]: + if self._blocked_receipt_recovers_candidate( + checkout, + path, + document, + prior_claim, + current_claim, + changes, + ): + continue + if self._delivery_binds_pull_request( + checkout, + path, + document, + prior_claim, + current_claim, + changes, + ): + continue if sequence_delta != 1 or current_claim["candidate"] is None: raise MailboxTransitionRejected( f"{path}: candidate changes require one publication checkpoint" ) publication = current_ledger[-1] candidate_head = current_claim["candidate"]["head_revision"] + candidate_remote_ref = current_claim["candidate"]["remote_ref"] if ( publication["phase"] != "candidate_published" or publication["candidate_head"] != candidate_head + or publication["candidate_remote_ref"] != candidate_remote_ref or publication["acknowledged_candidate_head"] != candidate_head ): raise MailboxTransitionRejected( f"{path}: candidate publication checkpoint does not match candidate" ) + def _blocked_receipt_recovers_candidate( + self, + checkout: Path, + path: str, + document: Mapping[str, Any], + prior_claim: Mapping[str, Any], + current_claim: Mapping[str, Any], + changes: tuple[FileChange, ...], + ) -> bool: + """Allow a blocked receipt to recover one exact previously pushed candidate.""" + + candidate = current_claim["candidate"] + if ( + candidate is None + or candidate == prior_claim["candidate"] + or current_claim["checkpoint"] != prior_claim["checkpoint"] + or document["status"] != "blocked" + or document["delivery_receipt_id"] is not None + ): + return False + ledger = current_claim["checkpoint"]["authorizations"] + if ( + not ledger + or ledger[-1]["phase"] != "pre_external_mutation" + or ledger[-1]["action"] != "repository.candidate.push" + or ledger[-1]["candidate_head"] != candidate["head_revision"] + or ledger[-1]["candidate_remote_ref"] != candidate["remote_ref"] + ): + return False + receipt_id = document["attempt_receipt_id"] + if not isinstance(receipt_id, str): + return False + receipt_path = f"work/{PurePosixPath(path).parts[1]}/receipts/{receipt_id}.md" + if not any( + change.path == receipt_path and change.content is not None for change in changes + ): + return False + receipt, _ = _read_yaml( + checkout / receipt_path, + frontmatter=True, + label=receipt_path, + ) + return bool( + receipt["outcome"] == "blocked" + and receipt["claim_id"] == current_claim["id"] + and receipt["worker_run_id"] == current_claim["worker_run_id"] + and receipt["candidate"] == candidate + and receipt["mutation_ownership"] == "retained" + ) + + def _delivery_binds_pull_request( + self, + checkout: Path, + path: str, + document: Mapping[str, Any], + prior_claim: Mapping[str, Any], + current_claim: Mapping[str, Any], + changes: tuple[FileChange, ...], + ) -> bool: + """Allow delivery to bind the exact PR created under the last authorization.""" + + prior_candidate = prior_claim["candidate"] + current_candidate = current_claim["candidate"] + if prior_candidate is None or current_candidate is None: + return False + expected = dict(prior_candidate) + expected["pull_request"] = current_candidate["pull_request"] + if ( + prior_candidate["pull_request"] is not None + or not isinstance(current_candidate["pull_request"], str) + or not current_candidate["pull_request"].startswith("https://github.com/") + or current_candidate != expected + or current_claim["checkpoint"] != prior_claim["checkpoint"] + or document["status"] != "delivered" + or document["attempt_receipt_id"] != document["delivery_receipt_id"] + ): + return False + ledger = current_claim["checkpoint"]["authorizations"] + if ( + not ledger + or ledger[-1]["phase"] != "pre_external_mutation" + or ledger[-1]["action"] != "pull_request.create" + or ledger[-1]["candidate_head"] != current_candidate["head_revision"] + ): + return False + receipt_id = document["delivery_receipt_id"] + receipt_path = f"work/{PurePosixPath(path).parts[1]}/receipts/{receipt_id}.md" + if not any( + change.path == receipt_path and change.content is not None for change in changes + ): + return False + receipt, _ = _read_yaml( + checkout / receipt_path, + frontmatter=True, + label=receipt_path, + ) + return bool( + receipt["outcome"] == "delivered" + and receipt["claim_id"] == current_claim["id"] + and receipt["worker_run_id"] == current_claim["worker_run_id"] + and receipt["candidate"] == current_candidate + and receipt["mutation_ownership"] == "retained" + ) + def _verify_new_claim( self, checkout: Path, @@ -818,9 +969,7 @@ def _read_back(self, checkout: Path, pending: PendingWrite) -> bool: if ancestry.returncode == 1: return False if ancestry.returncode != 0: - raise MailboxReadBackError( - f"{pending.operation}: could not verify commit ancestry" - ) + raise MailboxReadBackError(f"{pending.operation}: could not verify commit ancestry") for change in pending.changes: shown = self._run(checkout, ("show", f"{pending.commit}:{change.path}")) if change.content is None: @@ -850,9 +999,7 @@ def _require_canonical_descendant( f"{operation}: canonical mailbox branch moved backwards or diverged" ) if ancestry.returncode != 0: - raise MailboxReadBackError( - f"{operation}: could not verify canonical branch continuity" - ) + raise MailboxReadBackError(f"{operation}: could not verify canonical branch continuity") def _output(self, cwd: Path, arguments: Sequence[str], purpose: str) -> str: result = self._run(cwd, arguments) diff --git a/skills/atelier/scripts/host_boundary.py b/skills/atelier/scripts/host_boundary.py index 9ab2c140..9ccf7fa9 100644 --- a/skills/atelier/scripts/host_boundary.py +++ b/skills/atelier/scripts/host_boundary.py @@ -216,7 +216,7 @@ def load_descriptor(path: Path = DEFAULT_DESCRIPTOR) -> dict[str, Any]: raise HostBoundaryError("reference host must be codex") if ( descriptor["delegated_capability"] - != "agent-scripts.implement-ticket/delegated-execution/v1" + != "agent-scripts.implement-ticket/delegated-execution/v2" ): raise HostBoundaryError("delegated capability identifier is incompatible") if descriptor["native_state_access"] != "read-only": diff --git a/skills/atelier/scripts/mailbox.py b/skills/atelier/scripts/mailbox.py index 8375396c..f80236d5 100644 --- a/skills/atelier/scripts/mailbox.py +++ b/skills/atelier/scripts/mailbox.py @@ -920,16 +920,35 @@ def _validate_claim( and attempt_receipt["handoff"] == "transferable" and attempt_receipt["claim_id"] != claim["id"] ) + recovered_current_candidate = ( + attempt_receipt is not None + and attempt_receipt["outcome"] == "blocked" + and attempt_receipt["handoff"] == "transferable" + and attempt_receipt["claim_id"] == claim["id"] + and attempt_receipt["candidate"] == claim["candidate"] + and bool(ledger) + and ledger[-1]["phase"] == "pre_external_mutation" + and ledger[-1]["action"] == "repository.candidate.push" + and claim["candidate"] is not None + and ledger[-1]["candidate_head"] == claim["candidate"]["head_revision"] + and ledger[-1]["candidate_remote_ref"] == claim["candidate"]["remote_ref"] + ) latest_acknowledged_head = ( attempt_receipt["candidate"]["head_revision"] if transferable_handoff and attempt_receipt["candidate"] is not None else None ) + latest_acknowledged_ref = ( + attempt_receipt["candidate"]["remote_ref"] + if transferable_handoff and attempt_receipt["candidate"] is not None + else None + ) previous_entry: dict[str, Any] | None = None for entry in ledger: action = entry["action"] phase = entry["phase"] candidate_head = entry["candidate_head"] + candidate_remote_ref = entry["candidate_remote_ref"] acknowledged_head = entry["acknowledged_candidate_head"] if action not in authority_ceiling: diagnostics.append( @@ -945,6 +964,7 @@ def _validate_claim( and previous_entry["phase"] == "pre_external_mutation" and previous_entry["action"] == "repository.candidate.push" and previous_entry["candidate_head"] == candidate_head + and previous_entry["candidate_remote_ref"] == candidate_remote_ref ) if not paired_push: diagnostics.append( @@ -958,6 +978,7 @@ def _validate_claim( if ( action != "repository.candidate.push" or candidate_head is None + or candidate_remote_ref is None or acknowledged_head != candidate_head ): diagnostics.append( @@ -969,6 +990,7 @@ def _validate_claim( ) elif paired_push: latest_acknowledged_head = candidate_head + latest_acknowledged_ref = candidate_remote_ref else: if acknowledged_head is not None: diagnostics.append( @@ -978,18 +1000,23 @@ def _validate_claim( "pre_external_mutation must not acknowledge a candidate", ) ) - if action in CANDIDATE_REQUIRED_ACTIONS and candidate_head is None: + if action in CANDIDATE_REQUIRED_ACTIONS and ( + candidate_head is None or candidate_remote_ref is None + ): diagnostics.append( Diagnostic( path, "checkpoint-candidate", - f"authorization action {action!r} requires an exact candidate head", + f"authorization action {action!r} requires an exact candidate head and remote ref", ) ) elif ( action in CANDIDATE_REQUIRED_ACTIONS and action != "repository.candidate.push" - and candidate_head != latest_acknowledged_head + and ( + candidate_head != latest_acknowledged_head + or candidate_remote_ref != latest_acknowledged_ref + ) ): diagnostics.append( Diagnostic( @@ -1004,21 +1031,24 @@ def _validate_claim( publication_entries = [entry for entry in ledger if entry["phase"] == "candidate_published"] internally_invalid = any( entry["candidate_head"] is None + or entry["candidate_remote_ref"] is None or entry["acknowledged_candidate_head"] != entry["candidate_head"] for entry in publication_entries ) inherited_candidate = ( candidate is not None and not publication_entries - and transferable_handoff + and (transferable_handoff or recovered_current_candidate) and attempt_receipt["candidate"] == candidate ) current_unacknowledged = ( candidate is not None and not inherited_candidate + and not recovered_current_candidate and ( not publication_entries or publication_entries[-1]["candidate_head"] != candidate["head_revision"] + or publication_entries[-1]["candidate_remote_ref"] != candidate["remote_ref"] ) ) discarded_handoff = transferable_handoff and (