From c99becf139c9447674a248988ad3f43fea116bb4 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Wed, 21 Oct 2026 16:02:01 -0400 Subject: [PATCH 1/3] Factor Todo assignment queries and problem number ownership --- src/agent_comms/commander/order.py | 22 ++-- src/agent_comms/problem_number.py | 36 ++++++ src/agent_comms/storage_rules.py | 3 +- src/agent_comms/todos.py | 150 ++++++++++++++----------- src/agent_comms/tool_caller.py | 8 +- tests/test_commander_forks.py | 87 +++++++++++++- tools/cutover/carry_problem_numbers.py | 33 ++++-- 7 files changed, 238 insertions(+), 101 deletions(-) create mode 100644 src/agent_comms/problem_number.py diff --git a/src/agent_comms/commander/order.py b/src/agent_comms/commander/order.py index 30ce043b73..c3f872917a 100644 --- a/src/agent_comms/commander/order.py +++ b/src/agent_comms/commander/order.py @@ -56,6 +56,7 @@ from agent_comms.recorded_time import RecordedEventTime from agent_comms.repo import Repo from agent_comms.standing_rules import HuntDue +from agent_comms.problem_number import ProblemNumber from agent_comms.storage_rules import UnitStorage from agent_comms.thread_identity import ThreadIncarnation from agent_comms.todos import ( @@ -235,11 +236,9 @@ class CommanderContext(CommanderConfig): """The selected root's bus and canonical reads, shared by command families.""" _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, @@ -269,15 +268,7 @@ def post(self, audience: VerdictAudience, body: str, 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 ThreadCaller(self.commander).invoke(tool, self.comms) def todos(self): return self.comms.goals.todos.for_commander(self.incarnation) @@ -433,7 +424,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 IDs, or use order problem') if new and document.problems: problems.append('problems: fix: omit IDs when creating a new problem') @@ -583,7 +574,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("|", "|")} | {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 "-"} |') @@ -755,7 +746,8 @@ def who(self, ctx: Commander) -> str: 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)) + ctx = ctx.model_copy(update={ + 'commander': self.fork.commanding(ctx.comms, ctx.commander).name}) self.gate(ctx) self.run(ctx) @@ -1107,7 +1099,7 @@ class RetireVerb(OrderVerb): help = "stop a commander fork without discarding its open orders" def run(self, ctx: Commander) -> None: - ctx.retire() + self.fork.retire(ctx.comms) print(f'{ctx.commander} retired') diff --git a/src/agent_comms/problem_number.py b/src/agent_comms/problem_number.py new file mode 100644 index 0000000000..821c1eaa5c --- /dev/null +++ b/src/agent_comms/problem_number.py @@ -0,0 +1,36 @@ +"""The canonical spelling and numeric value of a global problem identity.""" +from dataclasses import dataclass +from typing import ClassVar, Self + + +@dataclass(frozen=True, slots=True) +class ProblemNumber: + number: int + prefix: ClassVar[str] = 'P' + + def __post_init__(self) -> None: + if self.number < 1: + raise ValueError('A problem number is positive') + + def __str__(self) -> str: + return f'{self.prefix}{self.number}' + + @classmethod + def parse(cls, text: str) -> Self: + number = cls(int(text.removeprefix(cls.prefix))) + if str(number) != text: + raise ValueError(f'Not a canonical problem number: {text!r}') + return number + + @classmethod + def accepts(cls, text: str) -> bool: + try: + cls.parse(text) + except ValueError: + return False + return True + + @classmethod + def sql_number(cls, column: str) -> str: + """The numeric projection of an already validated stored identity.""" + return f'CAST(substr({column}, {len(cls.prefix) + 1}) AS INTEGER)' diff --git a/src/agent_comms/storage_rules.py b/src/agent_comms/storage_rules.py index 76cc70ac14..7ac014cd75 100644 --- a/src/agent_comms/storage_rules.py +++ b/src/agent_comms/storage_rules.py @@ -18,6 +18,7 @@ from pathlib import Path from typing import TYPE_CHECKING, Annotated, ClassVar, Self +from agent_comms.problem_number import ProblemNumber from agent_comms.field_codec import FieldCodec, PathText from agent_comms.native_tool_call import SelectedToolDenied from agent_comms.repo import Repo @@ -145,7 +146,7 @@ def owns_checkout(self, owner: Thread, path: Path) -> bool: if path == owner.project_path: return True problem, _, repository = path.name.removeprefix(f'{owner.name}--').partition('--') - if not re.fullmatch(r'P[1-9][0-9]*', problem): + if not ProblemNumber.accepts(problem): return False if not repository: return path == self.order_checkout(owner.name, problem) diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 3db47c3c7d..48064c1ba9 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -16,10 +16,10 @@ import sys import time from abc import ABC, abstractmethod -from collections.abc import Iterator, Sequence +from collections.abc import Callable, Iterator, Sequence from contextlib import contextmanager from dataclasses import dataclass, field, fields, replace -from functools import cache, cached_property +from functools import cache, cached_property, partial from importlib.resources import files from math import ceil from pathlib import Path @@ -78,6 +78,7 @@ from agent_comms.order_story import ProblemStory, StoryEvent, StoryLine, StorySource from agent_comms.plan_phases import OrderPhase, UnplannedPhase from agent_comms.private_bus_checkpoint import PublicMessages +from agent_comms.problem_number import ProblemNumber from agent_comms.progress_claims import MergedClaim, ProgressClaim from agent_comms.recorded_time import ( EventTime, @@ -2258,6 +2259,11 @@ class Assignment(TodoHolder): def of(cls, owner: ThreadIncarnation, parent: ThreadIncarnation, generation: str) -> Assignment: return cls(owner.name, owner.wire_birth, parent.name, parent.wire_birth, generation) + @classmethod + def sql_field(cls, name: str) -> str: + """Queries use the codec's declared wire key, not another record shape.""" + return f"json_extract({{assignment}}, '$.{FieldCodec.record_key(cls, name)}')" + def __post_init__(self) -> None: _text(self.owner, "Thread name", 128) _text(self.parent, "Thread name", 128) @@ -3032,10 +3038,11 @@ def apply(self, current): @dataclass(frozen=True, slots=True) class Todo(TypedTable): unit_index: ClassVar[ExpressionIndex] = ExpressionIndex(( - "json_extract({assignment}, '$.owner')", - "json_extract({assignment}, '$.owner_created')", "{state}", + Assignment.sql_field('owner'), Assignment.sql_field('owner_created'), '{state}', )) - indexes = (Index(("creator", "creator_created", "state")), Index(("state",)), unit_index) + problem_index: ClassVar[ExpressionIndex] = ExpressionIndex((ProblemNumber.sql_number('{id}'),)) + indexes = (Index(("creator", "creator_created", "state")), Index(("state",)), + unit_index, problem_index) id: Annotated[str, Column(primary_key=True)] repo: Annotated[RepositoryName, Column(storage=TextStorage)] @@ -3057,6 +3064,27 @@ class Todo(TypedTable): messages: tuple[OrderMessage, ...] = () prior_orders: tuple[CheckedProblemOrder, ...] = () + @classmethod + def rows(cls, db: sqlite3.Connection, where: str, parameters: tuple) -> tuple[Todo, ...]: + return tuple(cls.read(db.execute( + f"SELECT * FROM {cls.declared_name} WHERE {cls._sql(where)} ORDER BY rowid", parameters))) + + @classmethod + def open_rows(cls, db: sqlite3.Connection, where: str, parameters: tuple, + state: type[TodoState] = TodoState) -> tuple[Todo, ...]: + states = state.open_names() + selected = ','.join('?' for state in states) + return cls.rows(db, f"state IN ({selected}) AND ({where})", (*states, *parameters)) + + @classmethod + def held_by(cls, db: sqlite3.Connection, where: str, parameters: tuple, + *, unit: ThreadIncarnation) -> tuple[Todo, ...]: + """The indexed open-assignment query, shared by snapshots and writers.""" + terms = cls.unit_index.terms[:-1] + match = f"({', '.join(terms)})=({','.join('?' for term in terms)})" + return cls.open_rows(db, f'{match} AND ({where})', + (unit.name, unit.wire_birth, *parameters)) + @property def judged_calls(self) -> tuple[str, ...]: return tuple(call for verdict in self.verdicts for call in verdict.judged_calls()) @@ -3291,20 +3319,27 @@ def _participant(thread: Thread) -> ThreadIncarnation: @dataclass(frozen=True) -class ProblemNumber(TypedTable): - """Allocated global problem numbers, including reservations not yet sent.""" +class ProblemAllocation(TypedTable): + """Only issued, unsent numbers; creating a Todo removes its allocation row. + + SQLite's sequence remembers issuance, while Todo.id owns every created identity. + """ number: Annotated[int, Column(primary_key=True, auto_increment=True)] @classmethod - def reserve(cls, db: sqlite3.Connection, problem: str) -> None: - if re.fullmatch(r'P[1-9]\d*', problem): - cls(int(problem[1:])).insert_if_absent(db) + def reserve(cls, db: sqlite3.Connection, problem: ProblemNumber) -> ProblemNumber: + cls(problem.number).insert(db) + return problem @classmethod - def allocate(cls, db: sqlite3.Connection) -> str: - cursor = db.execute(f'INSERT INTO "{cls.declared_name}" DEFAULT VALUES') - return f'P{cursor.lastrowid}' + def allocate(cls, db: sqlite3.Connection) -> ProblemNumber: + number, = db.execute( + f'SELECT max(COALESCE((SELECT max({Todo.problem_index.term_sql(Todo)}) ' + f'FROM "{Todo.declared_name}"), 0), ' + 'COALESCE((SELECT seq FROM sqlite_sequence WHERE name=?), 0)) + 1', + (cls.declared_name,)).fetchone() + return cls.reserve(db, ProblemNumber(number)) @dataclass(frozen=True, slots=True) @@ -3318,7 +3353,7 @@ class TodoStore(DurableStore): root: Path filename: ClassVar[str] = "todos.sqlite3" - rows: ClassVar[tuple[type[TypedTable], ...]] = (Todo, HuntRecord, ProblemNumber) + rows: ClassVar[tuple[type[TypedTable], ...]] = (Todo, HuntRecord, ProblemAllocation) @contextmanager def writing(self) -> Iterator[sqlite3.Connection]: @@ -3342,24 +3377,18 @@ def writer_connection( def allocate_problem(self) -> str: with self.writing() as db: - return ProblemNumber.allocate(db) + return str(ProblemAllocation.allocate(db)) - @staticmethod - def require_unit_in(db: sqlite3.Connection, unit: ThreadIncarnation, + @classmethod + def require_unit_in(cls, db: sqlite3.Connection, unit: ThreadIncarnation, assigner: ThreadIncarnation) -> None: """The original open Todo owns who assigns its unit, under this transaction.""" - match = (f"({Todo.unit_index.term_sql(Todo)})=" - f"({','.join('?' for _ in Todo.unit_index.terms)})") - for state in TodoState.open_names(): - held = Todo.select(db, - where=f"{match} AND NOT (creator=? AND creator_created=?)", - parameters=(unit.name, unit.wire_birth, state, assigner.name, assigner.wire_birth), - limit=1) - for todo in held: - raise TodoError( - f"{unit.name} is assigned by {todo.creator} on open order " - f"{todo.order.checked().id} ({todo.id}); " - "finish or release that order before assigning it again.") + for todo in Todo.held_by(db, 'NOT (creator=? AND creator_created=?)', + (assigner.name, assigner.wire_birth), unit=unit): + raise TodoError( + f"{unit.name} is assigned by {todo.creator} on open order " + f"{todo.order.checked().id} ({todo.id}); " + "finish or release that order before assigning it again.") def record_hunt(self, commander: ThreadIncarnation, hunt: Hunt) -> HuntRecord: hunt.require_commander(self, commander) @@ -3433,17 +3462,14 @@ def link( assignment=source.held(), goal=source.goal, review_at=source.review_at, messages=source.messages, revision=target.revision + 1).upsert(db) - def _select(self, where: str, parameters: tuple) -> tuple[Todo, ...]: - """Rows in creation order; an absent store has none, and a read never creates one.""" + def _select(self, where: str, parameters: tuple, *, + reader: Callable[[sqlite3.Connection, str, tuple], tuple[Todo, ...]] = Todo.rows, + ) -> tuple[Todo, ...]: + """An absent store has no rows; only this boundary opens a read snapshot.""" if not self.path.exists(): return () with self.snapshot() as db: - return self._rows(db, where, parameters) - - @staticmethod - def _rows(db: sqlite3.Connection, where: str, parameters: tuple) -> tuple[Todo, ...]: - return tuple(Todo.read(db.execute( - f"SELECT * FROM {Todo.declared_name} WHERE {where} ORDER BY rowid", parameters))) + return reader(db, where, parameters) @classmethod def _one(cls, rows: tuple[Todo, ...], todo_id: str) -> Todo: @@ -3453,7 +3479,7 @@ def _one(cls, rows: tuple[Todo, ...], todo_id: str) -> Todo: @classmethod def _current(cls, db: sqlite3.Connection, todo_id: str) -> Todo: - return cls._one(cls._rows(db, "id=?", (todo_id,)), todo_id) + return cls._one(Todo.rows(db, "id=?", (todo_id,)), todo_id) def for_order(self, order_id: str) -> tuple[Todo, ...]: """A sent order can cover several linked problems; its ID is not a Todo key.""" @@ -3481,17 +3507,18 @@ def list(self, repo: RepositoryName) -> tuple[Todo, ...]: return self._select("repo=?", (repo.to_wire(),)) def for_goal(self, owner: ThreadIncarnation, goal_id: str) -> tuple[Todo, ...]: - return self._select( - "json_extract(assignment,'$.owner')=? AND json_extract(assignment,'$.owner_created')=? " - "AND json_extract(assignment,'$.generation')=?", (owner.name, owner.created_at, goal_id)) + return self._select(Assignment.sql_field('generation') + '=?', (goal_id,), + reader=partial(Todo.held_by, unit=owner)) def held_orders(self) -> tuple[Todo, ...]: - return self._open("assignment IS NOT NULL", (), OpenTodoState) + return self._select("assignment IS NOT NULL", (), + reader=partial(Todo.open_rows, state=OpenTodoState)) def reassignable(self) -> tuple[Todo, ...]: """Work Core released after its objective strikes; never a commander's released backlog. No clock releases work: a held order stays held until its commander moves it.""" - return tuple(todo for todo in self._open("assignment IS NULL", (), OpenTodoState) + return tuple(todo for todo in self._select("assignment IS NULL", (), + reader=partial(Todo.open_rows, state=OpenTodoState)) if todo.pending and todo.handed_out_on_release) def merged_orders(self, commander: ThreadIncarnation, @@ -3526,9 +3553,8 @@ def board_disagreements( def holds_pending(self, owner: ThreadIncarnation) -> bool: """Whether a problem assigned to owner pends, under any generation; its Goal stays open.""" - return any(todo.pending for todo in self._open( - "json_extract(assignment,'$.owner')=? AND json_extract(assignment,'$.owner_created')=?", - (owner.name, owner.wire_birth))) + return any(todo.pending for todo in self._select( + '1', (), reader=partial(Todo.held_by, unit=owner))) def for_request(self, commander: ThreadIncarnation, goal_id: str) -> tuple[Todo, ...]: """Every order that served one request, a Goal on the commander's thread, open or done.""" @@ -3537,7 +3563,8 @@ def for_request(self, commander: ThreadIncarnation, goal_id: str) -> tuple[Todo, "AND json_extract(goal,'$.goal_id')=?", (commander.name, commander.wire_birth, goal_id)) def for_commander(self, commander: ThreadIncarnation) -> tuple[Todo, ...]: - return self._open("creator=? AND creator_created=?", (commander.name, commander.wire_birth)) + return self._select("creator=? AND creator_created=?", (commander.name, commander.wire_birth), + reader=Todo.open_rows) def timers_up_for_commander(self, commander: ThreadIncarnation, registry: Registration, now: float) -> tuple[tuple[Todo, RecordedEventTime], ...]: @@ -3549,13 +3576,13 @@ def timers_up_for_commander(self, commander: ThreadIncarnation, registry: Regist todo, {todo.id: OrderProgress.read(todo, registry, pulls)})) def held_for_commander(self, commander: ThreadIncarnation) -> tuple[Todo, ...]: - return self._open("creator=? AND creator_created=? AND assignment IS NOT NULL", - (commander.name, commander.wire_birth)) + return self._select("creator=? AND creator_created=? AND assignment IS NOT NULL", + (commander.name, commander.wire_birth), reader=Todo.open_rows) def held_by(self, commander: ThreadIncarnation, unit: str) -> tuple[Todo, ...]: - return self._open( - "creator=? AND creator_created=? AND json_extract(assignment,'$.owner')=?", - (commander.name, commander.wire_birth, unit)) + return self._select( + f"creator=? AND creator_created=? AND {Todo.unit_index.terms[0]}=?", + (commander.name, commander.wire_birth, unit), reader=Todo.open_rows) def require_recurrence_declaration( self, order: CheckedProblemOrder, where: FileLine, root: Path, @@ -3581,12 +3608,6 @@ def units(self, commander: ThreadIncarnation) -> frozenset[str]: for assignment in (*todo.holder_state.current_assignments(), *(release.previous for release in todo.released))) - def _open(self, where: str, parameters: tuple, - state: type[TodoState] = TodoState) -> tuple[Todo, ...]: - states = state.open_names() - selected = ','.join('?' for state in states) - return self._select(f"state IN ({selected}) AND ({where})", (*states, *parameters)) - def create( self, todo_id: str, @@ -3600,6 +3621,7 @@ def create( order: PendingProblemOrder, goal: GoalRef, ) -> Todo: + number = ProblemNumber.parse(todo_id) identity, assignee = _participant(creator), _thread(owner) DirectiveText.require(text) order.require_wording() @@ -3609,11 +3631,11 @@ def create( review_at, order, goal) with self.writing() as db: self.require_unit_in(db, assignee, identity) - ProblemNumber.reserve(db, todo_id) proposed.insert_if_absent(db) current = self._current(db, todo_id) current.require_same_creation(proposed) current.require_assignment(proposed.assignment) + ProblemAllocation.delete(db, number=number.number) return current def apply(self, todo_id: str, command: TodoCommand) -> Todo: @@ -3631,8 +3653,7 @@ def correct(self, unit: str, message_id: str, text: str) -> tuple[Todo, ...]: if not self.path.exists() or not (words := tuple(PlainText.found(text))): return () with self.writing() as db: - held = (todo for todo in self._rows(db, "json_extract(assignment,'$.owner')=?", (unit,)) - if todo.state.open) + held = Todo.open_rows(db, Todo.unit_index.terms[0] + '=?', (unit,)) corrected = tuple(WordingCorrection(todo.revision, message_id, words).apply(todo) for todo in held) for todo in corrected: @@ -3643,10 +3664,9 @@ def fire(self, unit: str, actor: ThreadIncarnation, reason: StrikeReason) -> tup """Release every open problem the unit holds from this commander in one write; each records the same reason, and its order and strikes stay.""" with self.writing() as db: - held = tuple((todo, todo.held()) for todo in self._rows( - db, "json_extract(assignment,'$.owner')=? AND creator=? AND creator_created=?", - (unit, actor.name, actor.wire_birth)) - if todo.state.open) + held = tuple((todo, todo.held()) for todo in Todo.open_rows( + db, f"{Todo.unit_index.terms[0]}=? AND creator=? AND creator_created=?", + (unit, actor.name, actor.wire_birth))) if not held: raise TodoError(f"{unit} holds no open problem.") released = tuple(ReleaseTodo(todo.revision, actor, assignment, reason).apply(todo) diff --git a/src/agent_comms/tool_caller.py b/src/agent_comms/tool_caller.py index 9751c8039a..54eb469411 100644 --- a/src/agent_comms/tool_caller.py +++ b/src/agent_comms/tool_caller.py @@ -6,7 +6,7 @@ from typing import TYPE_CHECKING, Annotated, NoReturn from .cli_result import ThreadToolCatalogResult, ToolCatalogResult -from .errors import RelationViolationError +from .errors import ToolRefusedError from .field_codec import WireForm from .launch_root import LaunchRoot from .launch_runtime import CallingThreadUnavailable @@ -57,6 +57,9 @@ def subject(self, named: str) -> str: def commanding(self, comms: Comms, commander: str) -> ThreadCaller: return ThreadCaller(self.acting_name(commander)) + def retire(self, comms: Comms) -> None: + raise ToolRefusedError('Only a commander fork can retire through this verb.') + @abstractmethod def may_reconfigure(self, comms: Comms, name: str) -> bool: ... @@ -139,9 +142,6 @@ def require_directive(self, comms: Comms, body: str) -> None: def sent(self, comms: Comms, message: Message) -> None: """An agent's message is only the bus record.""" - def retire(self, comms: Comms) -> None: - raise RelationViolationError('Only a commander fork can retire through this verb.') - @dataclass(frozen=True) class CommanderForkCaller(ThreadCaller): diff --git a/tests/test_commander_forks.py b/tests/test_commander_forks.py index f6c72311e7..f55cfcd015 100644 --- a/tests/test_commander_forks.py +++ b/tests/test_commander_forks.py @@ -25,10 +25,13 @@ from agent_comms.order_runs import MechanicalOrderRun from agent_comms.recorded_time import RecordedEventTime from agent_comms.repo import Repo +from agent_comms.problem_number import ProblemNumber from agent_comms.thread_execution import ExternalThreadExecution +from agent_comms.thread_identity import RecordedThreadIncarnation +from agent_comms.tools import CommsTodoListTool from agent_comms.thread_model import DeclaredModel from agent_comms.todos import ( - DeclarationDerivation, GoalRef, PendingProblemOrder, ProblemNumber, ReleaseTodo, + DeclarationDerivation, GoalRef, PendingProblemOrder, ProblemAllocation, ReleaseTodo, Slop, StrikeReason, Todo, TodoError, ) @@ -120,6 +123,24 @@ def test_fork_orders_refuse_from_todos_and_retirement_is_explicit(fork_bus, tmp_ held = comms.goals.todos.held_for_commander(fork.incarnation) assert [todo.id for todo in held] == ['P41'] original = held[0] + worker = comms.registry.require('worker') + assert comms.goals.todos.holds_pending(worker.incarnation) + assert comms.goals.todos.for_goal(worker.incarnation, original.held().generation) == (original,) + assert comms.goals.todos.for_goal(worker.incarnation, 'another-goal') == () + other_birth = RecordedThreadIncarnation(worker.name, worker.created_at + 1) + assert not comms.goals.todos.holds_pending(other_birth) + assert comms.goals.todos.for_goal(other_birth, original.held().generation) == () + with comms.goals.todos.snapshot() as db: + queries = [] + db.set_trace_callback(queries.append) + assert Todo.held_by(db, '1', (), unit=worker.incarnation) == (original,) + db.set_trace_callback(None) + plan = db.execute('EXPLAIN QUERY PLAN ' + queries[0]).fetchall() + print('Open assignment query plan:', [tuple(row) for row in plan]) + copied = Commander.read_root(comms.root).model_copy(update={'commander': fork.name}) + assert copied.model_copy(deep=True).call(CommsTodoListTool()).todos == (original,) + result = command(comms, 'order', 'retire') + assert result.returncode != 0 and 'Only a commander fork can retire' in result.stderr second = document(tmp_path / 'second.json', replace(order, id='second-order')) refusal = 'worker is assigned by fork-one on open order first-order (P41)' for prefix in (('--fork', 'fork-two'), ()): @@ -173,6 +194,8 @@ def test_fork_orders_refuse_from_todos_and_retirement_is_explicit(fork_bus, tmp_ comms.goals.todos.apply('P41', ReleaseTodo( original.revision, fork.incarnation, original.held(), StrikeReason('The fixture releases its actual original order after the completed review.'))) + assert not comms.goals.todos.holds_pending(worker.incarnation) + assert comms.goals.todos.for_goal(worker.incarnation, original.held().generation) == () result = command(comms, 'order', 'problem', second, '--to', 'worker', '--where', 'owner.py:2') assert result.returncode == 0, result.stderr @@ -213,17 +236,73 @@ def assign(name): print(refused.stderr.strip()) -def test_retained_store_carry_preserves_orders_and_seeds_global_numbers(fork_bus): +def test_retained_store_carry_preserves_original_identities(fork_bus): from cutover.carry_problem_numbers import ProblemNumbers comms, source, order = fork_bus store = comms.goals.todos retained = store.get('P40') with store.writing() as db: - db.execute(f'DROP TABLE "{ProblemNumber.declared_name}"') + db.execute(f'DROP TABLE "{ProblemAllocation.declared_name}"') db.execute(f'DROP INDEX "{Todo.unit_index.name(Todo)}"') + db.execute(f'DROP INDEX "{Todo.problem_index.name(Todo)}"') + assert ProblemNumbers.carry(comms.root, apply=False).changed == 3 + assert ProblemNumbers.carry(comms.root, apply=True).changed == 3 + assert store.get('P40') == retained + with store.snapshot() as db: + assert ProblemAllocation.select(db) == [] + assert store.allocate_problem() == 'P41' + assert ProblemNumbers.carry(comms.root, apply=True).changed == 0 + # Upgrade the installed old schema: keep unsent P41, delete only the P40 copy. + with store.writing() as db: + db.execute(f'ALTER TABLE "{ProblemAllocation.declared_name}" RENAME TO problem_number') + db.execute('INSERT INTO problem_number(number) VALUES (40)') + db.execute(f'DROP INDEX "{Todo.problem_index.name(Todo)}"') assert ProblemNumbers.carry(comms.root, apply=False).changed == 2 assert ProblemNumbers.carry(comms.root, apply=True).changed == 2 + with store.snapshot() as db: + assert [row.number for row in ProblemAllocation.select(db)] == [41] assert store.get('P40') == retained - assert store.allocate_problem() == 'P41' + assert store.allocate_problem() == 'P42' assert ProblemNumbers.carry(comms.root, apply=True).changed == 0 + + +def test_created_ids_live_only_on_todos_and_allocation_uses_numeric_index(fork_bus): + comms, source, order = fork_bus + store = comms.goals.todos + lead = comms.registry.require('lead') + retained = store.get('P40') + issued = store.allocate_problem() + assert issued == 'P41' + created = store.create(issued, source.github, 'Create a real problem', creator=lead, + owner=comms.registry.require('other'), generation='created', goal=retained.goal, + review_at=retained.review_at, order=order) + with store.snapshot() as db: + assert ProblemAllocation.select(db) == [] + # Direct creation must affect allocation without copying its key into the allocator. + store.create('P120', source.github, 'Exercise numeric ordering', creator=lead, + owner=comms.registry.require('worker'), generation='direct', goal=retained.goal, + review_at=retained.review_at, order=replace(order, id='direct-order')) + assert store.allocate_problem() == 'P121' + with store.snapshot() as db: + assert [row.number for row in ProblemAllocation.select(db)] == [121] + plan = db.execute(f'EXPLAIN QUERY PLAN SELECT max({Todo.problem_index.term_sql(Todo)}) ' + f'FROM "{Todo.declared_name}"').fetchall() + print('Numeric identity query plan:', [tuple(row) for row in plan]) + assert store.get(issued) == created + + +def test_malformed_problem_ids_are_refused_without_writes(fork_bus): + comms, source, order = fork_bus + store = comms.goals.todos + retained = store.get('P40') + for text in ('broken', '1', 'P0', 'P01', 'P-1', 'P+1', 'P١', 'P1.0'): + assert not ProblemNumber.accepts(text) + with pytest.raises(ValueError): + store.create(text, source.github, 'Refuse a malformed identity', + creator=comms.registry.require('lead'), owner=comms.registry.require('worker'), + generation='malformed', goal=retained.goal, + review_at=retained.review_at, order=order) + with store.snapshot() as db: + assert ProblemAllocation.select(db) == [] + assert store.list(source.github) == (retained,) diff --git a/tools/cutover/carry_problem_numbers.py b/tools/cutover/carry_problem_numbers.py index 0249c35947..6a3361c41f 100644 --- a/tools/cutover/carry_problem_numbers.py +++ b/tools/cutover/carry_problem_numbers.py @@ -1,4 +1,4 @@ -"""Seed global problem allocation and index the original Todo assignment. +"""Index original Todo identities and install allocation for unsent numbers. Delete this step once every retained bus has been carried. """ @@ -9,7 +9,8 @@ from agent_comms.coordination_database import CoordinationStore from agent_comms.installer.carry import Carried, Carry from agent_comms.locked_store import SQLiteFile -from agent_comms.todos import ProblemNumber, Todo, TodoStore +from agent_comms.problem_number import ProblemNumber +from agent_comms.todos import ProblemAllocation, Todo, TodoStore class ProblemNumbers(Carry): @@ -28,18 +29,26 @@ def carry(cls, root: Path, *, apply: bool) -> Carried: return Carried(0) with closing(CoordinationStore.read_snapshot(path)) as db: stored = {row.name for row in TodoStore.stored_schema(db)} - numbers_missing = ProblemNumber.declared_name not in stored - index_missing = Todo.unit_index.name(Todo) not in stored - changed = int(numbers_missing) + int(index_missing) - if not changed or not apply: + numbers_missing = ProblemAllocation.declared_name not in stored + migrate_numbers = numbers_missing and 'problem_number' in stored + missing_indexes = tuple(index for index in (Todo.unit_index, Todo.problem_index) + if index.name(Todo) not in stored) + changed = int(numbers_missing) + len(missing_indexes) + if not changed: + return Carried(0) + created = tuple(ProblemNumber.parse(todo.id) for todo in Todo.stream(db)) + if not apply: return Carried(changed) with closing(CoordinationStore.writer(path)) as db, SQLiteFile.write_transaction(db): - if numbers_missing: - ProblemNumber.create(db) - for todo in Todo.stream(db): - ProblemNumber.reserve(db, todo.id) - if index_missing: - db.execute(Todo.unit_index.sql(Todo)) + if migrate_numbers: + # SQLite renames its sequence too; even unsent issued IDs stay reserved. + db.execute(f'ALTER TABLE problem_number RENAME TO "{ProblemAllocation.declared_name}"') + for number in created: + ProblemAllocation.delete(db, number=number.number) + elif numbers_missing: + ProblemAllocation.create(db) + for index in missing_indexes: + db.execute(index.sql(Todo)) TodoStore.require_schema(db) return Carried( changed, rewritten=tuple(str(file.relative_to(root)) for file in SQLiteFile.files(path)) From 18720db4a126eb603b67fee2793756445dd941cf Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Wed, 21 Oct 2026 16:41:56 -0400 Subject: [PATCH 2/3] Resolve proof actor from its declared bus root --- tools/proofs/cn_phase3_record.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tools/proofs/cn_phase3_record.py b/tools/proofs/cn_phase3_record.py index 5b65ff195c..a25894ff9a 100644 --- a/tools/proofs/cn_phase3_record.py +++ b/tools/proofs/cn_phase3_record.py @@ -105,7 +105,7 @@ def create_order(problem: str = "P1") -> None: text="Show the real verdict and strike records without a second source", request="record-request", commander="record-lead", - assigner=comms.registry.require("record-lead").incarnation, + assigner=Comms(ROOT).registry.require("record-lead").incarnation, review_at=RecordedEventTime(now + 3600), order=PendingProblemOrder( id=f"record-order-{problem}", From c73e22976acf1734c1173c6f2d74bb8b50f57ed1 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Wed, 21 Oct 2026 16:47:56 -0400 Subject: [PATCH 3/3] Finish fork call bindings and canonical problem validation --- src/agent_comms/commander/order.py | 16 ++++++------ src/agent_comms/goal_actions.py | 34 ++++++++++++++------------ src/agent_comms/goal_management.py | 1 + src/agent_comms/problem_number.py | 24 +++++++++--------- src/agent_comms/todos.py | 14 +++++++---- src/agent_comms/tool_caller.py | 5 ++-- tests/test_proof_build.py | 2 +- tools/cutover/carry_problem_numbers.py | 13 +++++++--- 8 files changed, 60 insertions(+), 49 deletions(-) diff --git a/src/agent_comms/commander/order.py b/src/agent_comms/commander/order.py index 35f4d11f5c..9f966d0fe1 100644 --- a/src/agent_comms/commander/order.py +++ b/src/agent_comms/commander/order.py @@ -52,10 +52,10 @@ 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 -from agent_comms.problem_number import ProblemNumber from agent_comms.standing_rules import HuntDue, OwedAnswer from agent_comms.storage_rules import UnitStorage from agent_comms.thread_identity import ThreadIncarnation @@ -235,7 +235,9 @@ 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() + actor: Annotated[ + InstanceOf[CommanderForkCaller] | InstanceOf[NoThreadCaller], Field(exclude=True) + ] = NoThreadCaller() _comms: Comms = PrivateAttr() def model_post_init(self, context: object) -> None: @@ -939,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') diff --git a/src/agent_comms/goal_actions.py b/src/agent_comms/goal_actions.py index 7d138b1eb4..8e6b83ed06 100644 --- a/src/agent_comms/goal_actions.py +++ b/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 @@ -509,7 +509,9 @@ class ProblemOrderGoalAction(GoalAction[Goal], RuntimeInvocable): commander: str assigner: ThreadIncarnation - def bound_to(self, actor: ThreadIncarnation, commander: ThreadIncarnation) -> ProblemOrderGoalAction: + def bound_to( + self, actor: ThreadIncarnation, commander: ThreadIncarnation, + ) -> ProblemOrderGoalAction: return replace(self, commander=commander.name, assigner=actor) @abstractmethod diff --git a/src/agent_comms/goal_management.py b/src/agent_comms/goal_management.py index 5bc7830380..11b3e9a7f8 100644 --- a/src/agent_comms/goal_management.py +++ b/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, diff --git a/src/agent_comms/problem_number.py b/src/agent_comms/problem_number.py index 821c1eaa5c..e674715faa 100644 --- a/src/agent_comms/problem_number.py +++ b/src/agent_comms/problem_number.py @@ -1,4 +1,6 @@ """The canonical spelling and numeric value of a global problem identity.""" + +import re from dataclasses import dataclass from typing import ClassVar, Self @@ -6,31 +8,27 @@ @dataclass(frozen=True, slots=True) class ProblemNumber: number: int - prefix: ClassVar[str] = 'P' + prefix: ClassVar[str] = "P" + spelling: ClassVar[re.Pattern[str]] = re.compile(rf"{prefix}[1-9][0-9]*") def __post_init__(self) -> None: if self.number < 1: - raise ValueError('A problem number is positive') + raise ValueError("A problem number is positive") def __str__(self) -> str: - return f'{self.prefix}{self.number}' + return f"{self.prefix}{self.number}" @classmethod def parse(cls, text: str) -> Self: - number = cls(int(text.removeprefix(cls.prefix))) - if str(number) != text: - raise ValueError(f'Not a canonical problem number: {text!r}') - return number + if not cls.accepts(text): + raise ValueError(f"Not a canonical problem number: {text!r}") + return cls(int(text[len(cls.prefix) :])) @classmethod def accepts(cls, text: str) -> bool: - try: - cls.parse(text) - except ValueError: - return False - return True + return bool(cls.spelling.fullmatch(text)) @classmethod def sql_number(cls, column: str) -> str: """The numeric projection of an already validated stored identity.""" - return f'CAST(substr({column}, {len(cls.prefix) + 1}) AS INTEGER)' + return f"CAST(substr({column}, {len(cls.prefix) + 1}) AS INTEGER)" diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 9a2d3afa8f..0155b74434 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -3083,7 +3083,8 @@ class Todo(TypedTable): @classmethod def rows(cls, db: sqlite3.Connection, where: str, parameters: tuple) -> tuple[Todo, ...]: return tuple(cls.read(db.execute( - f"SELECT * FROM {cls.declared_name} WHERE {cls._sql(where)} ORDER BY rowid", parameters))) + f"SELECT * FROM {cls.declared_name} WHERE {cls._sql(where)} ORDER BY rowid", + parameters))) @property def judged_calls(self) -> tuple[str, ...]: @@ -3392,7 +3393,8 @@ def require_unit_in(cls, db: sqlite3.Connection, unit: ThreadIncarnation, assigner: ThreadIncarnation) -> None: """The original open Todo owns who assigns its unit, under this transaction.""" for todo in cls.held_in(db, - f'NOT ({Assignment.sql_field("parent")}=? AND {Assignment.sql_field("parent_created")}=?)', + f'NOT ({Assignment.sql_field("parent")}=? ' + f'AND {Assignment.sql_field("parent_created")}=?)', (assigner.name, assigner.wire_birth), unit=unit): raise TodoError( f"{unit.name} is assigned by {todo.held().parent} on open order " @@ -3589,8 +3591,8 @@ def for_request(self, commander: ThreadIncarnation, goal_id: str) -> tuple[Todo, "AND json_extract(goal,'$.goal_id')=?", (commander.name, commander.wire_birth, goal_id)) def for_commander(self, commander: ThreadIncarnation) -> tuple[Todo, ...]: - return self._select("creator=? AND creator_created=?", (commander.name, commander.wire_birth), - reader=self._open) + return self._select("creator=? AND creator_created=?", + (commander.name, commander.wire_birth), reader=self._open) def timers_up_for_commander(self, commander: ThreadIncarnation, registry: Registration, now: float) -> tuple[tuple[Todo, RecordedEventTime], ...]: @@ -3686,7 +3688,9 @@ def correct(self, unit: ThreadIncarnation, message_id: str, text: str) -> tuple[ todo.upsert(db) return corrected - def fire(self, unit: ThreadIncarnation, actor: ThreadIncarnation, reason: StrikeReason) -> tuple[Todo, ...]: + def fire( + self, unit: ThreadIncarnation, actor: ThreadIncarnation, reason: StrikeReason, + ) -> tuple[Todo, ...]: """Release every open problem the unit holds from this commander in one write; each records the same reason, and its order and strikes stay.""" with self.writing() as db: diff --git a/src/agent_comms/tool_caller.py b/src/agent_comms/tool_caller.py index c179b88bbd..6963c291a9 100644 --- a/src/agent_comms/tool_caller.py +++ b/src/agent_comms/tool_caller.py @@ -10,15 +10,15 @@ from .field_codec import WireForm from .launch_root import LaunchRoot from .launch_runtime import CallingThreadUnavailable -from .thread_identity import RecordedThreadIncarnation, ThreadIncarnation from .presentation import MessageNotification, ThreadRecord +from .thread_identity import RecordedThreadIncarnation, ThreadIncarnation from .tool_result import ToolResult if TYPE_CHECKING: from .comms import Comms from .messages import Message - from .tools import ToolRequest from .threads import Thread + from .tools import ToolRequest @dataclass(frozen=True) @@ -58,6 +58,7 @@ def subject(self, named: str) -> str: def commanding(self, comms: Comms, commander: str) -> None: """An ordinary caller needs no commander-child registration.""" + return None def retire(self, comms: Comms) -> None: raise ToolRefusedError('Only a commander fork can retire through this verb.') diff --git a/tests/test_proof_build.py b/tests/test_proof_build.py index 62716383a0..fe660a9108 100644 --- a/tests/test_proof_build.py +++ b/tests/test_proof_build.py @@ -297,7 +297,7 @@ def test_installed_proof_closes_and_older_build_is_refused_on_copied_bus( order=order, request="request", review_at=RecordedEventTime(4102444800), - commander="lead", assigner=caller.identity(comms), + commander="lead", assigner=caller.identity(copied), repo=graph.repository.github, text="Bind proof to its build", ), diff --git a/tools/cutover/carry_problem_numbers.py b/tools/cutover/carry_problem_numbers.py index 6a3361c41f..c98808ba5b 100644 --- a/tools/cutover/carry_problem_numbers.py +++ b/tools/cutover/carry_problem_numbers.py @@ -30,9 +30,12 @@ def carry(cls, root: Path, *, apply: bool) -> Carried: with closing(CoordinationStore.read_snapshot(path)) as db: stored = {row.name for row in TodoStore.stored_schema(db)} numbers_missing = ProblemAllocation.declared_name not in stored - migrate_numbers = numbers_missing and 'problem_number' in stored - missing_indexes = tuple(index for index in (Todo.unit_index, Todo.problem_index) - if index.name(Todo) not in stored) + migrate_numbers = numbers_missing and "problem_number" in stored + missing_indexes = tuple( + index + for index in (Todo.unit_index, Todo.problem_index) + if index.name(Todo) not in stored + ) changed = int(numbers_missing) + len(missing_indexes) if not changed: return Carried(0) @@ -42,7 +45,9 @@ def carry(cls, root: Path, *, apply: bool) -> Carried: with closing(CoordinationStore.writer(path)) as db, SQLiteFile.write_transaction(db): if migrate_numbers: # SQLite renames its sequence too; even unsent issued IDs stay reserved. - db.execute(f'ALTER TABLE problem_number RENAME TO "{ProblemAllocation.declared_name}"') + db.execute( + f'ALTER TABLE problem_number RENAME TO "{ProblemAllocation.declared_name}"' + ) for number in created: ProblemAllocation.delete(db, number=number.number) elif numbers_missing: