Repair source publication observation and same-carrier recapture

pyproject.toml

@@ -30,12 +30,12 @@ dependencies = [
     "nominal-refactor-advisor @ git+https://github.com/OpenHCSDev/NominalRefactorAdvisor@main",
     "packaging>=24",
     "pydantic>=2.13",
+    "pytest>=7.0",
     "python-introspect @ git+https://github.com/OpenHCSDev/python-introspect@main",
 ]
 
 [project.optional-dependencies]
 dev = [
-    "pytest>=7.0",
     "pytest-cov>=4.0",
     "pytest-asyncio>=0.23",
     "pytest-xdist>=3.6",

src/agent_comms/head_review.py

@@ -1,3 +1,3 @@
 """A reviewer's verdict on one exact pull-request head, posted as main's required status.
 
 Every path that posts that status goes through here.
@@ -11,6 +11,7 @@
 from collections.abc import Sequence
 from dataclasses import dataclass, field
 from pathlib import Path
+from time import monotonic, sleep
 from typing import TYPE_CHECKING, Annotated, ClassVar, Self
 
 from agent_comms_hooks import ToolRefusedError
@@ -134,6 +135,23 @@
             )
         return current
 
+    def published_after(self, previous: PullRequestHead) -> OpenPullRequestHead:
+        """Observe a known successful push; only its previous head may lag briefly."""
+        if not self.same_pull_request(previous):
+            raise ToolRefusedError("Publication observations must name the same pull request")
+        deadline = monotonic() + 5
+        current = self.at(self.repo, self.pr)
+        while not current.holds(self.head):
+            if not current.holds(previous.head) or monotonic() >= deadline:
+                raise ToolRefusedError(
+                    f"Published #{self.pr} expected {self.head}; forge observed {current.head}; "
+                    f"previous head {previous.head}. Preserve this attempt and resume it; "
+                    "do not repeat creation."
+                )
+            sleep(min(0.1, max(0, deadline - monotonic())))
+            current = self.at(self.repo, self.pr)
+        return current
+
     def closed_or_open(self) -> ReadPullRequestHead:
         """GitHub's read in whatever state the PR is in, refused unless this is still its head:
         a closed PR's head can only be dropped, so nothing here may arm or integrate it."""
@@ -465,8 +483,14 @@
 
     def validate(self, repositories: tuple[Repo, ...], carrier: PullRequestHead) -> None:
         """Git and forge prove this frozen input; Todo and verdict owners decide permission."""
+        self._validate_current(repositories, carrier, carrier.current())
+
+    def _validate_current(
+        self, repositories: tuple[Repo, ...], carrier: PullRequestHead,
+        current: OpenPullRequestHead,
+    ) -> None:
+        """One observed forge cut and the expected immutable capture answer different questions."""
         core = Repo.selected(repositories, carrier.repo.to_wire())
-        current = carrier.current()
         if current.base != WorkBranches.main:
             raise ToolRefusedError("The integration carrier no longer targets main")
         self.require_carrier(core, carrier)
@@ -488,6 +512,10 @@
         """An ordinary source record cannot become a combined publication."""
         raise ToolRefusedError("This input is not a resumable combined publication")
 
+    def require_recapture(self, core: Repo, carrier: PullRequestHead) -> None:
+        """A completed capture supplies the default recapture proof."""
+        self.require_carrier(core, carrier)
+
     def require_source_reviews(self) -> None:
         """Native promotion and the hosted gate require these same exact source decisions."""
         for source in self.sources:
@@ -628,6 +656,15 @@
     ) -> CombinedInput:
         return run.capture(context, self, carrier)
 
+    def require_recapture(self, core: Repo, carrier: PullRequestHead) -> None:
+        """A known published prospective cut may be replaced explicitly, never recreated."""
+        self.require_repositories(
+            self.declared_repositories(core, self.local.baseline.commit, carrier.head)
+        )
+        self.bind(core, carrier)
+        if self.from_message(WorkBranches(core.checkout).commit_message(carrier.head)) != (self,):
+            raise ToolRefusedError("The published prospective cut does not contain this exact input")
+
     def bind(self, repository: Repo, carrier: PullRequestHead) -> CombinedInput:
         """Bind the actual publication without substituting another source tree or history."""
         if carrier.repo != self.predecessor.repository:
@@ -680,6 +717,9 @@
         run.retain_capture(self, commit)
         return self
 
+    def validate_published(self, repositories: tuple[Repo, ...], carrier: PullRequestHead) -> None:
+        self._validate_current(repositories, carrier, carrier.published_after(self.carrier))
+
     def require_local_candidate(self, core: Repo, carrier: PullRequestHead) -> None:
         self.local.require_origin(core)
         self.local.require_heads(carrier)

