fix(orchestration): select terminal acceptance repository

Author: Ldsystem

scripts/orchestration/plans.py

@@ -164,6 +164,16 @@ def _git_tree_id(root: Path, spec: str) -> str | None:
     return result.stdout.strip() or None
 
 
+def _git_is_ancestor(root: Path, ancestor: str, descendant: str) -> bool:
+    result = subprocess.run(
+        ["git", "-C", str(root), "merge-base", "--is-ancestor", ancestor, descendant],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    return result.returncode == 0
+
+
 def _handoff_recorded_identities(handoff: dict[str, object]) -> list[str]:
     identities: list[str] = []
     review = handoff.get("acceptance_review") if isinstance(handoff.get("acceptance_review"), dict) else {}
@@ -206,8 +216,9 @@ def _verified_handoff_tree(root: Path, handoff: dict[str, object]) -> str | None
 def _material_repository_root(
     args: argparse.Namespace,
     validated: list[tuple[dict[str, object], dict[str, object]]],
+    commands: list[str],
 ) -> Path:
-    roots: set[Path] = set()
+    entries: list[tuple[Path, str]] = []
     for handoff, brief in validated:
         if not _handoff_has_material_changes(handoff, brief):
             continue
@@ -216,13 +227,44 @@ def _material_repository_root(
             if not isinstance(repository, dict):
                 continue
             recorded = str(repository.get("root") or "").strip()
-            if recorded:
-                roots.add(Path(recorded).expanduser().resolve())
-    if len(roots) > 1:
+            metadata = repository.get("metadata") if isinstance(repository.get("metadata"), dict) else {}
+            identity = str(metadata.get("actual_commit") or "").strip()
+            if recorded and identity:
+                entries.append((Path(recorded).expanduser().resolve(), identity))
+    if not entries:
+        return project_root(args)
+    acceptance_entries: list[tuple[Path, str]] = []
+    for handoff, _brief in validated:
+        if not any(_handoff_command_result(handoff, command) == "passed" for command in commands):
+            continue
+        repositories = handoff.get("repository") if isinstance(handoff.get("repository"), list) else []
+        for repository in repositories:
+            if not isinstance(repository, dict):
+                continue
+            recorded = str(repository.get("root") or "").strip()
+            metadata = repository.get("metadata") if isinstance(repository.get("metadata"), dict) else {}
+            identity = str(metadata.get("actual_commit") or "").strip()
+            if recorded and identity:
+                acceptance_entries.append((Path(recorded).expanduser().resolve(), identity))
+    fresh_acceptance_roots = {
+        root
+        for root, identity in acceptance_entries
+        if _git_tree_id(root, "HEAD") == _git_tree_id(root, identity)
+    }
+    if len(fresh_acceptance_roots) == 1:
+        return next(iter(fresh_acceptance_roots))
+    terminal: list[tuple[Path, str]] = []
+    identities = {identity for _root, identity in entries}
+    for root, identity in entries:
+        if _git_tree_id(root, identity) and all(_git_is_ancestor(root, other, identity) for other in identities):
+            terminal.append((root, identity))
+    terminal_identities = {identity for _root, identity in terminal}
+    if len(terminal_identities) == 1:
+        identity = next(iter(terminal_identities))
+        return next(root for root, candidate in terminal if candidate == identity)
+    if len({root for root, _identity in entries}) > 1:
         raise SystemExit("acceptance-blocked: final plan repository is ambiguous")
-    if roots:
-        return next(iter(roots))
-    return project_root(args)
+    return entries[0][0]
 
 
 def _acceptance_result_detail(results: set[str]) -> str:
@@ -313,7 +355,7 @@ def _assert_archive_plan_acceptance(
     commands = _declared_integration_commands(body)
     if not commands:
         return
-    git_root = _material_repository_root(args, validated)
+    git_root = _material_repository_root(args, validated, commands)
     terminal_tree = _git_tree_id(git_root, "HEAD")
     material = [pair for pair in validated if _handoff_has_material_changes(*pair)]
     for command in commands:

tests/test_orchestration_workflow_contracts.py

@@ -19,7 +19,7 @@
     validate_executor_result_for_task,
 )
 from handoffs import cmd_write_handoff, index_handoffs
-from plans import _verified_handoff_tree
+from plans import _material_repository_root, _verified_handoff_tree
 
 
 def read(path: str) -> str:
@@ -117,6 +117,49 @@ def test_handoff_tree_resolves_recorded_repository_instead_of_control_root(tmp_p
     assert _verified_handoff_tree(control_root, handoff) == tree
 
 
+def test_material_repository_prefers_fresh_plan_acceptance_root(tmp_path: Path) -> None:
+    from test_orchestration_execution_context import git
+
+    control_root = tmp_path / "control"
+    earlier_root = tmp_path / "earlier"
+    accepted_root = tmp_path / "accepted"
+    control_root.mkdir()
+    for root, content in ((earlier_root, "old\n"), (accepted_root, "accepted\n")):
+        root.mkdir()
+        git(root, "init", "-q")
+        git(root, "config", "user.email", "test@example.com")
+        git(root, "config", "user.name", "Test")
+        (root / "feature.ts").write_text(content, encoding="utf-8")
+        git(root, "add", ".")
+        git(root, "commit", "-qm", "feature")
+
+    command = "pnpm run ci"
+    earlier_head = git(earlier_root, "rev-parse", "HEAD").strip()
+    accepted_head = git(accepted_root, "rev-parse", "HEAD").strip()
+    validated = [
+        (
+            {
+                "changes": {"files": [{"path": "feature.ts", "action": "modified"}]},
+                "repository": [{"root": str(earlier_root), "metadata": {"actual_commit": earlier_head}}],
+                "validation": {"commands": []},
+            },
+            {"files": {"write": ["feature.ts"]}},
+        ),
+        (
+            {
+                "changes": {"files": [{"path": "feature.ts", "action": "modified"}]},
+                "repository": [{"root": str(accepted_root), "metadata": {"actual_commit": accepted_head}}],
+                "validation": {"commands": [{"command": command, "result": "passed"}]},
+            },
+            {"files": {"write": ["feature.ts"]}},
+        ),
+    ]
+
+    assert _material_repository_root(
+        argparse.Namespace(project_root=str(control_root)), validated, [command]
+    ) == accepted_root.resolve()
+
+
 def test_write_handoff_fills_missing_task_plan_from_authorized_args(tmp_path: Path) -> None:
     content = tmp_path / "handoff-content.txt"
     content.write_text(