P895: qualify edit ownership by source workplace, not latest session order

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}"
 

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

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:

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),
         )

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)

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)