src/agent_comms/integration.py

@@ -1,3 +1,3 @@
 """Combine reviewed heads and spend the host's test lease once, not once per PR."""
 
 from __future__ import annotations
@@ -9,17 +9,19 @@
 import sys
 import tomllib
 from abc import ABC, abstractmethod
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
 from pathlib import Path
 from typing import Annotated, ClassVar, Self
 
+from _pytest.config import get_config
+
 from .active_route import resolve_comms_route
 from .build_identity import RepositoryCommit
 from .child_process import STOP_GRACE_SECONDS, ResidentManager, SpawnedProcess
 from .team.order import TeamContext
 from .nominal.declared_family import DeclaredFamily
 from agent_comms_hooks import ToolRefusedError
-from .nominal.field_codec import FieldCodec, GitCommitText, PathText, WireName, boundary_decoder
+from .nominal.field_codec import FieldCodec, GitCommitText, GoalId, PathText, WireName, boundary_decoder
 from .head_review import (
     BaseSide,
     CombinedInput,
@@ -46,7 +48,7 @@
 from .repository_name import RepositoryName
 from .systemd_unit import AgentsSlice
 from .test_lease import LeaseStart, TestLease
-from .todos import CarriedHeadVerdict, PullRequestLink, PullRequests, RecordPullRequest, StoredHeadReview, VerdictReason
+from .todos import CarriedHeadVerdict, PullRequestLink, PullRequests, RecordPullRequest, StoredHeadReview, Todo, VerdictReason
 from .tool_caller import PeerCaller
 from .work_branches import WorkBranches
 
@@ -62,16 +64,19 @@
 
     @classmethod
     def read(cls, checkout: Path) -> TestSelection:
-        # Decode only this consumer's declared fields; pytest owns the other options.
+        # Register pytest's built-in declarations without loading project plugins,
+        # conftests or collection. Missing options belong to those declarations.
+        defaults = get_config()
         with (checkout / "pyproject.toml").open("rb") as source:
             match tomllib.load(source):
                 case {"tool": {"pytest": {"ini_options": {**configuration}}}}:
                     return FieldCodec.decode(
                         cls,
                         {
-                            entry.key: configuration[entry.key]
+                            entry.key: configuration.get(
+                                entry.key, defaults.getini(entry.key)
+                            )
                             for entry in FieldCodec.argument_fields(cls)
-                            if entry.key in configuration
                         },
                     )
         raise IntegrationFailed("pyproject.toml requires [tool.pytest.ini_options]")
@@ -82,7 +87,7 @@
                 {
                     str(path)
                     for path in changed
-                    if any(path.is_relative_to(root) for root in self.directories)
+                    if any(path.is_relative_to(root) for root in self.directories or (".",))
                     and any(path.match(pattern) for pattern in self.patterns)
                     and (checkout / path).is_file()
                 }
@@ -189,6 +194,8 @@
         owner = context.comms.registry.require(context.actor_name)
         context.comms.registry.storage_rule.require_owned_checkout(
             owner, repository.checkout, repository.github, RepositoryReads())
+        snapshot = context.comms.registry.snapshot()
+        snapshot.commander_project.require_scope(snapshot, context.actor_identity, repository.github)
         return repository
 
     def publication_branch(self) -> str:
@@ -206,13 +213,20 @@
         fetched = Repo.git(self.path, "rev-parse", "FETCH_HEAD").strip()
         if fetched != head.head:
             raise IntegrationFailed(f"#{head.pr} moved while fetching; capture its new head")
+        self.merge_revision(head.head)
+
+    def merge_revision(self, commit: str) -> None:
+        """The integration checkout owns merging and conflict cleanup for members and main."""
         try:
-            Repo.git(self.path, "merge", "--no-ff", "--no-edit", head.head, timeout=300)
+            Repo.git(self.path, "merge", "--no-ff", "--no-edit", commit, timeout=300)
         except GitCommandError:
             if Repo.answers(self.path, "rev-parse", "-q", "--verify", "MERGE_HEAD"):
                 Repo.git(self.path, "merge", "--abort")
             raise
 
+    def prepare_main(self, base: str) -> None:
+        """Ordinary preparation preserves its candidate; recapture can integrate main."""
+
     def combine_sources(self, base: str, heads: tuple[PullRequestHead, ...]) -> str:
         """Checking preserves every source member, including when their merge conflicts."""
         self.require_editable()
@@ -250,6 +264,20 @@
         return TestSelection.read(self.path).files(self.path, changed)
 
 
+@dataclass(frozen=True)
+class RecaptureCheckout(IntegrationCheckout):
+    """Integrate current main only after the shared source preparation checks."""
+
+    context: TeamContext
+    primary: Todo
+
+    def prepare_main(self, base: str) -> None:
+        self.owned_repository(self.context)
+        current = self.context.comms.goals.todos.get(self.primary.id)
+        current.require_assignee(self.context.actor_identity, self.primary.held().generation)
+        self.merge_revision(base)
+
+
 @dataclass(frozen=True)
 class TestAttempt:
     log: Path
@@ -830,16 +858,18 @@
                                    for excluded in excluded_carriers)}.values(),
                        key=lambda head: head.pr)
             )
