P440: Keep commander reads separate from registered fork actors

src/agent_comms/commander/capacity.py

@@ -27,7 +27,8 @@ def of(cls, comms: Comms, commander: str) -> Load:
         forks = tuple(sorted(
             thread.name for thread in snapshot.threads.values()
             if thread.parent.named(owner.name) and snapshot.status(thread.name).receiving
-            and todos.held_for_commander(thread.incarnation)
+            and any(assignment.parent_identity == thread.incarnation
+                    for todo in held for assignment in todo.holder_state.current_assignments())
         ))
         unjudged = sum(
             read.read() for read in OrderEditCountRead.for_orders(comms.root, snapshot, held)

src/agent_comms/commander/judge.py

@@ -242,7 +242,7 @@ class OkVerb(VerdictVerb):
     ] = ""
 
     def operation(self, ctx: order.Commander, todo: Todo, edit: Edit) -> AcceptedVerdict:
-        actor = ctx.incarnation
+        actor = ctx.actor_identity
         if self.peer:
             return PeerAcceptedVerdict(todo.revision, actor, todo.held(), edit.id,
                                        quote=self.serves, peer=self.peer)
@@ -273,7 +273,7 @@ class SupersededVerb(VerdictVerb):
 
     def operation(self, ctx: order.Commander, todo: Todo, edit: Edit) -> SupersededVerdict:
         return SupersededVerdict(todo.revision,
-                                 ctx.incarnation,
+                                 ctx.actor_identity,
                                  todo.held(), edit.id, replacement=self.by, reason=self.why)
 
     def check_edits(self, edits: list[Edit]) -> None:
@@ -289,7 +289,7 @@ class DroppedVerb(VerdictVerb):
 
     def operation(self, ctx: order.Commander, todo: Todo, edit: Edit) -> DroppedVerdict:
         added = tuple(line for hunk in edit.hunks for line in hunk.new_text.splitlines())
-        return DroppedVerdict(todo.revision, ctx.incarnation, todo.held(), edit.id,
+        return DroppedVerdict(todo.revision, ctx.actor_identity, todo.held(), edit.id,
                               path=edit.path, added=added)
 
     def check_edits(self, edits: list[Edit]) -> None:
@@ -335,7 +335,7 @@ def slop(self) -> Slop:
                                            at=self.at, derive=self.derive, by=self.by))
 
     def operation(self, ctx: order.Commander, todo: Todo, edit: Edit) -> SendBackVerdict:
-        return SendBackVerdict(todo.revision, ctx.incarnation,
+        return SendBackVerdict(todo.revision, ctx.actor_identity,
                                todo.held(), edit.id,
                                reason=StrikeReason(f'{todo.id} edit sent back: {self.why}'),
                                evidence=(FileLine.from_wire(self.at),))
@@ -354,7 +354,7 @@ def report(self, ctx: order.Commander, todos: tuple[Todo, ...]) -> None:
             audience = todo.verdicts[-1].audience.of(todo.held().owner, todo.goal)
             roast(ctx, audience, todo.holder, self.lines, self.roast, self.fix, self.slop())
             line = (f'{time.strftime("%Y-%m-%d %H:%M")} {todo.holder} '
-                    f'(strike {todo.strike_count} on {todo.id}; caught by {ctx.commander} at review): '
+                    f'(strike {todo.strike_count} on {todo.id}; caught by {ctx.actor_name} at review): '
                     f'{todo.strikes[-1].reason}')
             ctx.on_record(line, 'Send-back', audience)
 

src/agent_comms/commander/order.py

@@ -28,7 +28,7 @@
 from pathlib import Path
 from typing import Annotated, ClassVar, Self
 
-from pydantic import PrivateAttr
+from pydantic import Field, InstanceOf, PrivateAttr
 
 from agent_comms.change_report import FileLine, MainRevision, NoSourceReference
 from agent_comms.cli_bridges.shared import DeclaredVerb
@@ -52,6 +52,7 @@
 from agent_comms.message_reference import MessageReference
 from agent_comms.native_tool_call import SelectedToolDenied
 from agent_comms.plan_phases import OrderPhase
+from agent_comms.problem_number import ProblemNumber
 from agent_comms.progress_claims import MergedClaim, WorkingClaim
 from agent_comms.recorded_time import RecordedEventTime
 from agent_comms.repo import Repo
@@ -87,7 +88,7 @@
     VoidStrikeTodo,
     WithdrawTodo,
 )
