diff --git a/src/agent_comms/change_report.py b/src/agent_comms/change_report.py index 7e4213aa6c..c46ea09c6b 100644 --- a/src/agent_comms/change_report.py +++ b/src/agent_comms/change_report.py @@ -240,6 +240,10 @@ class CitedRepository(DeclaredFamily, ABC, affix="Repository"): def source(self, roots: SourceRoots, revision: SourceText) -> tuple[Path, SourceText]: """The checkout and revisions this citation's lines are read from.""" + def named_repositories(self) -> tuple[RepositoryName, ...]: + """Unqualified locations add no repository beyond the problem's primary.""" + return () + @abstractmethod def qualified(self, location: str) -> str: ... @@ -266,6 +270,9 @@ def source(self, roots: SourceRoots, revision: SourceText) -> tuple[Path, Source return (Repo.selected(roots.project.require().repos, self.name.to_wire()).checkout, MainRevision()) + def named_repositories(self) -> tuple[RepositoryName, ...]: + return (self.name,) + def qualified(self, location: str) -> str: return f"{self.name.to_wire()}:{location}" diff --git a/src/agent_comms/commander/vibe_check.py b/src/agent_comms/commander/vibe_check.py index ee29c0b04f..0e69ee42d0 100644 --- a/src/agent_comms/commander/vibe_check.py +++ b/src/agent_comms/commander/vibe_check.py @@ -141,7 +141,8 @@ def from_frame(cls, comms: Comms, commander: str, frame: HeartbeatFrame) -> Self frame.todos, len(tuple(ReviewTimerUp.conditions(comms, frame))), sum(read.read() for read in OrderEditCountRead.judgeable( - comms.root, frame.live, frame.pulls.progress(frame.todos, comms.registry))), + comms.root, frame.live, frame.pulls.progress(frame.todos, comms.registry), + frame.workplaces)), ) @classmethod diff --git a/src/agent_comms/commander_hud.py b/src/agent_comms/commander_hud.py index 12b5312361..8168553d38 100644 --- a/src/agent_comms/commander_hud.py +++ b/src/agent_comms/commander_hud.py @@ -291,7 +291,8 @@ def of(cls, comms: Comms, commander: str, frame: HeartbeatFrame, installed = InstalledBuild.read() edits = dict.fromkeys((todo.id for todo in held), 0) | { read.todo.id: read.read() - for read in OrderEditCountRead.judgeable(comms.root, frame.live, progress) + for read in OrderEditCountRead.judgeable( + comms.root, frame.live, progress, frame.workplaces) } units = [] for state in states: diff --git a/src/agent_comms/heartbeat_frame.py b/src/agent_comms/heartbeat_frame.py index eb3d6f03e8..b6f7e48719 100644 --- a/src/agent_comms/heartbeat_frame.py +++ b/src/agent_comms/heartbeat_frame.py @@ -12,6 +12,8 @@ from agent_comms.plan_phases import PlanPhase from agent_comms.repo import Repo from agent_comms.standing_rules import OwedAnswer +from agent_comms.thread_identity import ThreadIncarnation +from agent_comms.turn_digests import EditWorkplaces from agent_comms.todos import ( ClosedHeadReading, HeadReviewRecord, ProgressRead, StoredClosedHead, StoredHeadReview, ) @@ -30,6 +32,7 @@ class HeartbeatFrame(HeldOrders): answers: tuple[OwedAnswer, ...] closed: ClosedHeadReading head_reviews: Mapping[PullRequestHead, tuple[HeadReviewRecord, ...]] + workplaces: Mapping[ThreadIncarnation, EditWorkplaces] @classmethod def capture(cls, comms: Comms, commander: str, pulls: ProgressRead, @@ -43,16 +46,19 @@ def capture(cls, comms: Comms, commander: str, pulls: ProgressRead, closed = StoredClosedHead.current(db, project.repository_names()) reviews = StoredHeadReview.head_reviews_among( db, (head.request for head in closed.lost)) + plan = plans.read( + repository, + comms.goals.todos.history_for_commander(snapshot.require(commander).incarnation), + tuple(phase.phase_evidence() for phase in PlanPhase.members_with(PlanPhase)), + ) + pulls.register(scan.todos) return cls( scan.live, scan.orphaned, - pulls.register(scan.todos), - plans.read( - repository, - comms.goals.todos.history_for_commander(snapshot.require(commander).incarnation), - tuple(phase.phase_evidence() for phase in PlanPhase.members_with(PlanPhase)), - ), + pulls, + plan, OwedAnswer.read_scan(comms, commander, scan), closed, reviews, + EditWorkplaces.capture(comms, snapshot, scan.live, plan.todos, pulls), ) diff --git a/src/agent_comms/storage_rules.py b/src/agent_comms/storage_rules.py index 5f53c5c0e5..6164449fa1 100644 --- a/src/agent_comms/storage_rules.py +++ b/src/agent_comms/storage_rules.py @@ -77,7 +77,10 @@ class CheckoutAllocation(DeclaredFamily, affix="CheckoutAllocation"): def for_path(cls, rule: StorageRule, owner: Thread, path: Path) -> Iterator[CheckoutAllocation]: allocated = rule.roots_of(owner).allocation_name - problem, _, suffix = path.name.removeprefix(f'{allocated}--').partition('--') + prefix = f'{allocated}--' + name = (path.name.removeprefix(prefix) + if path.parent == rule.checkout_base and path.name.startswith(prefix) else '') + problem, _, suffix = name.partition('--') for member in cls.members_with(cls): yield from member.read(owner, path, problem, suffix) @@ -93,6 +96,14 @@ def owns_declared(self, rule: StorageRule, owner: Thread, path: Path, repository: RepositoryName) -> bool: return self.owns_present(rule, owner, path) + def problems(self) -> tuple[str, ...]: + """Only a problem allocation reserves an edit boundary for a problem.""" + return () + + def source_repositories(self, primary: RepositoryName, + companions: tuple[RepositoryName, ...]) -> tuple[RepositoryName, ...]: + return () + @dataclass(frozen=True) class HomeCheckoutAllocation(CheckoutAllocation): @@ -110,11 +121,19 @@ class ProblemCheckoutAllocation(CheckoutAllocation): """A checkout allocated to one problem; ``verb`` is the command that creates it.""" verb: ClassVar[str] + problem: str + + def problems(self) -> tuple[str, ...]: + return (self.problem,) + + def source_repositories(self, primary: RepositoryName, + companions: tuple[RepositoryName, ...]) -> tuple[RepositoryName, ...]: + return (primary,) @classmethod def read(cls, owner: Thread, path: Path, problem: str, suffix: str) -> Iterator[CheckoutAllocation]: - if path != owner.project_path and ProblemNumber.accepts(problem): + if ProblemNumber.accepts(problem): yield from cls.allocated(problem, suffix) @classmethod @@ -160,6 +179,10 @@ class RepositoryCheckoutAllocation(ProblemCheckoutAllocation, RepositoryAllocati problem: str suffix: str + def source_repositories(self, primary: RepositoryName, + companions: tuple[RepositoryName, ...]) -> tuple[RepositoryName, ...]: + return companions + @classmethod def allocated(cls, problem: str, suffix: str) -> Iterator[CheckoutAllocation]: if suffix: @@ -242,14 +265,6 @@ def repository_checkout(self, roots: StorageRoots, problem: str, checkout = self.order_checkout(roots, problem) return checkout.with_name(f'{checkout.name}--{repository.name}') - def checkout_problem(self, roots: StorageRoots, path: Path, other: str) -> str: - """The problem whose checkout holds path, read back from order_checkout's own name; - other for a path in no problem checkout.""" - allocated = self.order_checkout(roots, '') - names = path.relative_to(allocated.parent).parts[:1] if path.is_relative_to(allocated.parent) else () - return next((name.removeprefix(allocated.name).split('--')[0] for name in names - if name.startswith(allocated.name)), other) - def problem_checkouts(self, roots: StorageRoots, problem: str) -> tuple[Path, ...]: """Every checkout allocated to one problem: its check repository's, then each other's.""" checkout = self.order_checkout(roots, problem) diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index ac7577a0e3..b3df514c33 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -982,12 +982,12 @@ class PullRequestUpdatePage: class PullRequests: - """GitHub facts for recorded links, cached within one board read.""" + """Forge facts and checkout-local worktrees, cached within one board read.""" def __init__(self, todos: Sequence[Todo] = ()) -> None: self._read: dict[tuple[RepositoryName, int], ListedPullRequest] = {} self._open: dict[RepositoryName, tuple[int, ...]] = {} - self._worktrees: dict[RepositoryName, dict[str, tuple[str, ...]]] = {} + self._worktrees: dict[Path, Mapping[str, tuple[str, ...]]] = {} self._todos: dict[RepositoryName, dict[str, Todo]] = {} self.register(todos) @@ -1085,18 +1085,23 @@ def open(self, repo: RepositoryName) -> tuple[UnmergedListedPullRequest, ...]: return tuple(picked for number in self._open[repo] for picked in self._read[repo, number].as_unmerged()) + def worktrees(self, repository: Repo) -> Mapping[str, tuple[str, ...]]: + """One canonical local checkout's retained listing, not its forge namesake's.""" + checkout = Checkout(repository.checkout).resolved().path + if checkout not in self._worktrees: + self._worktrees[checkout] = WorkBranches(checkout).worktrees() + return self._worktrees[checkout] + def working_on(self, goals: Goals, repo: RepositoryName, row: ListedPullRequest) -> tuple[ThreadIncarnation, ...]: """The registered threads with the request's branch checked out: who is working on it - when no order says so. One worktree read per repository in this frame.""" + when no order says so. One worktree read per local checkout in this frame.""" snapshot = goals.registry.snapshot() - reader = WorkBranches(Repo.selected( - snapshot.commander_project.require().repos, repo.to_wire()).checkout) - if repo not in self._worktrees: - self._worktrees[repo] = reader.worktrees() + repository = Repo.selected(snapshot.commander_project.require().repos, repo.to_wire()) + reader = WorkBranches(repository.checkout) return tuple(dict.fromkeys( snapshot.require(name).incarnation - for checkout in reader.checkouts(snapshot, row.branch, self._worktrees[repo]) + for checkout in reader.checkouts(snapshot, row.branch, self.worktrees(repository)) for name in checkout.working())) def carry_reviews(self, goals: Goals, repo: RepositoryName) -> None: @@ -2074,6 +2079,9 @@ def instructions(self) -> tuple[str, ...]: @dataclass(frozen=True) class ProblemFinding(DeclaredFamily, affix="ProblemFinding"): + def source_locations(self) -> tuple[FileLine, ...]: + return () + def require_references(self, roots: SourceRoots, revision: SourceRevision) -> None: """An unlocated finding has no source reference to validate.""" @@ -2096,6 +2104,9 @@ class UnlocatedProblemFinding(ProblemFinding): class LocatedProblemFinding(ProblemFinding): where: FileLine + def source_locations(self) -> tuple[FileLine, ...]: + return (self.where,) + def require_references(self, roots: SourceRoots, revision: SourceRevision) -> None: self.where.require_exists(roots, revision) @@ -2349,6 +2360,10 @@ def __init_subclass__(cls, **options): if get_origin(base) is OrderScope ) + def source_locations(self) -> tuple[FileLine, ...]: + """Triage adds no implementation locations.""" + return () + def require_references(self, roots: SourceRoots, revision: SourceRevision) -> None: """Triage has no implementation declarations to validate.""" @@ -2406,6 +2421,9 @@ def completed(self, todo: Todo, delivery: ReportDelivery) -> Todo: class FixScope(OrderScope[CheckedDelivery]): shape: FixShape + def source_locations(self) -> tuple[FileLine, ...]: + return (*self.shape.parents, *self.shape.members) + def completion_problems(self, run: OrderRun, request: SourceCompletion) -> dict[str, str]: return request.source.evidence_problems(request.evidence, run) @@ -2491,6 +2509,15 @@ def cited_revision(self) -> SourceRevision: def source_roots(self, project: CommanderProject) -> SourceRoots: return OrderRoots(self.check.directory, project) + def source_repositories(self) -> tuple[RepositoryName, ...]: + """Companions are declared by qualified locations or explicit PR links.""" + locations = (self.owner, *self.finding.source_locations(), *self.scope.source_locations()) + return tuple(dict.fromkeys(( + *(link.repo for link in self.pull_requests), + *(repository for location in locations + for repository in location.repository.named_repositories()), + ))) + def require_references(self, project: CommanderProject) -> None: roots, revision = self.source_roots(project), self.cited_revision self.owner.require_exists(roots, revision) diff --git a/src/agent_comms/turn_digests.py b/src/agent_comms/turn_digests.py index 65b0bbc37b..0784efe81d 100644 --- a/src/agent_comms/turn_digests.py +++ b/src/agent_comms/turn_digests.py @@ -18,12 +18,12 @@ from .field_codec import FieldCodec from .held_orders import HeldOrders, LiveHold from .order_work import OrderWork -from .storage_rules import StorageRule +from .repo import Checkout, Repo +from .storage_rules import CheckoutAllocation, StorageRule from .thread_execution import ConversationPreparation from .thread_identity import ThreadIncarnation -from .thread_storage import StorageRoots from .threads import Thread -from .todos import ProgressRead, Todo, TodoStore +from .todos import ProgressRead, PullRequestLink, Todo, TodoStore from .tool_declaration import EditCall from .tool_result import ToolResult from .transcript_events import IncomingTranscript, TurnDigest @@ -31,6 +31,8 @@ if TYPE_CHECKING: from .comms import Comms + from .registry_document import RegistrySnapshot + from .repository_name import RepositoryName from .todos import OrderProgress from .transcript_sources import TranscriptSource @@ -230,35 +232,40 @@ class OrderEditRead[Result](OrderRead[Result]): @classmethod def for_commander(cls, comms: Comms, commander: str) -> Iterator[Self]: - held = HeldOrders.read(comms, commander) + snapshot = comms.registry.snapshot() + held = HeldOrders.of(comms.goals.todos, snapshot, commander) + history = comms.goals.todos.history_for_commander(snapshot.require(commander).incarnation) + pulls = ProgressRead(history) + workplaces = EditWorkplaces.capture(comms, snapshot, held.live, history, pulls) yield from cls.judgeable( - comms.root, held.live, ProgressRead().progress(held.todos, comms.registry)) + comms.root, held.live, pulls.progress(held.todos, comms.registry), workplaces) @classmethod def judgeable(cls, root: Path, holds: Iterable[LiveHold], - progress: Mapping[str, OrderProgress]) -> Iterator[Self]: + progress: Mapping[str, OrderProgress], + workplaces: Mapping[ThreadIncarnation, EditWorkplaces]) -> Iterator[Self]: """The reads a verdict can still clear: an order whose PR merged has no head left to judge, so its edits are nobody's review and its transcript is never read.""" - return (read for read in cls.for_orders(root, holds) if progress[read.todo.id].judgeable) + return (read for read in cls.for_orders(root, holds, workplaces) + if progress[read.todo.id].judgeable) @classmethod - def for_orders(cls, root: Path, holds: Iterable[LiveHold]) -> Iterator[Self]: + def for_orders(cls, root: Path, holds: Iterable[LiveHold], + workplaces: Mapping[ThreadIncarnation, EditWorkplaces]) -> Iterator[Self]: """Read each live hold's transcript from its canonical review cut; each edit is read under the one order its workplace names.""" store = TodoStore(root) - rule = StorageRule.load(root) - holds = tuple(holds) for hold in holds: yield cls( root, hold.holder, TurnsSince(store.unjudged_since(hold.todo)), hold.todo, - EditWorkplaces.of(rule, hold.holder, tuple( - other.todo for other in holds if other.holder.name == hold.holder.name)), + workplaces[hold.holder.incarnation], ) def owned(self, edits: Iterable[EditCall]) -> tuple[EditCall, ...]: judged = frozenset(self.todo.judged_calls) - return tuple(edit for edit in edits if edit.id not in judged - and self.workplaces.order_of(edit.path) == self.todo.id) + return tuple(replace(edit, problem=self.todo.id) for edit in edits + if edit.id not in judged + and self.workplaces.problems_for(edit.path) == frozenset((self.todo.id,))) def edits_from_turns(self, turns: tuple[TurnDigest, ...]) -> tuple[EditCall, ...]: since = self.bound.at @@ -270,22 +277,141 @@ def edits_from_turns(self, turns: tuple[TurnDigest, ...]) -> tuple[EditCall, ... @dataclass(frozen=True) class EditWorkplaces: - """Which of one unit's held orders an edit belongs to. An edit inside a problem checkout - belongs to that problem, held or not, so a released problem's edits join no other order; - any other path (scratch, a shared checkout) belongs to the order the unit was given last. - Each edit is read under at most one order.""" + """One read's derived boundaries. Allocations take precedence over generic source roots; + zero or multiple candidates never select a held order. Original edit records stay put.""" rule: StorageRule - roots: StorageRoots - latest: str + owners: Mapping[Path, Thread] + allocated: Mapping[Path, frozenset[str]] + sources: Mapping[Path, frozenset[str]] @classmethod - def of(cls, rule: StorageRule, holder: Thread, todos: tuple[Todo, ...]) -> Self: - return cls(rule, rule.roots_of(holder), - max(todos, key=lambda todo: todo.order.checked().at.recorded_at("An order")).id) - - def order_of(self, path: str) -> str: - return self.rule.checkout_problem(self.roots, Path(path), self.latest) + def capture(cls, comms: Comms, snapshot: RegistrySnapshot, holds: Iterable[LiveHold], + history: tuple[Todo, ...], pulls: ProgressRead + ) -> dict[ThreadIncarnation, EditWorkplaces]: + """Acquire source/forge facts once, then match edits without Git or network calls.""" + holds = tuple(holds) + if not holds: + return {} + repositories = snapshot.commander_project.require().repos + registered = frozenset(repo.github for repo in repositories) + rule = StorageRule.load(comms.root) + todos = {todo.id: todo for todo in history} + declared = { + todo.id: tuple(dict.fromkeys(repository for order in todo.order.declarations + for repository in order.source_repositories())) + for todo in history + } + trees: dict[RepositoryName, dict[str, tuple[str, ...]]] = {} + for repo in repositories: + branches = trees.setdefault(repo.github, {}) + for branch, worktrees in pulls.worktrees(repo).items(): + branches[branch] = (*branches.get(branch, ()), *worktrees) + listed = {Path(path) for branches in trees.values() + for worktrees in branches.values() for path in worktrees} + explicit = {order.check.directory for hold in holds + for order in hold.todo.order.declarations} + paths = listed | explicit + for hold in holds: + roots = rule.roots_of(hold.holder) + paths.add(rule.order_checkout(roots, hold.todo.id)) + paths.update(rule.repository_checkout(roots, hold.todo.id, repo) + for repo in declared[hold.todo.id]) + # Qualify only captured Git roots, never nearest parents of individual events. + # An alias cannot manufacture another problem allocation around the same source. + present = Checkout.git_roots(*paths) + actual: dict[Path, Repo] = {} + for path in present: + if path.is_absolute() and Checkout(path).resolved().path == path: + repository = Repo.of(path) + if repository.github in registered: + actual[path] = repository + # An explicit retained source may be a separate clone, not a worktree of the + # registered checkout. Read its existing worktree owner once, never per edit; + # only that already-declared root joins the branch facts, not arbitrary siblings. + for path in explicit - listed: + if path in actual: + branches = trees[actual[path].github] + for branch, worktrees in pulls.worktrees(actual[path]).items(): + if str(path) in worktrees: + branches[branch] = (*branches.get(branch, ()), str(path)) + links = tuple(dict.fromkeys(link for hold in holds + for order in hold.todo.order.declarations + for link in order.pull_requests)) + requests = {PullRequestLink(repo.github, row.number): row + for repo in repositories + for row in pulls.rows(repo.github, tuple( + link.number for link in links if link.repo == repo.github))} + owners = {rule.order_checkout(rule.roots_of(thread), ''): thread + for thread in snapshot.threads.values()} + allocations: dict[Path, tuple[tuple[Thread, CheckoutAllocation], ...]] = {} + for path in present: + candidates = tuple((thread, allocation) + for prefix, thread in owners.items() + if path.parent == prefix.parent and path.name.startswith(prefix.name) + for allocation in CheckoutAllocation.for_path(rule, thread, path) + if allocation.problems()) + if candidates: + allocations[path] = candidates + # Every recognized allocation reserves its boundary, even without registered Git proof. + allocated: dict[Path, frozenset[str]] = {path: frozenset() for path in allocations} + for path, repository in actual.items(): + if path in allocations: + allocated[path] = frozenset( + problem for thread, allocation in allocations[path] + for problem in allocation.problems() + if problem in todos + and repository.github in allocation.source_repositories( + todos[problem].repo, declared[problem]) + and allocation.owns_declared(rule, thread, path, repository.github)) + workplaces: dict[ThreadIncarnation, EditWorkplaces] = {} + for holder in dict.fromkeys(hold.holder.incarnation for hold in holds): + sources: dict[Path, set[str]] = {} + for hold in holds: + if hold.holder.incarnation != holder: + continue + todo = hold.todo + linked = tuple(link for order in todo.order.declarations + for link in order.pull_requests) + for order in todo.order.declarations: + path = order.check.directory + # Explicit pre-PR source is valid only for its primary actual repository, + # and ceases being a generic workplace once that source has a PR link. + if (path in actual and actual[path].github == todo.repo + and not any(link.repo == todo.repo for link in linked)): + sources.setdefault(path, set()).add(todo.id) + for link in linked: + if link not in requests: + continue # Unregistered repositories cannot qualify a source boundary. + for worktree_path in trees[link.repo].get(requests[link].branch, ()): + root = Path(worktree_path) + if root in actual and actual[root].github == link.repo: + sources.setdefault(root, set()).add(todo.id) + workplaces[holder] = cls(rule, owners, allocated, { + path: frozenset(problems) for path, problems in sources.items()}) + return workplaces + + def problems_for(self, path: str) -> frozenset[str]: + original = Path(path) + if not original.is_absolute(): + return frozenset() + # Resolve the edited path once, not its missing leaf with Git. Lexical membership + # rejects aliases into a checkout; resolved membership rejects symlink escapes. + resolved = original.resolve() + parts = (original.relative_to(self.rule.checkout_base).parts + if original.is_relative_to(self.rule.checkout_base) else ()) + boundary = self.rule.checkout_base.joinpath(*parts[:1]) + reserved = any(allocation.problems() + for prefix, thread in self.owners.items() + if boundary.parent == prefix.parent and boundary.name.startswith(prefix.name) + for allocation in CheckoutAllocation.for_path(self.rule, thread, boundary)) + # Even a missing/released/unknown allocation reserves its lexical boundary. Only + # captured actual Git identity may supply an owner; absence never enables spill. + candidates = ({boundary: self.allocated.get(boundary, frozenset())} + if reserved else self.sources) + return frozenset(problem for root, problems in candidates.items() + if original.is_relative_to(root) and resolved.is_relative_to(root) + for problem in problems) class OrderEditsRead(OrderEditRead[tuple[EditCall, ...]]): diff --git a/tests/test_commander_forks.py b/tests/test_commander_forks.py index 3bc683301d..0a54c4e43d 100644 --- a/tests/test_commander_forks.py +++ b/tests/test_commander_forks.py @@ -9,13 +9,14 @@ import time from concurrent.futures import ThreadPoolExecutor from dataclasses import replace +from datetime import UTC, datetime from pathlib import Path import pytest from agent_comms_hooks.watch import WatchClaim, WatchRecord -from agent_comms.change_report import FileLine +from agent_comms.change_report import FileLine, NamedRepository from agent_comms.commander.order import CommanderContext, OrderFile from agent_comms.commander.standing import Unjudged from agent_comms.commander_heartbeat import CommanderHeartbeat, CommanderObservation @@ -33,6 +34,12 @@ ) from agent_comms.goal_actions import OrderGoalAction, SetGoalAction from agent_comms.goals import CountCheckResult, OrderCheck +from agent_comms.held_orders import HeldOrders +from agent_comms.native_transcript import NativeTranscript +from agent_comms.repository_name import RepositoryName +from agent_comms.session_file import RegistrySessionLocation +from agent_comms.storage_rules import StorageRule +from agent_comms.turn_digests import EditWorkplaces, OrderEditsRead, OrderEditCountRead, TurnsSince from agent_comms.installer.backup import StoreBackup from agent_comms.maintenance_barrier import ( MaintenanceBarrier, @@ -43,21 +50,29 @@ from agent_comms.problem_number import ProblemNumber from agent_comms.recorded_time import RecordedEventTime from agent_comms.slop import DeclarationDerivation, ExistingLine, Slop -from agent_comms.thread_execution import ExternalThreadExecution +from agent_comms.thread_execution import ExternalThreadExecution, NativeThreadExecution from agent_comms.thread_identity import RecordedThreadIncarnation from agent_comms.thread_model import DeclaredModel from agent_comms.thread_status import ArchivedThreadStatus from agent_comms.threads import Thread from agent_comms.todos import ( + Assignment, + DraftListedPullRequest, ExpectedProblem, + FixScope, + FixShape, IncomingWork, + LocatedProblemFinding, PendingProblemOrder, ProblemAllocation, ProgressRead, + PullRequests, + PullRequestLink, ReleaseTodo, StrikeReason, SupersededVerdict, Todo, + UnlocatedProblemFinding, ) from agent_comms.tool_caller import CommanderForkCaller from agent_comms.tool_declaration import ToolDeclaration @@ -310,18 +325,271 @@ def test_unjudged_reminder_names_each_order_with_edits(fork_bus): def test_edits_under_a_merged_order_are_nobodys_review(fork_bus): """Overload, the HUD and the judge read one judgeable owner: a merged head gates nothing.""" - from agent_comms.held_orders import HeldOrders from agent_comms.todos import MergedProgress, NoCommitProgress - from agent_comms.turn_digests import OrderEditCountRead comms, source, _ = fork_bus record_edits(comms, source, 3) holds = HeldOrders.read(comms, "lead").live open_order = {hold.todo.id: NoCommitProgress() for hold in holds} merged = {hold.todo.id: MergedProgress("a" * 40) for hold in holds} + workplaces = EditWorkplaces.capture( + comms, comms.registry.snapshot(), holds, tuple(hold.todo for hold in holds), ProgressRead()) assert [read.read() for read in OrderEditCountRead.judgeable( - comms.root, holds, open_order)] == [3] - assert tuple(OrderEditCountRead.judgeable(comms.root, holds, merged)) == () + comms.root, holds, open_order, workplaces)] == [3] + assert tuple(OrderEditCountRead.judgeable(comms.root, holds, merged, workplaces)) == () + + +@pytest.fixture +def edit_sources(fork_bus, tmp_path, configured_repository): + """Two actual origins, a held Core-like source and its real allocated worktree.""" + comms, source, _ = fork_bus + foreign = SourceRepository.create(tmp_path / "foreign", RepositoryName("example", "foreign")) + configured_repository(comms, source, foreign) + holder = comms.registry.require("retained") + rule = StorageRule.load(comms.root) + primary = rule.order_checkout(rule.roots_of(holder), "P40") + primary.parent.mkdir(parents=True, exist_ok=True) + source.git(source.checkout, "worktree", "add", "-q", "-b", "held-P40", str(primary)) + return comms, source, foreign, primary + + +def edit_workplaces(comms): + snapshot = comms.registry.snapshot() + holds = HeldOrders.of(comms.goals.todos, snapshot, "lead").live + history = comms.goals.todos.history_for_commander(snapshot.require("lead").incarnation) + return holds, EditWorkplaces.capture(comms, snapshot, holds, history, ProgressRead(history)) + + +def save_edit_order(comms, todo): + """Publish an actual retained order row, not a preselected workplace answer.""" + with comms.goals.todos.writing() as db: + todo.upsert(db) + + +def another_edit_order(comms): + """A saved multi-order state, supported for records predating one-unit/one-order.""" + original = comms.goals.todos.get("P40") + lead = comms.registry.require("lead").incarnation + return Todo.ordered( + "P41", original.repo, "Independent second source change", lead, + Assignment.of(comms.registry.require("retained").incarnation, lead, GoalId.mint()), + RecordedEventTime(time.time() + 600), + replace(original.order.checked(), id=OrderId.mint(), + at=RecordedEventTime(time.time() + 10)), original.request.require(), + ) + + +def captured_edit_sources(comms, paths, session): + """Original external tool starts and an actual Pi journal, with the same edit paths.""" + actor = comms.registry.require("retained").incarnation + timestamp = time.time() + stamp = datetime.fromtimestamp(timestamp, UTC).isoformat() + calls = tuple((ToolCallId(f"source-edit-{i}"), { + "path": str(path), "edits": [{"oldText": "", "newText": f"VALUE = {i}"}], + }) for i, path in enumerate(paths)) + events = tuple(ToolStartTranscript( + position=i, tool_call_id=identity, tool_name="edit", timestamp=timestamp, + tool=ToolDeclaration.record("edit", arguments, call_id=identity), + ) for i, (identity, arguments) in enumerate(calls)) + ExternalTranscript(comms.root, actor).finish( + turn_id=NativeTurnId("source-work"), item_id=NativeItemId("source-work"), events=events) + rows = ( + {"type": "session", "version": 3, "id": "source-work", "timestamp": stamp}, + {"type": "message", "id": "source-input", "timestamp": stamp, + "message": {"role": "user", "content": [{"type": "text", "text": "Edit these sources"}]}}, + {"type": "message", "id": "source-tools", "parentId": "source-input", "timestamp": stamp, + "message": {"role": "assistant", "content": [ + {"type": "toolCall", "id": identity, "name": "edit", "arguments": arguments} + for identity, arguments in calls]}}, + ) + session.write_text("".join(json.dumps(row) + "\n" for row in rows)) + return calls + + +@pytest.mark.parametrize("execution", (NativeThreadExecution, ExternalThreadExecution)) +def test_edit_membership_preserves_native_and_external_foreign_records(edit_sources, tmp_path, execution): + comms, source, foreign, primary = edit_sources + scratch = StorageRule.load(comms.root).roots_of(comms.registry.require("retained")).scratch + paths = (primary / "owner.py", foreign.checkout / "owner.py", scratch / "comment.md", + scratch / "plugin" / "introspection_trace.py", scratch / "typemap" / "probe.py", + primary / "new_source.py", primary / "deleted_source.py") + deleted = paths[-1] + deleted.write_text("BEFORE = True\n") + deleted.unlink() + session = tmp_path / "original-native.jsonl" + calls = captured_edit_sources(comms, paths, session) + original_bytes = session.read_bytes() + before = comms.goals.todos.get("P40") + holds, workplaces = edit_workplaces(comms) + hold, = holds + holder = replace(hold.holder, execution=execution, + session_file=RegistrySessionLocation.from_path(session, comms.root)) + read = OrderEditsRead( + comms.root, holder, TurnsSince(before.order.checked().at.recorded_at("An order")), + before, workplaces[holder.incarnation]) + expected = {calls[index][0] for index in (0, 5, 6)} + assert {edit.id for edit in read.read()} == expected + count = OrderEditCountRead(read.root, holder, read.bound, before, read.workplaces) + assert count.read() == len(expected) + assert comms.goals.todos.get("P40") == before + assert before.judged_calls == () + assert session.read_bytes() == original_bytes + native = NativeTranscript(session) + native_events = tuple(event for turn in native.turns(TurnsSince(0)) for event in turn.events) + native_ids = {edit.id for event in native_events + for edit in event.edit_calls(holder.name, "P40", 0)} + with ExternalTranscript(comms.root, holder.incarnation).reading() as external: + external_ids = {edit.id for turn in external.turns(TurnsSince(0)) for event in turn.events + for edit in event.edit_calls(holder.name, "P40", 0)} + assert native_ids == external_ids == {identity for identity, _ in calls} + + +@pytest.mark.parametrize("declare_companion", ( + pytest.param(lambda order, location: replace( + order, scope=FixScope(FixShape(order.scope.shape.parents, (location,))), + finding=UnlocatedProblemFinding()), id="scope-unlocated-finding"), + pytest.param(lambda order, location: replace( + order, finding=LocatedProblemFinding(location)), id="located-finding"), +)) +def test_canonical_primary_companion_and_released_roots_take_precedence(edit_sources, declare_companion): + comms, source, foreign, primary = edit_sources + store = comms.goals.todos + todo = store.get("P40") + owner = comms.registry.require("retained") + rule = StorageRule.load(comms.root) + roots = rule.roots_of(owner) + companion = rule.repository_checkout(roots, todo.id, foreign.github) + foreign.git(foreign.checkout, "worktree", "add", "-q", "-b", "companion-P40", str(companion)) + # A qualified source declaration, not origin equality alone, authorizes the companion. + location = FileLine("owner.py", 2, NamedRepository(foreign.github)) + save_edit_order(comms, replace(todo, order=declare_companion(todo.order, location))) + released = another_edit_order(comms) + save_edit_order(comms, released) + released_path = rule.order_checkout(roots, released.id) + source.git(source.checkout, "worktree", "add", "-q", "-b", "released-P41", str(released_path)) + store.apply(released.id, ReleaseTodo( + released.revision, comms.registry.require("lead").incarnation, released.held(), + StrikeReason("The independent order is released; its checkout must not spill."))) + current = store.get("P40") + save_edit_order(comms, replace(current, order=replace( + current.order, check=replace(current.order.check, directory=released_path)))) + _, workplaces = edit_workplaces(comms) + selected = workplaces[owner.incarnation] + assert selected.problems_for(str(primary / "new.py")) == frozenset(("P40",)) + assert selected.problems_for(str(companion / "new.py")) == frozenset(("P40",)) + assert selected.problems_for(str(released_path / "new.py")) == frozenset(("P41",)) + # A registered foreign repository in an undeclared companion allocation stays unowned. + current = store.get("P40") + save_edit_order(comms, replace(current, order=replace( + current.order, scope=source.scope(), finding=UnlocatedProblemFinding()))) + _, workplaces = edit_workplaces(comms) + assert workplaces[owner.incarnation].problems_for(str(companion / "new.py")) == frozenset() + + +def test_misleading_primary_allocation_refuses_foreign_origin(edit_sources): + comms, source, foreign, _ = edit_sources + another = another_edit_order(comms) + save_edit_order(comms, another) + holder = comms.registry.require("retained") + rule = StorageRule.load(comms.root) + misleading = rule.order_checkout(rule.roots_of(holder), another.id) + foreign.git(foreign.checkout, "worktree", "add", "-q", "-b", "wrong-origin", str(misleading)) + _, workplaces = edit_workplaces(comms) + assert workplaces[holder.incarnation].problems_for(str(misleading / "owner.py")) == frozenset() + # P41's later timestamp never wins the ambiguous shared pre-PR source either. + assert workplaces[holder.incarnation].problems_for(str(source.checkout / "owner.py")) == frozenset(("P40", "P41")) + + +def test_noncanonical_worktree_needs_qualified_link_not_matching_origin(edit_sources, tmp_path, github_cli): + comms, source, _, _ = edit_sources + holder = comms.registry.require("retained") + linked = StorageRule.load(comms.root).roots_of(holder).scratch / "legacy-source" + linked.parent.mkdir(parents=True, exist_ok=True) + unrelated = tmp_path / "same-origin-no-link" + source.git(source.checkout, "worktree", "add", "-q", "-b", "linked-repair", str(linked)) + source.git(source.checkout, "worktree", "add", "-q", "-b", "unlinked-work", str(unrelated)) + _, workplaces = edit_workplaces(comms) + assert workplaces[holder.incarnation].problems_for(str(source.checkout / "new.py")) == frozenset(("P40",)) + assert workplaces[holder.incarnation].problems_for(str(linked / "new.py")) == frozenset() + todo = comms.goals.todos.get("P40") + row = DraftListedPullRequest(number=132, branch="linked-repair", head=source.main_commit, + updated="2026-10-28T12:00:00Z", base="main", reviewable_times=()) + github_cli.put(source.github, row, body="P40 source ownership") + save_edit_order(comms, replace(todo, order=replace( + todo.order, pull_requests=(PullRequestLink(source.github, row.number),)))) + _, workplaces = edit_workplaces(comms) + selected = workplaces[holder.incarnation] + assert selected.problems_for(str(linked / "new.py")) == frozenset(("P40",)) + assert selected.problems_for(str(unrelated / "new.py")) == frozenset() + assert selected.problems_for(str(source.checkout / "new.py")) == frozenset() + # The explicit retained source may be a different clone: it will not occur in the + # registered checkout's worktree list, but its own branch still has to match the PR. + separate = tmp_path / "explicit-source-clone" + source.git(tmp_path, "clone", "-q", str(source.remote), str(separate)) + source.git(separate, "checkout", "-q", "-b", "linked-repair") + independent = SourceRepository.of(separate) + assert independent.github == source.github + pulls = PullRequests() + registered_trees = pulls.worktrees(source) + independent_trees = pulls.worktrees(independent) + assert str(linked) in registered_trees["linked-repair"] + assert str(separate) not in registered_trees["linked-repair"] + assert str(separate) in independent_trees["linked-repair"] + assert str(linked) not in independent_trees["linked-repair"] + assert pulls.worktrees(source) is registered_trees + assert pulls.worktrees(independent) is independent_trees + alias = SourceRepository.of(separate / ".." / separate.name) + assert alias.checkout != independent.checkout + assert pulls.worktrees(alias) is independent_trees + current = comms.goals.todos.get("P40") + save_edit_order(comms, replace(current, order=replace( + current.order, check=replace(current.order.check, directory=separate)))) + _, workplaces = edit_workplaces(comms) + assert workplaces[holder.incarnation].problems_for(str(separate / "new.py")) == frozenset(("P40",)) + source.git(separate, "checkout", "-q", "-b", "wrong-branch") + _, workplaces = edit_workplaces(comms) + assert workplaces[holder.incarnation].problems_for(str(separate / "new.py")) == frozenset() + + +def test_ambiguous_pre_pr_work_never_picks_latest(edit_sources, tmp_path): + comms, source, _, primary = edit_sources + save_edit_order(comms, another_edit_order(comms)) + session = tmp_path / "ambiguous.jsonl" + calls = captured_edit_sources(comms, (source.checkout / "owner.py", primary / "new.py"), session) + shown = command(comms, "judge", "show") + assert shown.returncode == 0, shown.stderr + assert calls[0][0] not in shown.stdout and calls[1][0] in shown.stdout + assert shown.stdout.splitlines()[-1] == "1 unjudged" + assert all(not todo.judged_calls for todo in comms.goals.todos.history_for_commander( + comms.registry.require("lead").incarnation)) + + +def test_missing_unknown_allocation_still_overrides_an_enclosing_source(edit_sources): + comms, source, _, _ = edit_sources + # A genuine source encloses the allocation base; no physical P999 checkout exists. + # The source cannot swallow a historical/deleted allocation by generic containment. + rule = replace(StorageRule.load(comms.root), checkout_base=source.checkout) + StorageRule.path_in(comms.root).write_text(json.dumps(FieldCodec.encode(rule))) + holder = comms.registry.require("retained") + missing = rule.order_checkout(rule.roots_of(holder), "P999") + assert not missing.exists() + _, workplaces = edit_workplaces(comms) + selected = workplaces[holder.incarnation] + assert selected.problems_for(str(source.checkout / "ordinary.py")) == frozenset(("P40",)) + assert selected.problems_for(str(missing / "old.py")) == frozenset() + + +def test_symlink_aliases_and_foreign_escapes_cannot_mint_membership(edit_sources, tmp_path): + comms, source, foreign, primary = edit_sources + alias = tmp_path / "alias" + alias.symlink_to(primary, target_is_directory=True) + escape = primary / "foreign.py" + escape.symlink_to(foreign.checkout / "owner.py") + holder = comms.registry.require("retained") + _, workplaces = edit_workplaces(comms) + selected = workplaces[holder.incarnation] + assert selected.problems_for(str(alias / "owner.py")) == frozenset() + assert selected.problems_for(str(escape)) == frozenset() def test_open_unit_consumers_agree_on_one_assignment(fork_bus): diff --git a/tests/test_judge_dropped.py b/tests/test_judge_dropped.py index 214f423aeb..9710cfafec 100644 --- a/tests/test_judge_dropped.py +++ b/tests/test_judge_dropped.py @@ -1,9 +1,11 @@ """An edit the unit discarded is settled as dropped from git, through the real judge command.""" import json import time +from dataclasses import replace import pytest +from agent_comms.change_report import FileLine, NamedRepository from agent_comms.commander.judge import Edit, main from agent_comms.errors import TodoError from agent_comms.external_transcript import ExternalTranscript @@ -12,7 +14,7 @@ ToolCallId, ) from agent_comms.registry_document import RegisteredCommanderProject -from agent_comms.todos import DroppedVerdict, RewrittenVerdict +from agent_comms.todos import DroppedVerdict, FixScope, FixShape, RewrittenVerdict from agent_comms.tool_declaration import ToolDeclaration from agent_comms.transcript_events import ToolStartTranscript from agent_comms.turn_item import NativeItemId, NativeTurnId @@ -56,11 +58,13 @@ def judge_dropped(ctx, edit: Edit, line: str) -> None: @pytest.fixture -def discarded_and_kept(held_commander): +def discarded_and_kept(held_commander, cited_source): ctx = held_commander repo = ctx.repos[0] rule = ctx.comms.registry.storage_rule gone = rule.order_checkout(rule.roots_of(ctx.comms.registry.require("worker")), "P1") + gone.parent.mkdir(parents=True, exist_ok=True) + cited_source.git(repo.checkout, "worktree", "add", "-q", "-b", "held-P1", str(gone)) base = repo.git(repo.checkout, "rev-parse", "--abbrev-ref", "HEAD").strip() repo.git(repo.checkout, "checkout", "-q", "-b", "formatter") committed(repo, "dropped_module.py", DROPPED) @@ -160,7 +164,14 @@ def test_rewritten_reads_the_repository_the_edit_was_made_in(discarded_and_kept, rule = ctx.comms.registry.storage_rule allocated = rule.repository_checkout(rule.roots_of(ctx.comms.registry.require("worker")), "P1", other.github) + other.git(other.checkout, "worktree", "add", "-q", "-b", "companion-P1", str(allocated)) + todo = ctx.comms.goals.todos.get("P1") + scope = FixScope(FixShape(todo.order.scope.shape.parents, ( + *todo.order.scope.shape.members, FileLine("other_module.py", 1, NamedRepository(other.github))))) + with ctx.comms.goals.todos.writing() as db: + replace(todo, order=replace(todo.order, scope=scope)).upsert(db) recorded_edit(ctx, ToolCallId("other-call"), str(allocated / "other_module.py"), FOREIGN) + assert "other-call" in unjudged(ctx) committed(other, "other_module.py", "REPLACED_MARKER_LINE = 'what the later commit wrote'") other.git(other.checkout, "push", "-q", "origin", "HEAD:refs/heads/unit-rewrite") head = other.git(other.checkout, "rev-parse", "HEAD").strip() diff --git a/tests/test_judge_edit_ownership.py b/tests/test_judge_edit_ownership.py index 29cb4914c7..ce80781616 100644 --- a/tests/test_judge_edit_ownership.py +++ b/tests/test_judge_edit_ownership.py @@ -45,8 +45,12 @@ def two_orders(held_commander, cited_source): request, ).insert_if_absent(db) rule = StorageRule.load(ctx.root) - checkout = {problem: rule.problem_checkouts(rule.roots_of(worker), problem)[0] + checkout = {problem: rule.order_checkout(rule.roots_of(worker), problem) for problem in ("P1", "P2")} + for problem, path in checkout.items(): + path.parent.mkdir(parents=True, exist_ok=True) + cited_source.git(cited_source.checkout, "worktree", "add", "-q", "-b", + f"edit-ownership-{problem}", str(path)) return ctx, checkout @@ -56,7 +60,8 @@ def test_an_edit_belongs_to_the_order_whose_checkout_it_is_in(two_orders): recorded_edit(ctx, ToolCallId("p2-call"), str(checkout["P2"] / "two.py"), "TWO = 'second problem'") recorded_edit(ctx, ToolCallId("scratch-call"), str(ctx.root / "scratch" / "notes.md"), "notes for the PR") filed = [(edit.id, edit.problem) for edit in unjudged(ctx).values()] - assert sorted(filed) == [("p1-call", "P1"), ("p2-call", "P2"), ("scratch-call", "P2")] + assert sorted(filed) == [("p1-call", "P1"), ("p2-call", "P2")] + assert ctx.comms.goals.todos.get("P2").judged_calls == () def test_a_send_back_leaves_the_edits_it_does_not_name_unjudged(two_orders): @@ -88,10 +93,11 @@ def test_a_second_verdict_on_one_edit_is_refused(two_orders): store.apply("P1", replace(verdict, expected_revision=judged.revision)) -def test_an_edit_in_a_released_problems_checkout_joins_no_held_order(two_orders): +def test_an_edit_in_an_unknown_problems_checkout_joins_no_held_order(two_orders, cited_source): ctx, checkout = two_orders released = checkout["P1"].with_name(checkout["P1"].name.replace("--P1", "--P9")) - recorded_edit(ctx, ToolCallId("p9-call"), str(released / "nine.py"), "NINE = 'released problem'") + cited_source.git(cited_source.checkout, "worktree", "add", "-q", "-b", "unknown-P9", str(released)) + recorded_edit(ctx, ToolCallId("p9-call"), str(released / "nine.py"), "NINE = 'unknown problem'") recorded_edit(ctx, ToolCallId("p2-call"), str(checkout["P2"] / "two.py"), "TWO = 'second problem'") assert [(edit.id, edit.problem) for edit in unjudged(ctx).values()] == [("p2-call", "P2")]