-        self.output.mkdir(parents=True, exist_ok=False)
-        IntegrationCheckout.create(
-            core, BaseSide.checkout(self.output, core.github), base
-        )
-        tree = Repo.git(self.checkout.path, "rev-parse", "HEAD^{tree}").strip()
         resolved = {companion.repo: companion.identity for companion in companions}
         if len(resolved) != len(companions) or set(resolved) - companion_repositories.keys():
             raise IntegrationFailed(
                 "Select at most one actual carrier per declared companion repository"
             )
+        self.output.mkdir(parents=True, exist_ok=False)
+        self.checkout.prepare_main(base)
+        candidate_commit = Repo.git(self.checkout.path, "rev-parse", "HEAD").strip()
+        IntegrationCheckout.create(
+            core, BaseSide.checkout(self.output, core.github), base
+        )
+        tree = Repo.git(self.checkout.path, "rev-parse", "HEAD^{tree}").strip()
         sources: list[IntegrationSource] = [
             LocalIntegrationSource(RepositoryCommit(core.github, base), heads[core.github], tree)
         ]
@@ -930,10 +960,7 @@
                                     excluded_carriers=companions)
             primary = store.get(primary.id)
             primary.require_assignee(actor, generation)
-            prepared.write(self.output)
-            Repo.git(self.checkout.path, "commit", "--allow-empty", "-m",
-                     f"Prepare the actual source cut\n\n{prepared.trailer}")
-            Repo.git(self.checkout.path, "push", "-u", "origin", branch)
+            self.prepare_publication(prepared, branch)
             primary = store.get(primary.id)
             primary.require_assignee(actor, generation)
             link = primary.order.checked().open_pull_request(
@@ -942,16 +969,78 @@
                 store, actor)
             carrier = PullRequestHead.at(link.repo, link.number)
             inputs = prepared.resume_publication(self, context, carrier)
-        primary = store.get(primary.id)
-        primary.require_assignee(actor, generation)
+        return self.finish_publication(context, primary.id, generation, carrier, branch, inputs)
+
+    def prepare_publication(self, prepared: UnpublishedInput, branch: str) -> None:
+        """A durable prospective cut precedes both first publication and explicit recapture."""
+        prepared.write(self.output)
+        Repo.git(self.checkout.path, "commit", "--allow-empty", "-m",
+                 f"Prepare the actual source cut\n\n{prepared.trailer}")
+        Repo.git(self.checkout.path, "push", "-u", "origin", branch)
+
+    def finish_publication(
+        self, context: TeamContext, problem: str, generation: GoalId,
+        carrier: PullRequestHead, branch: str, inputs: CombinedInput,
+    ) -> CombinedInput:
+        """Link and validate the capture we actually pushed, not a lagging forge head."""
+        repository = self.checkout.owned_repository(context)
+        store = context.comms.goals.todos
+        primary = store.get(problem)
+        primary.require_assignee(context.actor_identity, generation)
         link = PullRequestLink(repository.github, carrier.pr)
-        store.apply(primary.id, RecordPullRequest.for_order(primary, actor, link))
+        store.apply(primary.id, RecordPullRequest.for_order(primary, context.actor_identity, link))
         Repo.git(self.checkout.path, "push", "origin", branch)
-        current = PullRequestHead.at(repository.github, carrier.pr)
-        inputs.validate(context.repos, current)
+        captured = PullRequestHead(repository.github, carrier.pr,
+                                   Repo.git(self.checkout.path, "rev-parse", "HEAD").strip())
+        inputs.validate_published(context.repos, captured)
         print(f"PUBLISHED {link.url}; qualification only; no merge permission", flush=True)
         return inputs
 