-from agent_comms.tool_caller import CommanderForkCaller, NoThreadCaller, ThreadCaller
+from agent_comms.tool_caller import CommanderForkCaller, NoThreadCaller
 from agent_comms.tools import (
     CommsChangeProblemTool,
     CommsFireTool,
@@ -234,12 +235,13 @@ def revision(repo: str, ref: str) -> str:
 class CommanderContext(CommanderConfig):
     """The selected root's bus and canonical reads, shared by command families."""
 
+    actor: Annotated[
+        InstanceOf[CommanderForkCaller] | InstanceOf[NoThreadCaller], Field(exclude=True)
+    ] = NoThreadCaller()
     _comms: Comms = PrivateAttr()
-    _caller: ThreadCaller = PrivateAttr()
 
     def model_post_init(self, context: object) -> None:
         self._comms = Comms(self.root)
-        self._caller = ThreadCaller(self.commander)
 
     @boundary_decoder
     def model_copy(self, *, update: Mapping[str, object] | None = None,
@@ -261,23 +263,23 @@ def comms(self) -> Comms:
     def incarnation(self) -> ThreadIncarnation:
         return self.comms.registry.require(self.commander).incarnation
 
+    @property
+    def actor_name(self) -> str:
+        return self.actor.acting_name(self.commander)
+
+    @property
+    def actor_identity(self) -> ThreadIncarnation:
+        return self.actor.as_thread(self.commander).identity(self.comms)
+
     def post(self, audience: VerdictAudience, body: str,
              answers: tuple[MessageReference, ...] = ()) -> str:
         target = audience.threads[0] if len(audience.threads) == 1 else '#comms'
-        published = self.comms.messaging.send(self.commander, target,
+        published = self.comms.messaging.send(self.actor_name, target,
             audience.addressed(target, body), answers=answers)
         return published.message.message_id
 
     def call(self, tool):
-        return self._caller.invoke(tool, self.comms)
-
-    def commanding(self, caller: ThreadCaller) -> Self:
-        bound = self.model_copy(update={'commander': caller.name})
-        bound._caller = caller
-        return bound
-
-    def retire(self) -> None:
-        self._caller.retire(self.comms)
+        return self.actor.as_thread(self.commander).invoke(tool, self.comms)
 
     def todos(self):
         return self.comms.goals.todos.for_commander(self.incarnation)
@@ -294,13 +296,14 @@ def held_orders(self) -> tuple[Todo, ...]:
         return tuple(todo for todo in self.todos() if todo.holder)
 
     def held_by(self, unit: str) -> tuple[Todo, ...]:
-        return self.comms.goals.todos.held_by(self.incarnation, unit)
+        return self.comms.goals.todos.held_by(
+            self.incarnation, self.comms.registry.require(unit).incarnation)
 
     def units(self) -> frozenset[str]:
         return self.comms.goals.todos.units(self.incarnation)
 
     def progress(self) -> dict[str, OrderProgress]:
-        listed: TodoList = self.call(CommsTodoListTool())
+        listed: TodoList = self.call(CommsTodoListTool(thread=self.commander))
         return listed.progress
 
     def importers(self, modules: set[str], check: OrderCheck) -> list[str]:
@@ -433,7 +436,7 @@ def send(self, unit, document: OrderFile, clean=False,
             problems.append(f'{unit} already holds orders '
                             f'{", ".join(todo.id for todo in busy)}; '
                             f'running units holding no order: {", ".join(idle) or "none"}')
-        if not group or any(not re.fullmatch(r'P[1-9]\d*', pid) for pid in group):
+        if not group or any(not ProblemNumber.accepts(pid) for pid in group):
             problems.append('problems: fix: supply an array of P<n> IDs, or use order problem')
         if new and document.problems:
             problems.append('problems: fix: omit IDs when creating a new problem')
@@ -508,18 +511,19 @@ def send(self, unit, document: OrderFile, clean=False,
         if new:
             self.call(CommsTodoCreateTool(owner=unit, order=CreateProblemGoalAction(
                 problem=new[0], repo=Repo.of(Path(check.directory)).github, text=new[1],
-                order=order, request=request, review_at=review_at, commander=self.commander)))
+                order=order, request=request, review_at=review_at, commander=self.commander,
+                assigner=self.actor_identity)))
         for pid in [] if new else group:
             self.call(CommsOrderTool(owner=unit, order=OrderGoalAction(
                 problem=pid, expected_revision=self.comms.goals.todos.get(pid).revision,
-                assigner=self.incarnation, traced=traced,
+                assigner=self.actor_identity, traced=traced,
                 order=order, request=request, review_at=review_at, commander=self.commander)))
         message += "\n" + "\n".join(line for pid in group for line in order.scope.instructions(pid))
         sent = self.post(VerdictAudience((unit,)), message, answers)
         # The message is each order's recorded link; a verdict citing it lands on that order.
         for pid in group:
             self.call(CommsChangeProblemTool(problem=pid, operation=OrderMessage(
-                self.comms.goals.todos.get(pid).revision, self.incarnation, sent)))
+                self.comms.goals.todos.get(pid).revision, self.actor_identity, sent)))
         for other, text in taken.items():
             self.post(VerdictAudience((other,)), text)
         print(f'{measured.summary}; review timer {review_at.clock()}')
