fix: separate commander authority from fork attribution

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,66 @@ 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)
+        return self.actor.as_thread(self.commander).invoke(tool, self.comms)
+
+    def on_record(self, line: str, recorded: str, audience: VerdictAudience,
+                  answers: tuple[MessageReference, ...] = ()) -> None:
+        """Publish what Core already recorded; never record it again after a failed post."""
+        try:
+            self.post(audience, f'On the record: {line}', answers)
+        except (SystemExit, ValueError) as failed:
+            print(f'{recorded} recorded, but the post failed ({failed}). '
+                  'Repost with order.post; do not record it again.', file=sys.stderr)
+            raise
+
+    def strike(self, unit: str, task: str, why: str, caught: str = 'review', *,
+               evidence: tuple[FileLine, ...], kind: type[StrikeTodo] = StrikeTodo,
+               answers: tuple[MessageReference, ...] = ()) -> None:
+        """Record and publish the acting commander's failure finding on the original Todo."""
+        caught = f'{self.actor_name} at {caught}' if caught in STAGES else caught
+        if not any(stage in caught for stage in STAGES):
+            sys.exit(f'Refused: --caught must say who caught it and at which stage {STAGES}, e.g. "{self.actor_name} at review".')
+        todo = self.comms.goals.todos.get(task)
+        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.actor_identity, todo.held(),
+                                      f'{why} (caught by {caught})',
+                                      evidence=evidence), answers)
 
-    def commanding(self, caller: ThreadCaller) -> Self:
-        bound = self.model_copy(update={'commander': caller.name})
-        bound._caller = caller
-        return bound
+    def recorded(self, unit: str, todo: Todo, strike: StrikeTodo,
+                 answers: tuple[MessageReference, ...] = ()) -> None:
+        """Core records the strike, then this acting context publishes its result."""
+        refuse(strike.reason, *OUTGOING)
+        if (self.call(CommsChangeProblemTool(problem=todo.id, operation=strike))
+                .todo.revision == todo.revision):
+            return
+        self.posted(unit, self.comms.goals.todos.get(todo.id), strike, answers)
 
-    def retire(self) -> None:
-        self._caller.retire(self.comms)
+    def posted(self, unit: str, todo: Todo, strike: StrikeTodo,
+               answers: tuple[MessageReference, ...] = ()) -> None:
+        """Publish a strike Core already recorded on the task's Todo."""
+        n = struck(todo, unit)
+        line = (f'{time.strftime("%Y-%m-%d")} {strike.at.clock()} {unit} (strike {n} on {todo.id}'
+                f'{f", {todo.id} taken away" if n >= 2 else ""}): {strike.reason}')
+        self.on_record(line, 'Strike', VerdictAudience.of(unit, todo.goal), answers)
+        print(f'{unit}: {n} strike(s) on {todo.id}' + (f'. Reassign {todo.id} now and freeze its branch.' if n >= 2 else ''))
 
     def todos(self):
         return self.comms.goals.todos.for_commander(self.incarnation)
@@ -294,13 +339,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 +479,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 +554,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()}')
@@ -582,7 +629,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 "-"} |')
@@ -591,55 +638,6 @@ def board_text(self) -> str:
                   *(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.
-
-        Core already holds what `recorded` names, so a failed post is reposted, never recorded again.
-        """
-        try:
-            self.post(audience, f'On the record: {line}', answers)
-        except (SystemExit, ValueError) as failed:
-            print(f'{recorded} recorded, but the post failed ({failed}). '
-                  'Repost with order.post; do not record it again.', file=sys.stderr)
-            raise
-
-    def strike(self, unit: str, task: str, why: str, caught: str = 'review', *,
-               evidence: tuple[FileLine, ...], kind: type[StrikeTodo] = StrikeTodo,
-               answers: tuple[MessageReference, ...] = ()) -> None:
-        """HARD RULE: every failure goes on the public record, in plain words, damning and exact, so no
-        failure stays private and none is described in words that hide what happened."""
-        caught = f'{self.commander} at {caught}' if caught in STAGES else caught
-        # What matters is how a mistake happened, how often, and how and when it was caught.
-        if not any(stage in caught for stage in STAGES):
-            sys.exit(f'Refused: --caught must say who caught it and at which stage {STAGES}, e.g. "{self.commander} at review".')
-        todo = self.comms.goals.todos.get(task)
-        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(),
-                                      f'{why} (caught by {caught})',
-                                      evidence=evidence), answers)
-
-    def recorded(self, unit: str, todo: Todo, strike: StrikeTodo,
-                 answers: tuple[MessageReference, ...] = ()) -> None:
-        """Core records the strike on the task's Todo, then posts it to #comms."""
-        refuse(strike.reason,
-            *OUTGOING)  # posted to #comms, so refused before Core records the strike
-        if (self.call(CommsChangeProblemTool(problem=todo.id, operation=strike))
-                .todo.revision == todo.revision):
-            return
-        self.posted(unit, self.comms.goals.todos.get(todo.id), strike, answers)
-
-    def posted(self, unit: str, todo: Todo, strike: StrikeTodo,
-               answers: tuple[MessageReference, ...] = ()) -> None:
-        """Publish a strike Core already recorded on the task's Todo to #comms."""
-        n = struck(todo, unit)
-        line = (f'{time.strftime("%Y-%m-%d")} {strike.at.clock()} {unit} (strike {n} on {todo.id}'
-                f'{f", {todo.id} taken away" if n >= 2 else ""}): {strike.reason}')
-        self.on_record(line, 'Strike', VerdictAudience.of(unit, todo.goal), answers)
-        print(f'{unit}: {n} strike(s) on {todo.id}' + (f'. Reassign {todo.id} now and freeze its branch.' if n >= 2 else ''))
-
     def done(self, pid, sha, screenshot='', seen=''):
         """Core runs the delivery's check at its checked commit and strikes a late delivered commit
         in the same revision; this posts the strikes Core recorded."""
@@ -750,11 +748,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 +934,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 +1099,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.",
                 ),