P440: keep the assigner typed so reissues and fork verdicts act under their commander (supersedes #2540)

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]:
@@ -317,6 +320,24 @@ def last_activity(self, unit):
         thread = self.comms.registry.require(unit)
         return thread.runtime.history_source(self.comms.root, thread.incarnation).last_tool_call()
 
+    def board_text(self) -> str:
+        """Render the live Todo view without saving a competing board file."""
+        lines = ['# Fleet board', '',
+                 "Live view of Core's Todo store; no board file is saved.", '',
+                 '| Problem | Owner file | Owner | Order | Due | Check | Progress | Order state '
+                 '| Strikes | Owner last acted |',
+                 '|---|---|---|---|---|---|---|---|---|---|']
+        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: ProblemNumber.parse(t.id).number):
+            lines.append(f'| {t.id}: {t.text.replace("|", "|")} | {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 "-"} |')
+        history = self.comms.goals.todos.history_for_commander(self.incarnation)
+        lines += ['', '## Withdrawn orders', '',
+                  *(f'- {line}' for t in history for line in t.order.withdrawals(t))]
+        return '\n'.join(lines) + '\n'
+
 
 class Reassignment(SubclassUnion, affix="Reassignment"):
     """Why a held problem changes owner, or no change of owner."""
@@ -433,7 +454,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 +529,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()}')
@@ -573,24 +595,6 @@ def close(self, pid, works, artifact=''):
         self.call(CommsProblemCloseTool(problem=pid, works=works, artifact=evidence))
         print(f'{pid} closed')
 
-    def board_text(self) -> str:
-        """Render the live Todo view without saving a competing board file."""
-        lines = ['# Fleet board', '',
-                 "Live view of Core's Todo store; no board file is saved.", '',
-                 '| Problem | Owner file | Owner | Order | Due | Check | Progress | Order state '
-                 '| Strikes | Owner last acted |',
-                 '|---|---|---|---|---|---|---|---|---|---|']
-        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):
-            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 "-"} |')
-        history = self.comms.goals.todos.history_for_commander(self.incarnation)
-        lines += ['', '## Withdrawn orders', '',
-                  *(f'- {line}' for t in history for line in t.order.withdrawals(t))]
-        return '\n'.join(lines) + '\n'
-
     def on_record(self, line: str, recorded: str, audience: VerdictAudience,
                   answers: tuple[MessageReference, ...] = ()) -> None:
         """Publish what Core already recorded; no second narrative file.
@@ -617,7 +621,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)
 
@@ -750,11 +754,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)
 
@@ -935,12 +940,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/standing.py

@@ -35,7 +35,6 @@
     UndatedStandingRule,
 )
 from agent_comms.storage_rules import OrderCheckoutWork
-from agent_comms.thread_identity import CoreWatcherIncarnation, ThreadIncarnation
 from agent_comms.threads import Thread
 from agent_comms.todos import Todo
 from agent_comms.turn_digests import OrderEditCountRead, TurnDigestRead, TurnsSince
@@ -353,11 +352,11 @@ def violations(cls, comms: Comms, commander: str) -> tuple[UnownedWheelSpin, ...
         actor = comms.registry.require(commander).incarnation
         rows = tuple(todo for todo in comms.goals.todos.for_commander(actor)
                      if not todo.holder_state.current_assignments())
-        return cls.paired(comms, commander, rows, actor)
+        return cls.paired(comms, commander, rows)
 
     @classmethod
-    def paired(cls, comms: Comms, commander: str, todos: Iterable[Todo],
-               assigner: ThreadIncarnation) -> tuple[UnownedWheelSpin, ...]:
+    def paired(cls, comms: Comms, commander: str,
+               todos: Iterable[Todo]) -> tuple[UnownedWheelSpin, ...]:
         """Both the heartbeat and automatic watcher preserve work before any dispatch."""
         rows = tuple(todo for todo in todos if todo.pending)
         retained = tuple(CheckoutHeldWheelSpin(todo, work)
@@ -371,7 +370,7 @@ def paired(cls, comms: Comms, commander: str, todos: Iterable[Todo],
                 if not todo.live_strikes(unit.name):
                     idle.remove(unit)
                     found.append(ReassignedWheelSpin(todo, ReusedWorkDispatch.for_unit(
-                        ReissuedOrderGoalAction.for_todo(todo, commander, assigner), unit)))
+                        ReissuedOrderGoalAction.for_todo(todo, commander), unit)))
                     break
         covered = {rule.todo.id for rule in (*retained, *found)}
         return (*retained, *found,
@@ -428,9 +427,7 @@ def observe(self, comms: Comms) -> tuple[UnownedObservation, ...]:
             mine = (todo for todo in available if todo.creator_identity.name == name)
             observed.extend(
                 UnownedObservation(commander, rule)
-                for rule in UnownedWheelSpin.paired(
-                    comms, name, mine, CoreWatcherIncarnation.watcher()
-                )
+                for rule in UnownedWheelSpin.paired(comms, name, mine)
             )
         return tuple(observed)
 

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