+    def recapture(
+        self, context: TeamContext, problems: tuple[str, ...], *, carrier: PullRequestHead,
+        companions: tuple[PullRequestHead, ...] = (),
+    ) -> CombinedInput:
+        """Explicitly replace a changed cut on its real carrier; preserve the old capture."""
+        if not problems:
+            raise IntegrationFailed("Recapture requires an actual primary problem")
+        repository = self.checkout.owned_repository(context)
+        branch = self.checkout.publication_branch()
+        store = context.comms.goals.todos
+        primary = store.get(problems[0])
+        generation = primary.held().generation
+        primary.require_assignee(context.actor_identity, generation)
+        current = carrier.current()
+        matching = tuple(row for row in PullRequests().rows(repository.github, (current.pr,))
+                         if row.branch == branch)
+        if len(matching) != 1:
+            raise IntegrationFailed("Recapture requires the real PR on this owned publication branch")
+        commit = Repo.git(self.checkout.path, "rev-parse", "HEAD").strip()
+        if current.repo != repository.github or not Repo.descends(
+            self.checkout.path, current.head, commit
+        ):
+            raise IntegrationFailed("Recapture must preserve the same real carrier's published history")
+        message = WorkBranches(self.checkout.path).commit_message(current.head)
+        recorded = (*IntegrationInput.from_message(message), *UnpublishedInput.from_message(message))
+        if len(recorded) != 1:
+            raise IntegrationFailed("Recapture requires one existing published source cut")
+        (previous,) = recorded
+        previous.require_recapture(repository, carrier)
+        primary = store.get(primary.id)
+        primary.require_assignee(context.actor_identity, generation)
+        prepared = replace(self, checkout=RecaptureCheckout(self.checkout.path, context, primary)).prepare(
+            context, problems, companions=companions,
+            excluded_carriers=(carrier, *companions))
+        primary = store.get(primary.id)
+        primary.require_assignee(context.actor_identity, generation)
+        self.prepare_publication(prepared, branch)
+        prospective = PullRequestHead(repository.github, carrier.pr,
+                                      Repo.git(self.checkout.path, "rev-parse", "HEAD").strip())
+        observed = prospective.published_after(current)
+        primary = store.get(primary.id)
+        primary.require_assignee(context.actor_identity, generation)
+        inputs = prepared.resume_publication(self, context, observed)
+        return self.finish_publication(context, primary.id, generation, observed, branch, inputs)
+
     def qualify(
         self,
         context: TeamContext,
@@ -1162,6 +1251,10 @@
         "--publish", default="", metavar="PROBLEM",
         help="prepare the actual source cut and create its first real carrier once",
     )
+    operation.add_argument(
+        "--recapture", default="", metavar="PROBLEM",
+        help="explicitly replace a changed source cut on the same existing captured carrier",
+    )
     parser.add_argument(
         "--workflow-output",
         type=Path,
@@ -1249,7 +1342,7 @@
         tuple(pytest_arguments),
     )
     # Decode foreign operations once; source publication never enters execute/test/lease.
-    if arguments.qualify or arguments.publish:
+    if arguments.qualify or arguments.publish or arguments.recapture:
         if pytest_arguments:
             parser.error("Source publication and qualification cannot run tests")
         context = TeamContext.read_root(resolve_comms_route(arguments.comms_root).root)
@@ -1262,7 +1355,7 @@
         selected = tuple(head.identity for head in companions)
         if arguments.publish:
             if arguments.carrier:
-                parser.error("First publication has no prior --carrier; resume with --qualify")
+                parser.error("First publication has no prior --carrier; resume the same --publish")
             if not arguments.callers or not arguments.title or not arguments.body_files:
                 parser.error("--publish requires --from, --title and --body-file")
             run.publish(
@@ -1274,16 +1367,22 @@
             if arguments.title or arguments.body_files:
                 parser.error("--title and --body-file belong to first --publish")
             if arguments.carrier <= 0:
-                parser.error("Qualification requires --carrier naming the actual separate combined PR")
+                parser.error("Qualification or recapture requires the actual separate --carrier PR")
             carrier = PullRequestHead.at(Repo.of(run.checkout.path).github, arguments.carrier)
-            run.qualify(
-                context, (arguments.qualify, *arguments.with_problem),
-                carrier=carrier.identity,
-                companions=selected,
-            )
+            if arguments.recapture:
+                if not arguments.callers:
+                    parser.error("Explicit recapture requires --from naming the actual holder")
+                run.recapture(context, (arguments.recapture, *arguments.with_problem),
+                              carrier=carrier.identity, companions=selected)
+            else:
+                run.qualify(
+                    context, (arguments.qualify, *arguments.with_problem),
+                    carrier=carrier.identity,
+                    companions=selected,
+                )
         return 0
     if any(values[option.dest] for option in source_options._group_actions):
-        parser.error("Source selection/publication options require --qualify or --publish")
+        parser.error("Source selection/publication options require a publication operation")
     return run.execute()