@@ -583,7 +587,7 @@ def board_text(self) -> str:
                  '|---|---|---|---|---|---|---|---|---|---|']
         listed, read = self.todos(), self.progress()
         activity = {unit: self.last_activity(unit) for unit in {t.holder for t in listed if t.holder}}
-        for t in sorted(listed, key=lambda t: int(t.id[1:]) if t.id[1:].isdigit() else 0):
+        for t in sorted(listed, key=lambda t: ProblemNumber.parse(t.id).number):
             lines.append(f'| {t.id}: {t.text.replace("|", "&#124;")} | {t.order.board_cells(t)} | '
                          f'{read[t.id].line if t.id in read else "-"} | {t.order.line} | '
                          f'{t.strike_count} | {activity[t.holder].clock() if t.holder else "-"} |')
@@ -618,7 +622,7 @@ def strike(self, unit: str, task: str, why: str, caught: str = 'review', *,
         todo.state.require_open(todo)
         if todo.held().owner != unit:
             sys.exit(f'Refused: {unit} does not hold {task}; strike the unit that holds it.')
-        self.recorded(unit, todo, kind(todo.revision, self.incarnation, todo.held(),
+        self.recorded(unit, todo, kind(todo.revision, self.actor_identity, todo.held(),
                                       f'{why} (caught by {caught})',
                                       evidence=evidence), answers)
 
@@ -751,11 +755,12 @@ class CommanderVerb(CommanderRole, DeclaredVerb, affix="Verb"):
     ] = NoThreadCaller()
 
     def who(self, ctx: Commander) -> str:
-        return self.fork.acting_name(ctx.commander)
+        return ctx.actor_name
 
     def apply(self, ctx: Commander) -> None:
         self.require_commander(ctx.comms.registry.require(ctx.commander))
-        ctx = ctx.commanding(self.fork.commanding(ctx.comms, ctx.commander))
+        self.fork.commanding(ctx.comms, ctx.commander)
+        ctx = ctx.model_copy(update={'actor': self.fork})
         self.gate(ctx)
         self.run(ctx)
 
@@ -936,12 +941,10 @@ def gate(self, ctx: Commander) -> None:
         )
 
     def run(self, ctx: Commander) -> None:
-        ThreadCaller(self.who(ctx)).invoke(
-            CommsOrderDeliverTool(
-                problem=self.problem, pull_request=self.pull_request,
-                head=self.head, report=Path(self.report_file).read_text(),
-            ), ctx.comms,
-        )
+        ctx.call(CommsOrderDeliverTool(
+            problem=self.problem, pull_request=self.pull_request,
+            head=self.head, report=Path(self.report_file).read_text(),
+        ))
         print(f'{self.problem}: delivery claim recorded')
 
 
@@ -1102,8 +1105,8 @@ class RetireVerb(OrderVerb):
     help = "stop a commander fork without discarding its open orders"
 
     def run(self, ctx: Commander) -> None:
-        ctx.retire()
-        print(f'{ctx.commander} retired')
+        ctx.actor.retire(ctx.comms)
+        print(f'{ctx.actor_name} retired')
 
 
 @dataclass(frozen=True, kw_only=True)

src/agent_comms/commander_parity/deletion.py

@@ -254,6 +254,7 @@ def send(cls, run: LifecycleRun, document: OrderFile) -> Todo:
                     ),
                     review_at=RecordedEventTime(time.time() + 3600),
                     commander=run.commander.commander,
