Separate owner requests from agent goals on the SQL bus

src/agent_comms/bus_publication.py

@@ -185,15 +185,18 @@ def require_source(self, source: BusReader, sender: str, message_id: str) -> Non
 
         with source.observing() as db:
             if MessageIdentitySource.exists(db, sender=sender, message_id=message_id):
-                raise HumanSendRefusedError("Human message identity was already published.")
+                raise HumanSendRefusedError(RelationViolationError(
+                    "Human message identity was already published."))
 
     def append(self, bus: Bus, build: RecordBuilder) -> BusRecord:
         try:
             return bus.append(build)
-        except CommittedMessageError:
+        except (CommittedMessageError, HumanSendRefusedError):
             raise
-        except Exception as error:
-            raise HumanSendRefusedError(f"Message was not sent: {error}") from error
+        except (OSError, sqlite3.Error, ValueError) as error:
+            # The transaction owns rollback for storage and declared-data refusals.
+            # Programming errors propagate as defects, not ordinary send refusals.
+            raise HumanSendRefusedError(error) from error
 
     def require_registered(self, snapshot: RegistrySnapshot) -> None:
         """Check the determining USER declaration, never a transport or prose role."""

src/agent_comms/channel_targets.py

@@ -146,7 +146,11 @@ class NamedTarget(CommandTarget):
     name: str
     sigil: ClassVar[str]
     batchable = False
-    direct_names = ()
+
+    @property
+    def direct_names(self) -> tuple[str, ...]:
+        return ()
+
     guidance_scope = ""
     default_audience = MessageAudience.DIRECT
 
@@ -241,7 +245,7 @@ def require_frozen_recipients(self, message, audience):
         message.role.require_recipients(self, audience)
 
     @property
-    def direct_names(self):
+    def direct_names(self) -> tuple[str, ...]:
         return self.names
 
     def canonical_name(self, canonical):

src/agent_comms/commander/judge.py

@@ -374,7 +374,7 @@ def check_edits(self, edits: list[Edit]) -> None:
 
     def report(self, ctx: order.Commander, todos: tuple[Todo, ...]) -> None:
         for todo in todos:
-            audience = todo.verdicts[-1].audience.of(todo.held().owner, todo.goal)
+            audience = todo.verdicts[-1].audience.of(todo.held().owner, todo)
             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.actor_name} at review): '

src/agent_comms/commander/order.py

@@ -54,6 +54,7 @@
 from agent_comms.message_reference import MessageReference
 from agent_comms.messages import Message
 from agent_comms.native_tool_call import SelectedToolDenied
+from agent_comms.owner_requests import NewestOpenRequest, OwnerRequestId, RequestSelection
 from agent_comms.plan_phases import OrderPhase
 from agent_comms.problem_number import ProblemNumber
 from agent_comms.recorded_time import RecordedEventTime
@@ -435,9 +436,8 @@ def message(self, context: CommanderContext, unit: str, group: Sequence[str],
 
     def record(self, context: CommanderContext, unit: str, group: Sequence[str],
                target: OrderedProblems, traced: FileLine | NoSourceReference,
-               review_at: RecordedEventTime) -> None:
+               review_at: RecordedEventTime, request: OwnerRequestId) -> None:
         """The native tools record the typed order; the document supplies no second store."""
