diff --git a/src/agent_comms/commander/capacity.py b/src/agent_comms/commander/capacity.py index ff7b00418d..cf738a886a 100644 --- a/src/agent_comms/commander/capacity.py +++ b/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) diff --git a/src/agent_comms/commander/judge.py b/src/agent_comms/commander/judge.py index 2b64c969e6..1b1a63c9c2 100644 --- a/src/agent_comms/commander/judge.py +++ b/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) diff --git a/src/agent_comms/commander/order.py b/src/agent_comms/commander/order.py index d0e71da667..ebc9a69987 100644 --- a/src/agent_comms/commander/order.py +++ b/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 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("|", "|")} | {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) diff --git a/src/agent_comms/commander_parity/deletion.py b/src/agent_comms/commander_parity/deletion.py index f6ab9574b8..ce2a1f6073 100644 --- a/src/agent_comms/commander_parity/deletion.py +++ b/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, ), ) ) diff --git a/src/agent_comms/commander_parity/story.py b/src/agent_comms/commander_parity/story.py index 2162c0e82a..b20249ad50 100644 --- a/src/agent_comms/commander_parity/story.py +++ b/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.", ), diff --git a/src/agent_comms/goal_actions.py b/src/agent_comms/goal_actions.py index 69670fce0a..c07eea44f2 100644 --- a/src/agent_comms/goal_actions.py +++ b/src/agent_comms/goal_actions.py @@ -499,9 +499,12 @@ class ProblemOrderGoalAction(GoalAction[Goal], RuntimeInvocable): request: str review_at: EventTime commander: str + assigner: ThreadIncarnation - @abstractmethod - def bound_to(self, actor: ThreadIncarnation) -> ProblemOrderGoalAction: ... + def bound_to( + self, actor: ThreadIncarnation, commander: ThreadIncarnation, + ) -> ProblemOrderGoalAction: + return replace(self, commander=commander.name, assigner=actor) @abstractmethod def reserve(self, ctx: GoalActionContext, generation: str, goal: GoalRef) -> Todo: ... @@ -548,11 +551,6 @@ class OrderGoalAction(ProblemOrderGoalAction): traced: Annotated[FileLine | NoSourceReference, field_codec.WireOmitDefault()] = ( NoSourceReference() ) - assigner: ThreadIncarnation - - def bound_to(self, actor: ThreadIncarnation) -> OrderGoalAction: - return replace(self, commander=actor.name, assigner=actor) - def reserve(self, ctx: GoalActionContext, generation: str, goal: GoalRef) -> Todo: return ctx.goals.todos.assign(self.problem, expected_revision=self.expected_revision, owner=ctx.thread, assigner=self.assigner, @@ -577,13 +575,11 @@ class CreateProblemGoalAction(ProblemOrderGoalAction): repo: RepositoryName text: str - def bound_to(self, actor: ThreadIncarnation) -> CreateProblemGoalAction: - return replace(self, commander=actor.name) - def reserve(self, ctx: GoalActionContext, generation: str, goal: GoalRef) -> Todo: return ctx.goals.todos.create(self.problem, self.repo, self.text, creator=ctx.goals.registry.require(self.commander), owner=ctx.thread, - generation=generation, goal=goal, review_at=self.review_at, order=self.order) + assigner=self.assigner, generation=generation, goal=goal, + review_at=self.review_at, order=self.order) @dataclass(frozen=True, kw_only=True) diff --git a/src/agent_comms/goal_management.py b/src/agent_comms/goal_management.py index 905b81d808..347dafdbde 100644 --- a/src/agent_comms/goal_management.py +++ b/src/agent_comms/goal_management.py @@ -41,6 +41,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, @@ -98,11 +99,11 @@ def send_back(self, problem: str, verdict: SendBackVerdict) -> Todo: sent_back: Todo = self.todos.get(problem) return sent_back - def fire(self, unit: str, reason: StrikeReason, commander: str) -> tuple[Todo, ...]: - actor = self.registry.require(commander).incarnation + def fire(self, unit: str, reason: StrikeReason, actor: ThreadIncarnation) -> tuple[Todo, ...]: + owner = self.registry.require(unit).incarnation store: TodoStore = self.todos with _store_lock(self._wire_lock_path): - return store.fire(unit, actor, reason) + return store.fire(owner, actor, reason) def open_goal_store(self) -> GoalAttemptStore: if self.goal_store is None: diff --git a/src/agent_comms/messaging.py b/src/agent_comms/messaging.py index b881199cfd..b4f9d443f2 100644 --- a/src/agent_comms/messaging.py +++ b/src/agent_comms/messaging.py @@ -78,7 +78,7 @@ def send( publisher: OwnerPublisher = self.bus.publisher committed = publisher.publish_ordinary(message, origin=AgentOrigin()) try: - todos.correct(owner.name, committed.message_id, body) + todos.correct(owner.incarnation, committed.message_id, body) except TodoError as error: correction = CorrectionFailed(f'{error.__class__.__name__}: {error}') else: diff --git a/src/agent_comms/owner_lifecycle.py b/src/agent_comms/owner_lifecycle.py index 9704cbbfa8..8a793208d0 100644 --- a/src/agent_comms/owner_lifecycle.py +++ b/src/agent_comms/owner_lifecycle.py @@ -310,7 +310,7 @@ def _register_owner_unlocked(self, previous: Thread, identity: ProcessIdentity) orders = tuple( todo for commander in CommanderHeartbeat.commanders(self.registry, self.goals) - for todo in self.goals.todos.held_by(commander.incarnation, owned.name) + for todo in self.goals.todos.held_by(commander.incarnation, owned.incarnation) if todo.pending and todo.held().owner_identity == owned.incarnation ) if orders: diff --git a/src/agent_comms/problem_number.py b/src/agent_comms/problem_number.py new file mode 100644 index 0000000000..e674715faa --- /dev/null +++ b/src/agent_comms/problem_number.py @@ -0,0 +1,34 @@ +"""The canonical spelling and numeric value of a global problem identity.""" + +import re +from dataclasses import dataclass +from typing import ClassVar, Self + + +@dataclass(frozen=True, slots=True) +class ProblemNumber: + number: int + 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") + + def __str__(self) -> str: + return f"{self.prefix}{self.number}" + + @classmethod + def parse(cls, text: str) -> Self: + 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: + 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)" diff --git a/src/agent_comms/storage_rules.py b/src/agent_comms/storage_rules.py index 0995f5588d..a7d93c52a9 100644 --- a/src/agent_comms/storage_rules.py +++ b/src/agent_comms/storage_rules.py @@ -4,7 +4,6 @@ import getopt import json import math -import re import shutil import subprocess import time @@ -16,6 +15,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 @@ -146,7 +146,7 @@ def owns_checkout(self, owner: Thread, path: Path, source: 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/thread_identity.py b/src/agent_comms/thread_identity.py index 348937080a..ac8be15821 100644 --- a/src/agent_comms/thread_identity.py +++ b/src/agent_comms/thread_identity.py @@ -103,6 +103,10 @@ def recorded_or( ) -> T: return otherwise() + @property + def commanding_identity(self) -> ThreadIncarnation: + return self + def require_may_change(self, todo: Todo) -> None: todo.require_commander(self) @@ -126,7 +130,7 @@ def recorded_names(self, snapshot: RegistryProvenance) -> tuple[ThreadIncarnatio if name == canonical.name)) -@dataclass(frozen=True) +@dataclass(frozen=True, eq=False) class RecordedThreadIncarnation(ThreadIncarnation, ParticipantIdentity): """The recorded creation owns identity; names remain registry-resolved routes. @@ -137,6 +141,14 @@ class RecordedThreadIncarnation(ThreadIncarnation, ParticipantIdentity): name: str = field(compare=False) created_at: float + def __eq__(self, other: object) -> bool: + if not isinstance(other, ThreadIncarnation): + return NotImplemented + return self.wire_birth == other.wire_birth + + def __hash__(self) -> int: + return hash(self.wire_birth) + @property def wire_birth(self) -> float: return self.created_at diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index cd4020d4f4..4beb40a0f6 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 @@ -80,6 +80,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, @@ -2173,6 +2174,9 @@ def require_reference(self) -> None: def threads(self) -> tuple[str, ...]: return () + def authorize_order(self, current: Todo, actor: ThreadIncarnation) -> None: + actor.require_may_change(current) + @dataclass(frozen=True) class NoGoalReference(NullMember, GoalReference): @@ -2208,6 +2212,9 @@ def threads(self) -> tuple[str, ...]: def request(self) -> str: return self.goal_id + def authorize_order(self, current: Todo, actor: ThreadIncarnation) -> None: + current.require_commander(self.incarnation) + class TodoHolder(ABC): """Current, released and absent holders own their assignment behaviour.""" @@ -2257,6 +2264,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) @@ -2272,7 +2284,7 @@ def owner_identity(self) -> ThreadIncarnation: @property def parent_identity(self) -> ThreadIncarnation: - """Who made the assignment: the order's commander, or Core's watcher.""" + """The acting assigner: commander, registered fork, or watcher; not Todo.creator.""" return ThreadIncarnation.from_wire({"name": self.parent, "created_at": self.parent_created}) def held(self, current: Todo) -> Assignment: @@ -2409,6 +2421,9 @@ def apply(self, current: Todo) -> Todo: ... class TodoOperation(TodoCommand, declared_name=None): """Commands the ordinary commander's tools accept.""" + def authored_by(self, actor: ThreadIncarnation) -> Self: + return replace(self, actor=actor) + @dataclass(frozen=True) class AssignTodo(TodoOperation): @@ -2420,8 +2435,12 @@ class AssignTodo(TodoOperation): goal: GoalReference traced: Annotated[FileLine | NoSourceReference, WireOmitDefault()] = NoSourceReference() + def authored_by(self, actor: ThreadIncarnation) -> Self: + return replace(self, proposed=replace(self.proposed, + parent=actor.name, parent_created=actor.wire_birth)) + def require_unit(self, db: sqlite3.Connection, current: Todo) -> None: - TodoStore.require_unit_in(db, self.proposed.owner_identity, current.creator_identity) + TodoStore.require_unit_in(db, self.proposed.owner_identity, self.actor) def apply(self, current: Todo) -> Todo: self.review_at.require_recorded("A review time") @@ -2430,7 +2449,7 @@ def apply(self, current: Todo) -> Todo: proposed, self.review_at, self.order, self.goal): return current current.require_revision(self.expected_revision) - self.actor.require_may_change(current) + self.goal.authorize_order(current, self.actor) if self.order != current.order: self.order.require_wording() self.order.require_writable() @@ -2858,6 +2877,9 @@ class WordingCorrection(TodoOperation): message_id: str words: tuple[str, ...] + def authored_by(self, actor: ThreadIncarnation) -> Self: + return self + def apply(self, current: Todo) -> Todo: current.require_revision(self.expected_revision) corrections = (*current.corrections, self) @@ -3024,10 +3046,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)] @@ -3049,6 +3072,12 @@ 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))) + @property def judged_calls(self) -> tuple[str, ...]: return tuple(call for verdict in self.verdicts for call in verdict.judged_calls()) @@ -3237,11 +3266,11 @@ def require_fold(self, target: Todo, why: str) -> None: self.order.folded(target.id, why) def require_commander(self, actor: ThreadIncarnation) -> None: - if self.creator_identity != actor: + if self.creator_identity != actor.commanding_identity: raise TodoConflict(self) def require_actor(self, actor: ThreadIncarnation, generation: str | None) -> None: - if self.creator_identity != actor: + if self.creator_identity != actor.commanding_identity: self.require_assignee(actor, generation) def require_same_creation(self, proposed: Todo) -> None: @@ -3283,20 +3312,36 @@ 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, text: str) -> ProblemNumber: + try: + problem = ProblemNumber.parse(text) + except ValueError as error: + raise TodoError(f'Malformed problem ID: {text!r}') from error + return cls.issue(db, problem) + + @classmethod + def issue(cls, db: sqlite3.Connection, problem: ProblemNumber) -> ProblemNumber: + if not Todo.select(db, id=str(problem), limit=1): + cls(problem.number).insert_if_absent(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.issue(db, ProblemNumber(number)) @dataclass(frozen=True, slots=True) @@ -3310,7 +3355,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]: @@ -3334,24 +3379,20 @@ 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 cls.held_in(db, + 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 " + 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) @@ -3365,10 +3406,11 @@ def hunts_for_commander(self, commander: ThreadIncarnation) -> tuple[Hunt, ...]: return tuple(record.hunt for record in HuntRecord.select(db, commander=commander)) def resolve_order(self, commander: ThreadIncarnation, unit: str, problem: str) -> Todo: - found = [t for t in self.held_by(commander, unit) if t.id == problem] - if len(found) != 1: + found = self.get(problem) + found.require_commander(commander) + if not found.state.open or found.holder != unit: raise ValueError(f'{unit} holds no order on {problem}') - return found[0] + return found def drop(self, commander: ThreadIncarnation, unit: str, problem: str, by: str) -> Todo: old = self.resolve_order(commander, unit, problem) @@ -3425,17 +3467,30 @@ 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.""" + @staticmethod + def _open(db: sqlite3.Connection, where: str, parameters: tuple, + state: type[TodoState] = TodoState) -> tuple[Todo, ...]: + states = state.open_names() + selected = ','.join('?' for state in states) + return Todo.rows(db, f"state IN ({selected}) AND ({where})", (*states, *parameters)) + + @classmethod + def held_in(cls, db: sqlite3.Connection, where: str, parameters: tuple, + *, unit: ThreadIncarnation) -> tuple[Todo, ...]: + """The one open-unit query, used by every reader and writer.""" + terms = Todo.unit_index.terms[:-1] + match = f"({', '.join(terms)})=({','.join('?' for term in terms)})" + return cls._open(db, f'{match} AND ({where})', + (unit.name, unit.wire_birth, *parameters)) + + 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: @@ -3445,7 +3500,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.""" @@ -3473,17 +3528,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(self.held_in, 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(self._open, 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(self._open, state=OpenTodoState)) if todo.pending and todo.handed_out_on_release) def merged_orders(self, commander: ThreadIncarnation, @@ -3518,9 +3574,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(self.held_in, 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.""" @@ -3529,7 +3584,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=self._open) def timers_up_for_commander(self, commander: ThreadIncarnation, registry: Registration, now: float) -> tuple[tuple[Todo, RecordedEventTime], ...]: @@ -3541,13 +3597,12 @@ 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=self._open) - 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)) + def held_by(self, commander: ThreadIncarnation, unit: ThreadIncarnation) -> tuple[Todo, ...]: + return self._select('creator=? AND creator_created=?', + (commander.name, commander.wire_birth), reader=partial(self.held_in, unit=unit)) def require_recurrence_declaration( self, order: CheckedProblemOrder, where: FileLine, root: Path, @@ -3573,12 +3628,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, @@ -3587,25 +3636,27 @@ def create( *, creator: Thread, owner: Thread, + assigner: ThreadIncarnation, generation: str, review_at: EventTime, order: PendingProblemOrder, goal: GoalRef, ) -> Todo: + number = ProblemNumber.parse(todo_id) identity, assignee = _participant(creator), _thread(owner) DirectiveText.require(text) order.require_wording() order.require_writable() proposed = Todo.ordered( - todo_id, repo, text, identity, Assignment.of(assignee, identity, generation), + todo_id, repo, text, identity, Assignment.of(assignee, assigner, generation), review_at, order, goal) with self.writing() as db: - self.require_unit_in(db, assignee, identity) - ProblemNumber.reserve(db, todo_id) + self.require_unit_in(db, assignee, assigner) 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: @@ -3618,27 +3669,28 @@ def apply(self, todo_id: str, command: TodoCommand) -> Todo: result.upsert(db) return self._current(db, todo_id) - def correct(self, unit: str, message_id: str, text: str) -> tuple[Todo, ...]: + def correct(self, unit: ThreadIncarnation, message_id: str, text: str) -> tuple[Todo, ...]: """Correct every open order the unit holds when its published text uses jargon.""" 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 = self.held_in(db, '1', (), unit=unit) corrected = tuple(WordingCorrection(todo.revision, message_id, words).apply(todo) for todo in held) for todo in corrected: todo.upsert(db) return corrected - def fire(self, unit: str, 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: - 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) + commander = actor.commanding_identity + held = tuple((todo, todo.held()) for todo in self.held_in( + db, 'creator=? AND creator_created=?', + (commander.name, commander.wire_birth), unit=unit)) 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..6963c291a9 100644 --- a/src/agent_comms/tool_caller.py +++ b/src/agent_comms/tool_caller.py @@ -6,16 +6,18 @@ 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 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 .threads import Thread from .tools import ToolRequest @@ -54,8 +56,15 @@ def own_thread(self) -> str: def subject(self, named: str) -> str: return named or self.own_thread() - def commanding(self, comms: Comms, commander: str) -> ThreadCaller: - return ThreadCaller(self.acting_name(commander)) + 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.') + + @abstractmethod + def as_thread(self, commander: str) -> ThreadCaller: ... @abstractmethod def may_reconfigure(self, comms: Comms, name: str) -> bool: ... @@ -71,6 +80,9 @@ def sent(self, comms: Comms, message: Message) -> None: @dataclass(frozen=True) class NoThreadCaller(ToolCaller): + def as_thread(self, commander: str) -> ThreadCaller: + return ThreadCaller(commander) + def invoke(self, request: ToolRequest, comms: Comms) -> ToolResult: return request.apply_without_thread(comms, self) @@ -106,6 +118,15 @@ def sent(self, comms: Comms, message: Message) -> None: class ThreadCaller(ToolCaller): name: str + def as_thread(self, commander: str) -> ThreadCaller: + return self + + def commander(self, comms: Comms) -> Thread: + return comms.registry.require(self.name) + + def identity(self, comms: Comms) -> ThreadIncarnation: + return comms.registry.require(self.name).incarnation + def invoke(self, request: ToolRequest[ThreadCaller], comms: Comms) -> ToolResult: return request.apply(comms, self) @@ -139,17 +160,36 @@ 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, eq=False) +class CommanderForkIdentity(RecordedThreadIncarnation): + """A registered actor's parent authority, resolved before any Todo transaction. + + The ordinary incarnation wire form records the actor, not a reusable authority. + """ + + authority: ThreadIncarnation + + @property + def commanding_identity(self) -> ThreadIncarnation: + return self.authority @dataclass(frozen=True) class CommanderForkCaller(ThreadCaller): """Commander verbs use a registered child without copying registry procedures.""" - def commanding(self, comms: Comms, commander: str) -> ThreadCaller: + def commanding(self, comms: Comms, commander: str) -> None: comms.registry.register_commander_child(self.name, commander, comms.goals) - return self + + def commander(self, comms: Comms) -> Thread: + fork = comms.registry.require(self.name) + parent, = fork.parent.names() + return comms.registry.require(parent) + + def identity(self, comms: Comms) -> ThreadIncarnation: + fork = comms.registry.require(self.name) + return CommanderForkIdentity(fork.name, fork.created_at, self.commander(comms).incarnation) def retire(self, comms: Comms) -> None: comms.registry.unregister(self.name) diff --git a/src/agent_comms/tools.py b/src/agent_comms/tools.py index d7d9d2c422..4b9f39cdf0 100644 --- a/src/agent_comms/tools.py +++ b/src/agent_comms/tools.py @@ -884,9 +884,9 @@ class CommsTodoTool(ThreadToolRequest): def apply(self, comms: Comms, caller: ThreadCaller) -> TodoChanged | ProblemClosed: todo = comms.goals.todos.get(self.problem) - actor = comms.registry.require(caller.name).incarnation + actor = caller.identity(comms) return self.result(comms.goals.change_problem( - self.problem, self.operation_for(todo, actor, comms), caller.name)) + self.problem, self.operation_for(todo, actor, comms), caller.commander(comms).name)) @abstractmethod def operation_for(self, todo: Todo, actor: ThreadIncarnation, comms: Comms) -> TodoOperation: @@ -903,7 +903,7 @@ class CommsChangeProblemTool(CommsTodoTool, declared_name="comms_todo"): ] def operation_for(self, todo: Todo, actor: ThreadIncarnation, comms: Comms) -> TodoOperation: - return self.operation + return self.operation.authored_by(actor) @dataclass(frozen=True, kw_only=True) @@ -949,7 +949,7 @@ class CommsFireTool(ThreadToolRequest): reason: Annotated[StrikeReason, *tool_field("What the unit did, plainly")] def apply(self, comms: Comms, caller: ThreadCaller) -> TodoList: - return TodoList(comms.goals.fire(self.unit, self.reason, caller.name)) + return TodoList(comms.goals.fire(self.unit, self.reason, caller.identity(comms))) @dataclass(frozen=True, kw_only=True) @@ -1028,7 +1028,7 @@ class CommsOrderTool(ThreadToolRequest): def apply(self, comms: Comms, caller: ThreadCaller) -> ProblemOrdered: goal = comms.goals.update_goal(self.owner, self.order.bound_to( - comms.registry.require(caller.name).incarnation)) + caller.identity(comms), caller.commander(comms).incarnation)) return ProblemOrdered(goal, comms.goals.todos.get(self.order.problem)) diff --git a/src/agent_comms/work_dispatch.py b/src/agent_comms/work_dispatch.py index 7186a45b80..6980497f8d 100644 --- a/src/agent_comms/work_dispatch.py +++ b/src/agent_comms/work_dispatch.py @@ -149,6 +149,7 @@ def start( order=order, request=order.request(commander, (self.problem,)), commander=commander.name, + assigner=commander.incarnation, review_at=RecordedEventTime(now + self.review_seconds), ) WorkDispatch(action, revision=0).start(comms) diff --git a/tests/test_bus_files.py b/tests/test_bus_files.py index c7c8bf7674..0ee6967f9d 100644 --- a/tests/test_bus_files.py +++ b/tests/test_bus_files.py @@ -43,12 +43,12 @@ def test_every_bus_file_is_a_declared_root_file(tmp_path: Path, source_repositor ) assert claim.seq > sent.seq todo = TodoStore(comms.root).create( - "declared-files", RepositoryName.from_wire("OpenHCSDev/agent-comms"), - "Keep declared bus files", creator=owner, owner=owner, + "P1", RepositoryName.from_wire("OpenHCSDev/agent-comms"), + "Keep declared bus files", creator=owner, assigner=owner.incarnation, owner=owner, generation="declared-files", goal=GoalRef(owner.name, owner.created_at, "goal-1"), review_at=RecordedEventTime(1000.0), order=order(source_repository), ) - assert todo.id == "declared-files" + assert todo.id == "P1" drain_candidate_catchup(comms.root) declared = RootFile.paths_in(comms.root) assert PrivateOwnerRenameIntent.path_in(comms.root) in declared diff --git a/tests/test_commander_forks.py b/tests/test_commander_forks.py index f6c72311e7..fca75763a9 100644 --- a/tests/test_commander_forks.py +++ b/tests/test_commander_forks.py @@ -16,19 +16,26 @@ from source_revision_fixture import SourceRepository from agent_comms.change_report import FileLine from agent_comms.commander.cli import main -from agent_comms.commander.order import Commander, OrderFile +from agent_comms.commander.order import Commander, CommanderContext, OrderFile from agent_comms.comms import Comms from agent_comms.field_codec import FieldCodec +from agent_comms.external_transcript import ExternalTranscript +from agent_comms.tool_caller import CommanderForkCaller +from agent_comms.tool_declaration import ToolDeclaration +from agent_comms.transcript_events import ToolStartTranscript from agent_comms.goal_actions import SetGoalAction from agent_comms.installer.backup import StoreBackup from agent_comms.goals import CountCheckResult, OrderCheck, RequestGoalOrder 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, ) @@ -61,7 +68,7 @@ def fork_bus(tmp_path): at=RecordedEventTime(now), first_call_deadline=RecordedEventTime(now + 600)) lead = comms.registry.require('lead') comms.goals.todos.create('P40', source.github, 'Existing parent problem', creator=lead, - owner=comms.registry.require('retained'), generation='retained', + owner=comms.registry.require('retained'), assigner=lead.incarnation, generation='retained', review_at=RecordedEventTime(now + 600), order=replace(order, id='retained-order'), goal=GoalRef('lead', lead.created_at, 'request')) copied = tmp_path / 'copied-bus' @@ -96,9 +103,9 @@ def process_command(comms, *args): capture_output=True, text=True, timeout=90) -def document(path, order): +def document(path, order, problems=()): path.write_text(json.dumps(FieldCodec.encode(OrderFile( - order=order, + order=order, problems=problems, defect='Two commander forks can assign conflicting work to the same unit.', why_stupid='This is stupid because the original assignment is not checked.', unplanned_reason='This failure blocks the commander from assigning independent work safely.', @@ -107,76 +114,94 @@ def document(path, order): return str(path) -@pytest.mark.timeout(180) -def test_fork_orders_refuse_from_todos_and_retirement_is_explicit(fork_bus, tmp_path): +def test_registered_fork_sends_parent_problem_with_fork_actor(fork_bus, tmp_path): comms, source, order = fork_bus - first = document(tmp_path / 'first.json', order) - result = command(comms, 'order', 'problem', first, '--to', 'worker', - '--where', 'owner.py:2', '--fork', 'fork-one') - assert result.returncode == 0, result.stderr + store = comms.goals.todos + parent = comms.registry.require('lead') + original = store.get('P40') + store.apply(original.id, ReleaseTodo(original.revision, parent.incarnation, original.held(), + StrikeReason('The original worker stopped and the parent has explicitly released its assignment.'))) + body = document(tmp_path / 'parent-problem.json', order, ('P40',)) + sent = command(comms, 'order', 'send', 'worker', body, '--fork', 'fork-one') + assert sent.returncode == 0, sent.stderr fork = comms.registry.require('fork-one') - assert fork.parent.named('lead') - assert fork.goal.require().id != comms.registry.require('lead').goal.require().id - held = comms.goals.todos.held_for_commander(fork.incarnation) - assert [todo.id for todo in held] == ['P41'] - original = held[0] - 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'), ()): - result = command(comms, 'order', 'problem', second, '--to', 'worker', - '--where', 'owner.py:2', *prefix) - assert result.returncode != 0 and refusal in result.stderr - print(result.stderr.strip()) - result = command(comms, 'order', 'problem', second, '--to', 'other', - '--where', 'owner.py:2', '--fork', 'fork-two') - assert result.returncode == 0, result.stderr - result = command(comms, 'capacity') - assert result.returncode == 0 and "2 live forks ['fork-one', 'fork-two']" in result.stdout - print(result.stdout.strip()) - second_owned = comms.goals.todos.held_for_commander( - comms.registry.require('fork-two').incarnation)[0] - # Reassign this fork's own problem, not the other commander's P number. - resend = replace(OrderFile.read(Path(second).read_text()), problems=(second_owned.id,)) - resend_file = tmp_path / 'reassign.json' - resend_file.write_text(json.dumps(FieldCodec.encode(resend))) - result = command(comms, 'order', 'send', 'worker', str(resend_file), - '--fork', 'fork-two', '--reassign', - 'The copied fixture exercises transfer after the first worker failed its assigned review.') - assert result.returncode != 0 and refusal in result.stderr - assert comms.goals.todos.get(second_owned.id) == second_owned - lead = comms.registry.require('lead') - worker = comms.registry.require('worker') - retained = comms.goals.todos.get('P40') - with pytest.raises(TodoError, match='fork-one on open order first-order'): - comms.goals.todos.create('P999', source.github, 'Conflicting native order', - creator=lead, owner=worker, generation='competing', goal=retained.goal, - review_at=retained.review_at, order=order) - with pytest.raises(TodoError, match='fork-one on open order first-order'): - comms.goals.todos.assign('P40', expected_revision=retained.revision, - owner=worker, assigner=lead.incarnation, generation='competing', - goal=retained.goal, review_at=retained.review_at, order=order) + saved = store.get('P40') + assert saved.creator_identity == parent.incarnation + assert saved.goal.owner == parent.name + assert saved.held().parent_identity == fork.incarnation + assert saved.messages[-1].actor == fork.incarnation + assert fork.parent.named(parent.name) + assert fork.goal.require().id != parent.goal.require().id + print(f'Core send actor: {saved.held().parent}; problem commander: {saved.creator}') + ctx = Commander.read_root(comms.root).model_copy(update={'actor': CommanderForkCaller(fork.name)}) + assert set(ctx.model_copy(deep=True).progress()) == {'P40'} + assert ctx.held_orders() == (saved,) + assert ctx.call(CommsTodoListTool(thread=ctx.commander)).todos == (saved,) + assert "1 live forks ['fork-one']" in command(comms, 'capacity').stdout + refused = command(comms, 'order', 'send', 'worker', body, '--fork', 'fork-two') + assert refused.returncode != 0 and 'worker is assigned by fork-one' in refused.stderr + assert store.get('P40') == saved interim = tmp_path / 'interim.txt' interim.write_text('The first review is complete; another check remains.') - result = command(comms, 'order', 'say', '#comms', str(interim), '--fork', 'fork-one') - assert result.returncode == 0, result.stderr + posted = command(comms, 'order', 'say', '#comms', str(interim), '--fork', 'fork-one') + assert posted.returncode == 0, posted.stderr assert comms.registry.status('fork-one').receiving - assert comms.goals.todos.get('P41') == original - result = command(comms, 'order', 'retire', '--fork', 'fork-one') - assert result.returncode == 0, result.stderr + assert store.get('P40') == saved + acting = CommanderContext.read_root(comms.root).model_copy( + update={'actor': CommanderForkCaller(fork.name)}) + acting.strike('worker', 'P40', + 'The fork found the requested source change missing during its copied review.', + evidence=(order.owner,)) + saved = store.get('P40') + assert saved.strikes[-1].actor == fork.incarnation + assert 'caught by fork-one at review' in str(saved.strikes[-1].reason) + refused = command(comms, 'order', 'retire') + assert refused.returncode != 0 and 'Only a commander fork can retire' in refused.stderr + retired = command(comms, 'order', 'retire', '--fork', 'fork-one') + assert retired.returncode == 0, retired.stderr assert not comms.registry.status('fork-one').receiving - assert comms.goals.todos.get('P41') == original - assert "1 live forks ['fork-two']" in command(comms, 'capacity').stdout - # Stopping a thread cannot discard its open work. Only the original Todo changes it. - result = command(comms, 'order', 'problem', second, '--to', 'worker', - '--where', 'owner.py:2') - assert result.returncode != 0 and refusal in result.stderr - comms.goals.todos.apply('P41', ReleaseTodo( - original.revision, fork.incarnation, original.held(), - StrikeReason('The fixture releases its actual original order after the completed review.'))) - result = command(comms, 'order', 'problem', second, '--to', 'worker', - '--where', 'owner.py:2') - assert result.returncode == 0, result.stderr - assert comms.goals.todos.get('P40') == retained + assert store.get('P40') == saved + + +def test_fork_and_parent_judge_show_same_unjudged_edits(fork_bus): + comms, source, order = fork_bus + actor = comms.registry.require('retained').incarnation + declaration = ToolDeclaration.record('edit', { + 'path': str(source.checkout / 'owner.py'), + 'edits': [{'oldText': '', 'newText': 'class ReviewedOwner: pass'}]}, call_id='parent-edit') + ExternalTranscript(comms.root, actor).finish(turn_id='parent-work', item_id='parent-edit', events=( + ToolStartTranscript(position=0, tool_call_id='parent-edit', tool_name='edit', + timestamp=time.time(), tool=declaration),)) + own = command(comms, 'judge', 'show') + fork = command(comms, 'judge', 'show', '--fork', 'review-fork') + for result in (own, fork): + assert result.returncode == 0, result.stderr + assert result.stdout.splitlines()[-1] == '1 unjudged' + assert 'parent-edit' in result.stdout + print('judge show: 1 unjudged; judge show --fork review-fork: 1 unjudged') + + +def test_open_unit_consumers_agree_on_one_assignment(fork_bus): + comms, source, order = fork_bus + store = comms.goals.todos + todo = store.get('P40') + owner = comms.registry.require('retained').incarnation + assert store.holds_pending(owner) + assert store.for_goal(owner, todo.held().generation) == (todo,) + assert store.held_by(comms.registry.require('lead').incarnation, owner) == (todo,) + other_birth = RecordedThreadIncarnation(owner.name, owner.wire_birth + 1) + assert not store.holds_pending(other_birth) + assert store.for_goal(other_birth, todo.held().generation) == () + with store.snapshot() as db: + store.require_unit_in(db, owner, todo.held().parent_identity) + store.require_unit_in(db, other_birth, comms.registry.require('other').incarnation) + queries = [] + db.set_trace_callback(queries.append) + with pytest.raises(TodoError, match='retained is assigned by lead'): + store.require_unit_in(db, owner, comms.registry.require('other').incarnation) + db.set_trace_callback(None) + plan = db.execute('EXPLAIN QUERY PLAN ' + queries[0]).fetchall() + print('Shared open-unit query plan:', [tuple(row) for row in plan]) def test_two_commanders_allocate_distinct_global_problem_ids(fork_bus, tmp_path): @@ -189,11 +214,12 @@ def assign(pair): with ThreadPoolExecutor(max_workers=2) as commands: results = tuple(commands.map(assign, (('fork-a', 'worker'), ('fork-b', 'other')))) assert all(result.returncode == 0 for result in results), [result.stderr for result in results] - todos = tuple(todo for name in ('fork-a', 'fork-b') - for todo in comms.goals.todos.held_for_commander(comms.registry.require(name).incarnation)) + todos = tuple(todo for todo in comms.goals.todos.held_for_commander( + comms.registry.require('lead').incarnation) if todo.id != 'P40') assert {todo.id for todo in todos} == {'P41', 'P42'} - assert {todo.creator for todo in todos} == {'fork-a', 'fork-b'} - print('Concurrent fork allocations:', sorted((todo.creator, todo.id) for todo in todos)) + assert {todo.creator for todo in todos} == {'lead'} + assert {todo.held().parent for todo in todos} == {'fork-a', 'fork-b'} + print('Concurrent fork allocations:', sorted((todo.held().parent, todo.id) for todo in todos)) def test_two_commanders_cannot_assign_the_same_unit_concurrently(fork_bus, tmp_path): @@ -205,25 +231,81 @@ def assign(name): with ThreadPoolExecutor(max_workers=2) as commands: results = tuple(commands.map(assign, ('fork-a', 'fork-b'))) assert sum(result.returncode == 0 for result in results) == 1, [r.stderr for r in results] - held = tuple(todo for name in ('fork-a', 'fork-b') - for todo in comms.goals.todos.held_for_commander(comms.registry.require(name).incarnation)) + held = tuple(todo for todo in comms.goals.todos.held_for_commander( + comms.registry.require('lead').incarnation) if todo.id != 'P40') assert len(held) == 1 and held[0].holder == 'worker' refused = next(result for result in results if result.returncode) - assert f'is assigned by {held[0].creator}' in refused.stderr + assert f'is assigned by {held[0].held().parent}' in refused.stderr 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, + assigner=lead.incarnation, 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, + assigner=lead.incarnation, 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 ('P0', 'X1', 'broken', '1', 'P01', 'P-1', 'P+1', 'P١', 'P1.0'): + assert not ProblemNumber.accepts(text) + with store.writing() as db, pytest.raises(TodoError) as refused: + ProblemAllocation.reserve(db, text) + assert text in str(refused.value) + with store.writing() as db: + assert ProblemAllocation.reserve(db, retained.id) == ProblemNumber.parse(retained.id) + with store.snapshot() as db: + assert ProblemAllocation.select(db) == [] + assert store.list(source.github) == (retained,) diff --git a/tests/test_commander_run_rules.py b/tests/test_commander_run_rules.py index e8c0818ca5..6d231ebbc7 100644 --- a/tests/test_commander_run_rules.py +++ b/tests/test_commander_run_rules.py @@ -20,7 +20,7 @@ def test_declared_run_round_trips_and_short_runs_are_refused(tmp_path, source_re lead = thread('lead', tmp_path, 1) todo = store.create('P1', RepositoryName.from_wire('OpenHCSDev/agent-comms'), 'Repair the thread list', - creator=lead, owner=lead, generation='one', goal=GoalRef('lead', 1, 'one'), review_at=RecordedEventTime(1000), + creator=lead, assigner=lead.incarnation, owner=lead, generation='one', goal=GoalRef('lead', 1, 'one'), review_at=RecordedEventTime(1000), order=checked) assert FieldCodec.decode(Todo, FieldCodec.encode(todo)) == todo assert planned in todo.order.instructions(todo) @@ -38,7 +38,7 @@ def test_clean_transfers_keep_history_and_third_owner_requires_a_trace(tmp_path, lead = thread('lead', tmp_path, 1) first = store.create('P1', RepositoryName.from_wire('OpenHCSDev/agent-comms'), 'Repair the thread list', - creator=lead, owner=lead, generation='one', goal=GoalRef('lead', 1, 'one'), review_at=RecordedEventTime(1000), + creator=lead, assigner=lead.incarnation, owner=lead, generation='one', goal=GoalRef('lead', 1, 'one'), review_at=RecordedEventTime(1000), order=order(source_repository)) second = store.assign('P1', expected_revision=first.revision, owner=thread('second', tmp_path, 2), assigner=lead.incarnation, generation='two', goal=GoalRef('second', 2, 'two'), diff --git a/tests/test_declared_schema_carry.py b/tests/test_declared_schema_carry.py index 8c5002b54f..9bc0219c62 100644 --- a/tests/test_declared_schema_carry.py +++ b/tests/test_declared_schema_carry.py @@ -170,8 +170,8 @@ def test_a_bus_with_a_todo_deadline_and_old_proofs_carries_through_the_whole_cha banned=("No extra store",), report="Report the delivered commit", at=RecordedEventTime(10.0), first_call_deadline=RecordedEventTime(130.0)) created = TodoStore(tmp_path).create( - "todo-1", RepositoryName.from_wire("OpenHCSDev/agent-comms"), "Fix the sidebar", - creator=lead, goal=GoalRef("lead", 1.0, "goal-1"), owner=lead, generation="initial", + "P1", RepositoryName.from_wire("OpenHCSDev/agent-comms"), "Fix the sidebar", + creator=lead, assigner=lead.incarnation, goal=GoalRef("lead", 1.0, "goal-1"), owner=lead, generation="initial", review_at=RecordedEventTime(1000.0), order=order) with closing(sqlite3.connect(TodoStore.path_in(tmp_path))) as db: db.execute('ALTER TABLE "todo" RENAME COLUMN "review_at" TO "deadline"') @@ -195,7 +195,7 @@ def test_a_bus_with_a_todo_deadline_and_old_proofs_carries_through_the_whole_cha for step in CarryChain.at(REPO).steps: step.carry(tmp_path, apply=True) - assert TodoStore(tmp_path).get("todo-1").review_at == created.review_at + assert TodoStore(tmp_path).get("P1").review_at == created.review_at with closing(sqlite3.connect(TodoStore.path_in(tmp_path))) as db: assert "deadline" not in Todo.stored_columns(db) with closing(sqlite3.connect(proof)) as db: diff --git a/tests/test_held_problem_orders.py b/tests/test_held_problem_orders.py index 3ee3c79613..f071025c39 100644 --- a/tests/test_held_problem_orders.py +++ b/tests/test_held_problem_orders.py @@ -33,7 +33,7 @@ def held_commander(pr_commander): RequestAddVerb(text="Repair the copied order workflow without changing original retained state").run(source) lead = source.comms.registry.require(source.commander) todo = source.comms.goals.todos.create( - "P1", source.repos[0].github, "Repair the held workflow", creator=lead, + "P1", source.repos[0].github, "Repair the held workflow", creator=lead, assigner=lead.incarnation, owner=worker, generation="initial-order", goal=GoalRef(lead.name, lead.created_at, lead.goal.require().id), review_at=RecordedEventTime(time.time() + 1000), order=order(source.repos[0]), ) @@ -93,7 +93,7 @@ def test_second_unplanned_order_is_refused_without_recording(held_commander): ) lead = ctx.comms.registry.require(ctx.commander) ctx.comms.goals.todos.create( - "P2", ctx.repos[0].github, "Repair the second workflow", creator=lead, owner=lead, + "P2", ctx.repos[0].github, "Repair the second workflow", creator=lead, assigner=lead.incarnation, owner=lead, generation="second-workflow", goal=GoalRef(lead.name, lead.created_at, lead.goal.require().id), review_at=RecordedEventTime(time.time() + 1000), order=document.order, ) diff --git a/tests/test_holder_standing_rules.py b/tests/test_holder_standing_rules.py index d382dc2871..94de6e3aae 100644 --- a/tests/test_holder_standing_rules.py +++ b/tests/test_holder_standing_rules.py @@ -74,6 +74,7 @@ def held_bus(comms, tmp_path, source_repository, configured_repository): for problem in ("P1", "P2"): comms.goals.update_goal("worker", CreateProblemGoalAction( problem=problem, commander="lead", request="request", + assigner=comms.registry.require('lead').incarnation, repo=RepositoryName.from_wire("OpenHCSDev/agent-comms"), text=f"Repair {problem}", review_at=RecordedEventTime(now + 3600), order=PendingProblemOrder( @@ -218,6 +219,7 @@ def test_blocked_holder_is_reissued_its_accepted_order_after_editing_its_scope(h now = time.time() held_bus.goals.update_goal("worker", CreateProblemGoalAction( problem="P3", commander="lead", request="request", + assigner=held_bus.registry.require('lead').incarnation, repo=RepositoryName.from_wire("OpenHCSDev/agent-comms"), text="Repair P3", review_at=RecordedEventTime(now + 3600), order=replace(original, id='order-P3'), diff --git a/tests/test_order_progress.py b/tests/test_order_progress.py index 8f7ae573b2..ade713ef8e 100644 --- a/tests/test_order_progress.py +++ b/tests/test_order_progress.py @@ -124,7 +124,8 @@ def held(tmp_path): text='Repair the worker', goal_id='request', order=RequestGoalOrder())) ThreadCaller('lead').invoke(CommsTodoCreateTool(owner='worker', order=CreateProblemGoalAction( problem='P1', order=order(source), request='request', review_at=RecordedEventTime(1000), - commander='lead', repo=RepositoryName.from_wire('org/repo'), text='Repair the worker')), comms) + commander='lead', assigner=comms.registry.require('lead').incarnation, + repo=RepositoryName.from_wire('org/repo'), text='Repair the worker')), comms) return comms, unit_checkout(tmp_path / 'wt' / 'worker--P1') diff --git a/tests/test_order_scopes.py b/tests/test_order_scopes.py index 44bc3998e8..00054bdf18 100644 --- a/tests/test_order_scopes.py +++ b/tests/test_order_scopes.py @@ -75,6 +75,7 @@ def requested(comms, scope, source, problem='P1'): CommsTodoCreateTool(owner='worker', order=CreateProblemGoalAction( problem=problem, order=replace(order(source), scope=scope), request='request', review_at=RecordedEventTime(1000), commander='lead', + assigner=comms.registry.require('lead').incarnation, repo=RepositoryName.from_wire('OpenHCSDev/agent-comms'), text='Classify the source without changing it', )).apply(comms, caller) @@ -136,6 +137,7 @@ def test_carry_scopes_current_nested_and_prior_orders_on_a_copy(comms, tmp_path, worker = comms.registry.require('worker') for problem in TRIAGE_PROBLEMS: store.create(problem, original.repo, 'Classify the source', creator=lead, owner=worker, + assigner=lead.incarnation, generation=original.held().generation, review_at=original.review_at, goal=original.goal, order=replace(order(source_repository), scope=TriageScope())) copied = tmp_path / 'copy' diff --git a/tests/test_order_story.py b/tests/test_order_story.py index de1f4887d3..2611cca151 100644 --- a/tests/test_order_story.py +++ b/tests/test_order_story.py @@ -34,7 +34,8 @@ def test_order_story_copied_bus_strike_and_send_back_once(comms, tmp_path): caller.invoke(CommsTodoCreateTool(owner="worker", order=CreateProblemGoalAction( problem="P1", order=replace(order(source), edits=("Repair the existing record owner",)), request="request", review_at=RecordedEventTime(1000), - commander="lead", repo=RepositoryName.from_wire("org/repo"), text="Repair the worker")), comms) + commander="lead", assigner=comms.registry.require('lead').incarnation, + repo=RepositoryName.from_wire("org/repo"), text="Repair the worker")), comms) todo = comms.goals.todos.get("P1") actor = comms.registry.require("lead").incarnation reason = "The worker changed the wrong owner and left the decision duplicated." diff --git a/tests/test_owned_turn_goal_replacement.py b/tests/test_owned_turn_goal_replacement.py index 681f1b95cc..489d4f8004 100644 --- a/tests/test_owned_turn_goal_replacement.py +++ b/tests/test_owned_turn_goal_replacement.py @@ -61,6 +61,7 @@ async def test_turn_after_order_replacement_is_allowed(tmp_path, native_runtime_ comms.goals.update_goal(worker, CreateProblemGoalAction( problem='P1', order=order(source), request='request', review_at=RecordedEventTime(1000), commander='lead', + assigner=comms.registry.require('lead').incarnation, repo=RepositoryName.from_wire('org/repo'), text='Repair following input', )) saved.provider.response_gate = asyncio.Event() @@ -127,6 +128,7 @@ async def test_goal_published_during_preparation_is_current_at_reservation(tmp_p initial = comms.goals.update_goal('worker', CreateProblemGoalAction( problem='P1', order=order(source), request='request', review_at=RecordedEventTime(1000), commander='lead', + assigner=comms.registry.require('lead').incarnation, repo=RepositoryName.from_wire('org/repo'), text='Repair the turn owner', )) actor = comms.registry.require('lead').incarnation diff --git a/tests/test_proof_build.py b/tests/test_proof_build.py index 3f6a25f0b6..a7c0036839 100644 --- a/tests/test_proof_build.py +++ b/tests/test_proof_build.py @@ -148,7 +148,7 @@ def test_merge_in_head_cannot_close_until_the_recorded_build_contains_it( merged = replace(order(graph.repository), run=run).merged(graph.newer) todo = comms.goals.todos.create( "P1", graph.repository.github, "Prove installation instead of inferring it from merge", - creator=lead, owner=lead, generation="fixture-goal", + creator=lead, assigner=lead.incarnation, owner=lead, generation="fixture-goal", goal=GoalRef(lead.name, lead.created_at, "fixture-goal"), review_at=RecordedEventTime(1000), order=merged, ) @@ -298,7 +298,7 @@ def test_installed_proof_closes_and_older_build_is_refused_on_copied_bus( order=order, request="request", review_at=RecordedEventTime(4102444800), - commander="lead", + commander="lead", assigner=caller.identity(copied), repo=graph.repository.github, text="Bind proof to its build", ), diff --git a/tests/test_storage_violations.py b/tests/test_storage_violations.py index be19a0d850..1dd225f9bb 100644 --- a/tests/test_storage_violations.py +++ b/tests/test_storage_violations.py @@ -96,10 +96,10 @@ def test_storage_family_reads_once_and_renders_each_broken_fact( comms.registry.new_thread("worker", frozenset(), str(home)) ) comms.goals.todos.create( - "storage-case", + "P1", RepositoryName.from_wire("OpenHCSDev/agent-comms"), "Fix the filesystem paths", - creator=lead, + creator=lead, assigner=lead.incarnation, owner=worker, generation="storage-case", review_at=RecordedEventTime(1000), diff --git a/tests/test_todos.py b/tests/test_todos.py index f95590fe5c..7e740f9ffd 100644 --- a/tests/test_todos.py +++ b/tests/test_todos.py @@ -94,7 +94,7 @@ def _assign_process(root: str, checkout: str, name: str, created: float, gate) - gate.wait(timeout=5) try: row = store.assign( - "todo-1", expected_revision=1, + "P1", expected_revision=1, owner=thread(name, Path(root), created), assigner=thread("lead", Path(root), 1.0).incarnation, generation=f"generation-{name}", goal=GoalRef("lead", 1.0, "goal-1"), review_at=RecordedEventTime(1000.0), order=order(SourceRepository.of(Path(checkout))), @@ -109,7 +109,7 @@ def test_create_is_idempotent_and_goal_reference_is_historical(tmp_path: Path, s lead = thread("lead", tmp_path, 1.0) goal = GoalRef("lead", 1.0, "goal-1") original = store.create( - "todo-1", RepositoryName.from_wire("OpenHCSDev/agent-comms"), "Fix the sidebar", creator=lead, goal=goal, + "P1", RepositoryName.from_wire("OpenHCSDev/agent-comms"), "Fix the sidebar", creator=lead, assigner=lead.incarnation, goal=goal, owner=lead, generation="initial", review_at=RecordedEventTime(1000.0), order=order(source_repository), ) assert original.repo == RepositoryName.from_wire("OPENHCSDEV/agent-comms") @@ -117,22 +117,22 @@ def test_create_is_idempotent_and_goal_reference_is_historical(tmp_path: Path, s assert original.assignment == Assignment("lead", 1.0, "lead", 1.0, "initial") assert ( store.create( - "todo-1", RepositoryName.from_wire("openhcsdev/agent-comms"), "Fix the sidebar", creator=lead, goal=goal, + "P1", RepositoryName.from_wire("openhcsdev/agent-comms"), "Fix the sidebar", creator=lead, assigner=lead.incarnation, goal=goal, owner=lead, generation="initial", review_at=RecordedEventTime(1000.0), order=order(source_repository), ) == original ) with pytest.raises(TodoConflict): store.create( - "todo-1", RepositoryName.from_wire("openhcsdev/agent-comms"), "A different task", creator=lead, goal=goal, + "P1", RepositoryName.from_wire("openhcsdev/agent-comms"), "A different task", creator=lead, assigner=lead.incarnation, goal=goal, owner=lead, generation="initial", review_at=RecordedEventTime(1000.0), order=order(source_repository), ) assert TodoStore(tmp_path).list(RepositoryName.from_wire("openhcsdev/agent-comms")) == (original,) newer = thread("lead", tmp_path, 2.0) assert store.get(original.id).goal == goal with pytest.raises(TodoConflict): - store.apply("todo-1", ChangeTodoState(1, DoneTodoState, newer.incarnation, None)) - assert store.get("todo-1") == original + store.apply("P1", ChangeTodoState(1, DoneTodoState, newer.incarnation, None)) + assert store.get("P1") == original def test_withdraw_retires_a_released_problem(tmp_path, source_repository): @@ -140,7 +140,7 @@ def test_withdraw_retires_a_released_problem(tmp_path, source_repository): lead = thread("lead", tmp_path, 1.0) worker = thread("worker", tmp_path, 2.0) todo = store.create("P1", RepositoryName.from_wire("org/repo"), "Obsolete problem", - creator=lead, owner=worker, generation="first", goal=GoalRef("lead", 1.0, "request"), + creator=lead, assigner=lead.incarnation, owner=worker, generation="first", goal=GoalRef("lead", 1.0, "request"), review_at=RecordedEventTime(1000), order=order(source_repository)) released = store.apply(todo.id, ReleaseTodo(todo.revision, lead.incarnation, todo.held(), StrikeReason("The previous worker stopped before completing this assigned repair on main."))) @@ -165,7 +165,7 @@ def test_withdraw_refuses_a_delivered_problem(tmp_path, released, source_reposit store = TodoStore(tmp_path) lead = thread("lead", tmp_path, 1.0) todo = store.create("P1", RepositoryName.from_wire("org/repo"), "Delivered problem", - creator=lead, owner=lead, generation="first", goal=GoalRef("lead", 1.0, "request"), + creator=lead, assigner=lead.incarnation, owner=lead, generation="first", goal=GoalRef("lead", 1.0, "request"), review_at=RecordedEventTime(1000), order=order(source_repository)) delivered = store.apply(todo.id, DeliverTodo(todo.revision, lead.incarnation, OperationDelivery(at=RecordedEventTime(20)), todo.order.id)) @@ -182,7 +182,7 @@ def test_two_processes_assign_one_todo_before_any_worker_is_started(tmp_path: Pa store = TodoStore(tmp_path) lead = thread("lead", tmp_path, 1.0) store.create( - "todo-1", RepositoryName.from_wire("OpenHCSDev/agent-comms"), "One implementation", creator=lead, goal=GoalRef("lead", 1.0, "goal-1"), + "P1", RepositoryName.from_wire("OpenHCSDev/agent-comms"), "One implementation", creator=lead, assigner=lead.incarnation, goal=GoalRef("lead", 1.0, "goal-1"), owner=lead, generation="initial", review_at=RecordedEventTime(1000.0), order=order(source_repository), ) with Manager() as manager, ProcessPoolExecutor(max_workers=2) as pool: @@ -195,7 +195,7 @@ def test_two_processes_assign_one_todo_before_any_worker_is_started(tmp_path: Pa observed = [future.result(timeout=10) for future in results] assert sorted(result[0] for result in observed) == ["conflict", "won"] assert observed[0][1] == observed[1][1] - persisted = TodoStore(tmp_path).get("todo-1") + persisted = TodoStore(tmp_path).get("P1") assert persisted.assignment.owner == observed[0][1] assert persisted.revision == 2 assert len(store.list(RepositoryName.from_wire("OpenHCSDev/agent-comms"))) == 1 @@ -207,13 +207,13 @@ def test_validation_rejects_ambiguous_identity_and_bad_state(tmp_path: Path, sou for index, repo in enumerate(("org/repo/branch", "../repo", "org/", "org/repo name")): with pytest.raises(ValueError): store.create( - f"invalid-{index}", RepositoryName.from_wire(repo), "task", creator=lead, goal=GoalRef("lead", 1.0, "goal-1"), + f"invalid-{index}", RepositoryName.from_wire(repo), "task", creator=lead, assigner=lead.incarnation, goal=GoalRef("lead", 1.0, "goal-1"), owner=lead, generation="initial", review_at=RecordedEventTime(1000.0), order=order(source_repository), ) with pytest.raises(TodoError): GoalRef("lead", float("nan"), "goal-1") store.create( - "todo-1", RepositoryName.from_wire("org/repo"), "task", creator=lead, goal=GoalRef("lead", 1.0, "goal-1"), + "P1", RepositoryName.from_wire("org/repo"), "task", creator=lead, assigner=lead.incarnation, goal=GoalRef("lead", 1.0, "goal-1"), owner=lead, generation="initial", review_at=RecordedEventTime(1000.0), order=order(source_repository), ) with pytest.raises(ValueError): @@ -224,7 +224,7 @@ def test_merged_order_keeps_its_checked_fields(tmp_path: Path, source_repository store = TodoStore(tmp_path) lead = thread("lead", tmp_path, 1.0) todo = store.create( - "todo-1", RepositoryName.from_wire("OpenHCSDev/agent-comms"), "Fix the sidebar", creator=lead, goal=GoalRef("lead", 1.0, "goal-1"), + "P1", RepositoryName.from_wire("OpenHCSDev/agent-comms"), "Fix the sidebar", creator=lead, assigner=lead.incarnation, goal=GoalRef("lead", 1.0, "goal-1"), owner=lead, generation="initial", review_at=RecordedEventTime(1000.0), order=order(source_repository)) merged = replace(todo, order=todo.order.checked().merged("2b9c1f1ab")) cells = merged.order.board_cells(merged) @@ -276,7 +276,7 @@ def pr_order(ctx, pid, run: OrderRun = MechanicalOrderRun()): source = ctx.repos[0] lead = ctx.comms.registry.require(ctx.commander) return ctx.comms.goals.todos.create( - pid, source.github, "Fix the sidebar", creator=lead, owner=lead, + pid, source.github, "Fix the sidebar", creator=lead, assigner=lead.incarnation, owner=lead, generation="pr-work", goal=GoalRef(lead.name, 1.0, "pr-work"), review_at=RecordedEventTime(1000), order=replace(order(source), id=f"order-{pid}", run=run)) @@ -518,7 +518,7 @@ def test_installed_merge_closes_only_the_run_that_has_been_proven(pr_commander): originals = tuple( ctx.comms.goals.todos.create( f"P{index}", source.github, "Keep application work held until its run is recorded", - creator=lead, owner=worker, generation="merge-run", + creator=lead, assigner=lead.incarnation, owner=worker, generation="merge-run", goal=GoalRef(lead.name, lead.created_at, "merge-run"), review_at=RecordedEventTime(1000), order=replace(order(source), id=f"merge-order-{index}", run=run, diff --git a/tests/test_unowned_checkout_work.py b/tests/test_unowned_checkout_work.py index b8295b3500..e84d4583eb 100644 --- a/tests/test_unowned_checkout_work.py +++ b/tests/test_unowned_checkout_work.py @@ -142,7 +142,7 @@ def checkout_bus(tmp_path: Path) -> Comms: for problem in ('P1', 'P2'): c.goals.todos.create( problem, RepositoryName.from_wire('OpenHCSDev/agent-comms'), - 'Keep changes with their author', creator=lead, owner=worker, + 'Keep changes with their author', creator=lead, assigner=lead.incarnation, owner=worker, generation='shared-work', review_at=RecordedEventTime(now + 3600), order=order, goal=GoalRef('lead', lead.created_at, 'request')) # Both current IDs change, as they do when Core hands work out again. @@ -155,7 +155,7 @@ def checkout_bus(tmp_path: Path) -> Comms: goal=GoalRef('lead', lead.created_at, 'request')) c.goals.todos.create( 'P3', RepositoryName.from_wire('OpenHCSDev/agent-comms'), - 'Keep unrelated work separate', creator=lead, owner=worker, + 'Keep unrelated work separate', creator=lead, assigner=lead.incarnation, owner=worker, generation='other-work', review_at=RecordedEventTime(now + 3600), order=order.reissued(now + 2), goal=GoalRef('lead', lead.created_at, 'request')) row = c.goals.todos.get('P2') diff --git a/tests/verify_connector_tools.py b/tests/verify_connector_tools.py index 63b8f68aaf..e441ce9915 100644 --- a/tests/verify_connector_tools.py +++ b/tests/verify_connector_tools.py @@ -84,7 +84,7 @@ def arguments(comms: Comms, name: str, project: Path, checkout: Path) -> dict: order=pending, request="request", review_at=RecordedEventTime(time.time() + 600), - commander="claude-ai", + commander="claude-ai", assigner=comms.registry.require('claude-ai').incarnation, repo=RepositoryName.from_wire("OpenHCSDev/agent-comms"), text="Measure the connector order.", ) diff --git a/tools/cutover/carry_problem_numbers.py b/tools/cutover/carry_problem_numbers.py index 0249c35947..c98808ba5b 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,31 @@ 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)) diff --git a/tools/proofs/cn_phase3_record.py b/tools/proofs/cn_phase3_record.py index 7664cbc3c4..a25894ff9a 100644 --- a/tools/proofs/cn_phase3_record.py +++ b/tools/proofs/cn_phase3_record.py @@ -105,6 +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(ROOT).registry.require("record-lead").incarnation, review_at=RecordedEventTime(now + 3600), order=PendingProblemOrder( id=f"record-order-{problem}",