+                    assigner=run.commander.actor_identity,
                 ),
             )
         )

src/agent_comms/commander_parity/story.py

@@ -58,6 +58,7 @@ def exercise(cls, run: LifecycleRun) -> tuple[Todo, ...]:
                     .id,
                     review_at=saved.review_at,
                     commander=run.commander.commander,
+                    assigner=run.commander.actor_identity,
                     repo=saved.repo,
                     text="The second problem shares the same real sent order.",
                 ),

src/agent_comms/goal_actions.py

@@ -5,10 +5,24 @@
 from abc import abstractmethod
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field, replace
-from typing import TYPE_CHECKING, ClassVar, Annotated
+from typing import TYPE_CHECKING, Annotated, ClassVar
 from uuid import uuid4
 
 from agent_comms import field_codec
+from agent_comms.goals import (
+    Goal,
+    GoalChangedError,
+    GoalCheckpoint,
+    GoalDeclaration,
+    GoalOrder,
+    NoGoal,
+    ProblemGoalOrder,
+    ReportedTurn,
+    RequestDirective,
+    RequestGoalOrder,
+    UnobservedGoalCheckpoint,
+    UnorderedGoalOrder,
+)
 from agent_comms.input_disposition import InputDispositions
 
 from .change_report import FileLine, NoSourceReference
@@ -28,20 +42,6 @@
     RuntimePause,
 )
 from .goal_waits import GoalWaits, RecordedGoalWait
-from agent_comms.goals import (
-    Goal,
-    GoalCheckpoint,
-    GoalChangedError,
-    GoalDeclaration,
-    GoalOrder,
-    NoGoal,
-    ProblemGoalOrder,
-    ReportedTurn,
-    RequestDirective,
-    RequestGoalOrder,
-    UnorderedGoalOrder,
-    UnobservedGoalCheckpoint,
-)
 from .progress_claims import ProgressClaim
 from .recorded_time import EventTime
 from .repository_name import RepositoryName
@@ -507,9 +507,12 @@ class ProblemOrderGoalAction(GoalAction[Goal], RuntimeInvocable):
     request: str
     review_at: EventTime
     commander: str
+    assigner: ThreadIncarnation
 