-        request = self.order.request(context.comms.registry.require(context.commander), tuple(group))
         target.record(context, self.order, unit, group, traced, request, review_at)
 
     def send_back(self, context: Commander, unit: str, group: Sequence[str],
@@ -496,10 +496,9 @@ def actor_identity(self) -> ThreadIncarnation:
 
     def post(self, audience: VerdictAudience, body: str,
              answers: tuple[MessageReference, ...] = ()) -> str:
-        target = (audience.threads[0] if len(audience.threads) == 1
-                  else CommsTag.channel())
-        published = self.comms.messaging.send(self.actor_name, target,
-            audience.addressed(target, body), answers=answers)
+        target = audience.target
+        published = self.comms.messaging.send(self.actor_name, target.name,
+            target.addressed(audience.threads, body), answers=answers)
         return published.message.message_id
 
     def call(self, tool):
@@ -579,7 +578,7 @@ def replacing(self, unit: str, group: Sequence[str], listed: Mapping[str, Todo])
 
     @abstractmethod
     def record(self, context: CommanderContext, order: PendingProblemOrder, unit: str,
-               group: Sequence[str], traced: FileLine | NoSourceReference, request: str,
+               group: Sequence[str], traced: FileLine | NoSourceReference, request: OwnerRequestId,
                review_at: RecordedEventTime) -> None: ...
 
 
@@ -601,7 +600,7 @@ def replacing(self, unit: str, group: Sequence[str], listed: Mapping[str, Todo])
         return False
 
     def record(self, context: CommanderContext, order: PendingProblemOrder, unit: str,
-               group: Sequence[str], traced: FileLine | NoSourceReference, request: str,
+               group: Sequence[str], traced: FileLine | NoSourceReference, request: OwnerRequestId,
                review_at: RecordedEventTime) -> None:
         context.call(CommsTodoCreateTool(owner=unit, order=CreateProblemGoalAction(
             problem=self.problem, repo=Repo.of(Path(order.check.directory)).github,
@@ -625,7 +624,7 @@ def replacing(self, unit: str, group: Sequence[str], listed: Mapping[str, Todo])
         return bool(group) and all(pid in listed and listed[pid].holder == unit for pid in group)
 
     def record(self, context: CommanderContext, order: PendingProblemOrder, unit: str,
-               group: Sequence[str], traced: FileLine | NoSourceReference, request: str,
+               group: Sequence[str], traced: FileLine | NoSourceReference, request: OwnerRequestId,
                review_at: RecordedEventTime) -> None:
         for pid in group:
             context.call(CommsOrderTool(owner=unit, order=OrderGoalAction(
@@ -736,7 +735,7 @@ def send(self, unit: str, document: OrderFile, clean: bool = False,
              reassign: Reassignment = NoReassignment(), target: OrderedProblems = ListedProblems(),
              traced: FileLine | NoSourceReference = NoSourceReference(),
              disk: bool = False, finding: ProblemFinding = UnlocatedProblemFinding(),
-             answers: tuple[MessageReference, ...] = ()) -> None:
+             answers: tuple[MessageReference, ...] = (), *, request: RequestSelection) -> None:
         """Record the decoded order, or report every failed prerequisite before any write."""
         document = document.resolved(self, finding)
         group = target.group(document)
@@ -748,7 +747,8 @@ def send(self, unit: str, document: OrderFile, clean: bool = False,
         measured = MeasuredReview.read(self.comms.goals.todos)
         review_at = measured.review_at(document.order)
         message = document.message(self, unit, group, replacing, reassign)
-        document.record(self, unit, group, target, traced, review_at)
+        request_id = request.for_order(self.comms.goals.todos, self.incarnation)
+        document.record(self, unit, group, target, traced, review_at, request_id)
         sent = self.post(VerdictAudience((unit,)), message, answers)
         for pid in group:
             self.call(CommsChangeProblemTool(problem=pid, operation=OrderMessage(
@@ -781,20 +781,6 @@ def say(self, units: str, body_file: str,
         refuse(body, *OUTGOING)
         print(self.post(VerdictAudience(tuple(units.split(','))), body.rstrip(), answers))
 
-    def problem(self, order_file: str, unit: str, where: str, disk: bool = False,
-                answers: tuple[MessageReference, ...] = ()) -> None:
-        """A problem is born owned: Core records it with its order in one step, or not at all."""
-        document = OrderFile.read(Path(order_file).read_text())
-        location = FileLine.from_wire(where)
-        checkout = self.repo_directory(document.order.check.directory)
-        if checkout.is_dir():
-            self.comms.goals.todos.require_recurrence_declaration(
-                document.order, location, checkout, document.slop, time.time())
-        pid = self.comms.goals.todos.allocate_problem()
-        self.send(unit, document, target=NewProblem(pid, document.defect),
-                  disk=disk, finding=LocatedProblemFinding(location), answers=answers)
-        print(pid)
-
     def close(self, pid, works, artifact=''):
         """Core owns closing: the installed evidence, the artifact and the no-open-order rule live on the Todo."""
         evidence = artifact or self.comms.goals.todos.get(pid).order.closing_artifact()
@@ -847,7 +833,7 @@ def posted(self, unit: str, todo: Todo, strike: StrikeTodo,
         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)
+        self.on_record(line, 'Strike', VerdictAudience.of(unit, todo), 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: str, delivery: OrderDelivery) -> None:
@@ -899,7 +885,7 @@ def unstrike(self, task: str, why: str, at: StrikeSelector = NewestStrike()) ->
         n = struck(self.comms.goals.todos.get(task), unit)
         line = (f'{time.strftime("%Y-%m-%d %H:%M")} {unit} (strike voided on {task}, '
                 f'recorded {voided.at.clock()}; {n} strike(s) remain): {why}')
-        self.on_record(line, 'Void', VerdictAudience.of(unit, todo.goal))
+        self.on_record(line, 'Void', VerdictAudience.of(unit, todo))
         print(f'{unit}: {n} strike(s) on {task}')
 
     def withdraw(self, pid: str, why: str, superseded_by: Successor = NoSuccessor()) -> None:
@@ -916,7 +902,7 @@ def withdraw(self, pid: str, why: str, superseded_by: Successor = NoSuccessor())
         unit = previous.owner
         withdrawal, = self.comms.goals.todos.get(pid).order.withdrawals(todo)
         self.on_record(f'{time.strftime("%Y-%m-%d %H:%M")} {withdrawal}', 'Withdrawal',
-                  VerdictAudience.of(unit, todo.goal))
+                  VerdictAudience.of(unit, todo))
         self.post(VerdictAudience((unit,)),
              f'Withdrawn: {withdrawal}. Stop that work; it counts against no one.')
         print(withdrawal)
@@ -1004,7 +990,15 @@ class OrderVerb(CommanderVerb):
 
 
 @dataclass(frozen=True, kw_only=True)
-class SendVerb(OrderVerb, DirectiveVerb):
+class RequestedOrderVerb(OrderVerb, DirectiveVerb, declared_name=None):
+    request: Annotated[
+        RequestSelection, Argument(flags=("--request",),
+                                   options={"help": "owner request ID; defaults to newest open"}),
+    ] = NewestOpenRequest()
+
+
+@dataclass(frozen=True, kw_only=True)
+class SendVerb(RequestedOrderVerb):
     help = "record the order in Core and send it"
     unit: Annotated[str, Argument(flags=("unit",))]
     body_file: Annotated[
@@ -1043,7 +1037,7 @@ class SendVerb(OrderVerb, DirectiveVerb):
     def respond(self, ctx: Commander, references: tuple[MessageReference, ...]) -> None:
         ctx.send(self.unit, OrderFile.read(Path(self.body_file).read_text()), clean=self.clean,
              reassign=self.reassign, traced=self.traced,
-             disk=self.disk, answers=references)
+             disk=self.disk, answers=references, request=self.request)
 
 
 @dataclass(frozen=True, kw_only=True)
@@ -1284,7 +1278,7 @@ def run(self, ctx: Commander) -> None:
 
 
 @dataclass(frozen=True, kw_only=True)
-class ProblemVerb(OrderVerb, DirectiveVerb):
+class ProblemVerb(RequestedOrderVerb):
     help = "Core records the problem with its order, or nothing"
     order: Annotated[
         str, Argument(flags=("order",), options={"metavar": "ORDER.json"})
@@ -1299,8 +1293,17 @@ class ProblemVerb(OrderVerb, DirectiveVerb):
     disk: Annotated[bool, Argument(flags=("--disk",))] = False
 
     def respond(self, ctx: Commander, references: tuple[MessageReference, ...]) -> None:
-        ctx.problem(self.order, self.to, self.where, self.disk,
-                    references)
+        document = OrderFile.read(Path(self.order).read_text())
+        location = FileLine.from_wire(self.where)
+        checkout = ctx.repo_directory(document.order.check.directory)
+        if checkout.is_dir():
+            ctx.comms.goals.todos.require_recurrence_declaration(
+                document.order, location, checkout, document.slop, time.time())
+        pid = ctx.comms.goals.todos.allocate_problem()
+        ctx.send(self.to, document, target=NewProblem(pid, document.defect), disk=self.disk,
+                 finding=LocatedProblemFinding(location), answers=references,
+                 request=self.request)
+        print(pid)
 
 
 @dataclass(frozen=True, kw_only=True)

src/agent_comms/commander/request.py

@@ -1,71 +1,67 @@
-"""What the owner asked for, recorded as the commander's own Goal so no request is forgotten.
+"""Record, list and complete independent owner requests.
 
-`request add "TEXT"` records one and prints its id; `request complete ID --why "..."` closes
-it, refused while an order serving it is open; `request list` shows the open request with its
-open orders. A thread holds one Goal, so one request is open at a time. Every order names the
-request it serves (`order send|problem --request ID`); rotation prints REQUEST STALLED for an
-open request that no open order serves.
+Every request keeps its own id and serving orders. Completing one request never
+replaces the commander's current goal or another request.
 """
 
 from __future__ import annotations
 
-from typing import Annotated
 from dataclasses import dataclass
+from functools import partial
+from typing import Annotated
 
 from agent_comms import field_codec
+from agent_comms.commander.order import Commander, CommanderVerb
+from agent_comms.message_bus import HeldBusWrite
+from agent_comms.owner_requests import OwnerRequestId
 
-from ..goal_actions import CompletedGoalAction, GoalPrecondition, SetGoalAction
-from ..goal_states import ActiveGoal
-from ..goals import Goal, RequestGoalOrder
-from ..todos import Todo
-from .order import Commander, CommanderVerb
 
+class RequestVerb(CommanderVerb):
+    """A verb of `request`."""
 
-def open_requests(ctx: Commander) -> tuple[tuple[Goal, tuple[Todo, ...]], ...]:
-    """The commander's open owner requests and their undelivered orders."""
-    thread = ctx.comms.registry.require(ctx.commander)
-    return tuple(request for goal in thread.goal.active()
-                 for request in goal.order.open_requests(goal, ctx.comms.goals, thread.incarnation))
 
+class RequestMutation(HeldBusWrite, RequestVerb, declared_name=None):
+    """Request mutations use the same bus writer as orders and directives."""
 
-class RequestVerb(CommanderVerb):
-    """A verb of `request`."""
+    def apply(self, ctx: Commander) -> None:
+        self.write_when_open(ctx.comms, partial(super().apply, ctx))
 
 
 @dataclass(frozen=True, kw_only=True)
-class RequestAddVerb(RequestVerb, declared_name="add"):
-    help = "record an owner request as the commander's Goal; prints its id"
+class RequestAddVerb(RequestMutation, declared_name="add"):
+    help = "record an owner request; prints its id"
     text: Annotated[str, field_codec.Argument(flags=("text",))]
 
     def run(self, ctx: Commander) -> None:
-        goal = ctx.comms.goals.update_goal(ctx.commander,
-                                          SetGoalAction(text=self.text, order=RequestGoalOrder()))
-        print(goal.require().id)
+        commander = ctx.comms.registry.require(self.who(ctx)).incarnation
+        request = ctx.comms.goals.todos.add_request(commander, self.text)
+        print(request.id.value)
 
 
 @dataclass(frozen=True, kw_only=True)
-class RequestCompleteVerb(RequestVerb, declared_name="complete"):
-    help = "close a request; refused while an order serving it is open"
-    id: Annotated[str, field_codec.Argument(flags=("id",))]
+class RequestCompleteVerb(RequestMutation, declared_name="complete"):
+    help = "close a request; refused while an order serving it is pending"
+    id: Annotated[OwnerRequestId, field_codec.Argument(flags=("id",))]
     why: Annotated[str, field_codec.Argument(flags=("--why",))]
 
     def run(self, ctx: Commander) -> None:
-        expect = GoalPrecondition(goal_id=self.id, expected_state=ActiveGoal())
-        ctx.comms.goals.update_goal(ctx.commander,
-                                   CompletedGoalAction(progress=self.why, expect=expect))
-        print(f"{self.id} done")
+        commander = ctx.comms.registry.require(self.who(ctx)).incarnation
+        ctx.comms.goals.todos.complete_request(commander, self.id, self.why)
+        print(f"{self.id.value} done")
 
 
 class RequestListVerb(RequestVerb, declared_name="list"):
-    help = "the open request and the open orders serving it"
+    help = "every open request and its pending orders"
 
     def run(self, ctx: Commander) -> None:
-        requests = open_requests(ctx)
+        commander = ctx.comms.registry.require(self.who(ctx)).incarnation
+        store = ctx.comms.goals.todos
+        requests = store.open_requests(commander)
         if not requests:
             print("No open request.")
-        for goal, served in requests:
-            print(f"{goal.id} {goal.text}")
-            for todo in served:
+        for request in requests:
+            print(f"{request.id.value} {request.text}")
+            for todo in store.serving_orders(request.id):
                 print(f"  {todo.id} review {todo.review_at.clock()}: {todo.text}")
 
 

src/agent_comms/commander/standing.py

@@ -15,7 +15,6 @@
 from agent_comms.commander_heartbeat import CommanderHeartbeat
 from agent_comms.comms import Comms
 from agent_comms.goal_actions import ReissuedOrderGoalAction
-from agent_comms.goals import Goal
 from agent_comms.message_bus import WatchPass
 from agent_comms.message_page import MessageReadRequest, PageBounds
 from agent_comms.messages import Message
@@ -35,7 +34,7 @@
 from agent_comms.storage_rules import OrderCheckoutWork
 from agent_comms.thread_identity import ThreadIncarnation
 from agent_comms.threads import Thread
-from agent_comms.todos import Todo
+from agent_comms.todos import OwnerRequest, Todo
 from agent_comms.turn_digests import OrderEditCountRead, TurnDigestRead, TurnsSince
 from agent_comms.ui_model.unit_state import AwaitingCommander, UnitState
 from agent_comms.work_dispatch import ReusedWorkDispatch, WorkDispatch
@@ -81,18 +80,17 @@ def line(self) -> str:
 
 @dataclass(frozen=True)
 class RequestStalled(IndependentStandingRule, UndatedStandingRule):
-    goal: Goal
+    request: OwnerRequest
 
     @classmethod
     def violations(cls, comms: Comms, commander: str) -> tuple[RequestStalled, ...]:
         thread = comms.registry.require(commander)
-        return tuple(cls(request) for goal in thread.goal.active()
-                     for request, served in goal.order.open_requests(
-                         goal, comms.goals, thread.incarnation)
-                     if not served)
+        store = comms.goals.todos
+        return tuple(cls(request) for request in store.open_requests(thread.incarnation)
+                     if not store.serving_orders(request.id))
 
     def line(self) -> str:
-        return (f"REQUEST STALLED: {self.goal.id} {self.goal.text}"
+        return (f"REQUEST STALLED: {self.request.id.value} {self.request.text}"
                 " -> order its next step or mark it done")
 
 
@@ -191,7 +189,8 @@ def scan_observed(cls, comms: Comms, commander: str,
 
     @classmethod
     def read_open(cls, source: BusReader, snapshot: RegistrySnapshot,
-                  held: Mapping[ThreadIncarnation, Sequence[Todo]], commander: str) -> tuple[Self, ...]:
+                  held: Mapping[ThreadIncarnation, Sequence[Todo]], commander: str,
+                  requests: Sequence[OwnerRequest]) -> tuple[Self, ...]:
         return tuple(cls(message, owner.name)
                      for owner, questions in UnansweredUnitAsks.for_orders(
                          source, snapshot, commander, held, time.time(),