-    @abstractmethod
-    def bound_to(self, actor: ThreadIncarnation) -> ProblemOrderGoalAction: ...
+    def bound_to(
+        self, actor: ThreadIncarnation, commander: ThreadIncarnation,
+    ) -> ProblemOrderGoalAction:
+        return replace(self, commander=commander.name, assigner=actor)
 
     @abstractmethod
     def reserve(self, ctx: GoalActionContext, generation: str, goal: GoalRef) -> Todo: ...
@@ -556,11 +559,6 @@ class OrderGoalAction(ProblemOrderGoalAction):
     traced: Annotated[FileLine | NoSourceReference, field_codec.WireOmitDefault()] = (
         NoSourceReference()
     )
-    assigner: ThreadIncarnation
-
-    def bound_to(self, actor: ThreadIncarnation) -> OrderGoalAction:
-        return replace(self, commander=actor.name, assigner=actor)
-
     def reserve(self, ctx: GoalActionContext, generation: str, goal: GoalRef) -> Todo:
         return ctx.goals.todos.assign(self.problem, expected_revision=self.expected_revision,
             owner=ctx.thread, assigner=self.assigner,
@@ -585,13 +583,11 @@ class CreateProblemGoalAction(ProblemOrderGoalAction):
     repo: RepositoryName
     text: str
 
-    def bound_to(self, actor: ThreadIncarnation) -> CreateProblemGoalAction:
-        return replace(self, commander=actor.name)
-
     def reserve(self, ctx: GoalActionContext, generation: str, goal: GoalRef) -> Todo:
         return ctx.goals.todos.create(self.problem, self.repo, self.text,
             creator=ctx.goals.registry.require(self.commander), owner=ctx.thread,
-            generation=generation, goal=goal, review_at=self.review_at, order=self.order)
+            assigner=self.assigner, generation=generation, goal=goal,
+            review_at=self.review_at, order=self.order)
 
 
 @dataclass(frozen=True, kw_only=True)

src/agent_comms/goal_management.py

@@ -40,6 +40,7 @@
     from .message_reference import MessageReference
     from .native_input_owner import RegistryOwner
     from .registry_document import RegistrySnapshot
+    from .thread_identity import ThreadIncarnation
 from .goal_presentation import (
     GoalPublicationSnapshot,
     GoalSnapshot,
@@ -97,11 +98,11 @@ def send_back(self, problem: str, verdict: SendBackVerdict) -> Todo:
         sent_back: Todo = self.todos.get(problem)
         return sent_back
 
-    def fire(self, unit: str, reason: StrikeReason, commander: str) -> tuple[Todo, ...]:
-        actor = self.registry.require(commander).incarnation
+    def fire(self, unit: str, reason: StrikeReason, actor: ThreadIncarnation) -> tuple[Todo, ...]:
+        owner = self.registry.require(unit).incarnation
         store: TodoStore = self.todos
         with _store_lock(self._wire_lock_path):
-            return store.fire(unit, actor, reason)
+            return store.fire(owner, actor, reason)
 
     def open_goal_store(self) -> GoalAttemptStore:
         if self.goal_store is None:

src/agent_comms/messaging.py

@@ -78,7 +78,7 @@ def send(
             publisher: OwnerPublisher = self.bus.publisher
             committed = publisher.publish_ordinary(message, origin=AgentOrigin())
         try:
-            todos.correct(owner.name, committed.message_id, body)
+            todos.correct(owner.incarnation, committed.message_id, body)
         except TodoError as error:
             correction = CorrectionFailed(f'{error.__class__.__name__}: {error}')
         else:

src/agent_comms/owner_lifecycle.py

@@ -309,7 +309,7 @@ def _register_owner_unlocked(self, previous: Thread, identity: ProcessIdentity)
             orders = tuple(
                 todo
                 for commander in CommanderHeartbeat.commanders(self.registry, self.goals)
-                for todo in self.goals.todos.held_by(commander.incarnation, owned.name)
+                for todo in self.goals.todos.held_by(commander.incarnation, owned.incarnation)
                 if todo.pending and todo.held().owner_identity == owned.incarnation
             )
             if orders: