diff --git a/src/agent_comms/activity.py b/src/agent_comms/activity.py index 7b92623a57..64973ff204 100644 --- a/src/agent_comms/activity.py +++ b/src/agent_comms/activity.py @@ -103,9 +103,11 @@ class DrainDiagnostic(DeclaredFamily, affix="Diagnostic"): @abstractmethod def recovery(self) -> str: ... - @property - def headline(self) -> str: - return "Inbox unavailable" + headline: ClassVar[str] = "Inbox unavailable" + + @classmethod + def from_error(cls, owner: OwnerIdentity, error: Exception) -> Self: + return cls(owner, type(error).__name__, str(error), recorded_at=time.time()) @property def summary(self) -> str: @@ -125,12 +127,14 @@ def recovery(self) -> str: return "Waiting for recovery" +class FailedWakeDiagnostic(UnavailableDrainDiagnostic): + headline = "Wake failed" + + class NativeInputDrainDiagnostic(DrainDiagnostic): """A native request failure is not evidence that the inbox cannot be read.""" - @property - def headline(self) -> str: - return "Native request failed" + headline = "Native request failed" @property def recovery(self) -> str: diff --git a/src/agent_comms/bus_publication.py b/src/agent_comms/bus_publication.py index ac35eebdfe..1c1a6a8ad4 100644 --- a/src/agent_comms/bus_publication.py +++ b/src/agent_comms/bus_publication.py @@ -189,10 +189,11 @@ def deliveries(self) -> tuple[CommittedDelivery, ...]: def require_claim(self, metadata) -> None: """Factory already bound root/audience; deliveries need no retained-public exception.""" - def checkpoint_rows(self, offset: int, length: int, previous_sequence: int): + def checkpoint_rows(self, offset: int, length: int, previous_sequence: int, + claimed_after_seq: int): from .private_bus_checkpoint import Addressed, DeliverySources, Replies, ResponseKeys - yield from super().checkpoint_rows(offset, length, previous_sequence) + yield from super().checkpoint_rows(offset, length, previous_sequence, claimed_after_seq) if self.acknowledgement is not None: yield ResponseKeys(self.acknowledgement.publication_key) yield DeliverySources(self.message.seq, self.message.message_id, diff --git a/src/agent_comms/cli_commands.py b/src/agent_comms/cli_commands.py index 49ce4dd288..ef57c0120e 100644 --- a/src/agent_comms/cli_commands.py +++ b/src/agent_comms/cli_commands.py @@ -1237,7 +1237,7 @@ def apply(self, ctx: Comms) -> CliResult: home = Path.home() project = self.project.resolve() config = CommanderConfig( - commander=self.commander, root=ctx.root, cli=Harness.bin("agent-comms"), + commander=self.commander, root=ctx.root, state=home / ".local/state/agent-comms/commander", board=project / "docs/agents/fleet-board.md", narrative=project / "docs/agents/fleet-narrative.md", repos=tuple(Repo.of(repo) for repo in self.repos or (project,)), diff --git a/src/agent_comms/commander/config.py b/src/agent_comms/commander/config.py index c837f375d7..1b5714f46c 100644 --- a/src/agent_comms/commander/config.py +++ b/src/agent_comms/commander/config.py @@ -35,7 +35,6 @@ class CommanderConfig(CommanderRole, BaseModel): commander: str root: Path - cli: Path state: Path board: Path narrative: Path diff --git a/src/agent_comms/commander/order.py b/src/agent_comms/commander/order.py index 33e6ee33de..5148554443 100644 --- a/src/agent_comms/commander/order.py +++ b/src/agent_comms/commander/order.py @@ -105,19 +105,6 @@ def struck(todo, unit): return len(todo.live_strikes(unit)) -def hhmm(event): - return time.strftime('%H:%M', time.localtime(event.at)) - - -def deadline(due): - """HH:MM local time: the first such minute at or after now.""" - hour, minute = map(int, due.split(':')) - now = time.time() - day = time.localtime(now) - at = time.mktime((day.tm_year, day.tm_mon, day.tm_mday, hour, minute, 0, 0, 0, -1)) - return RecordedEventTime(at + 86400 if at < now - 60 else at) - - # Eliminate forwarding with inheritance and multiple parents; decompose the hierarchy if it is not sound. # A forwarder restates an implementation pair (paper 1 Fig 3: fields + forwarding leave routes to maintain), so # nothing the commander approves, delivers or orders may add or keep one. The cure is the hierarchy, never a @@ -404,14 +391,17 @@ def send(self, unit, due, document: OrderFile, clean=False, reassign=None, new=N # A miss already struck on the order this replaces is one miss: # the send-back records no second strike. struck_already = replacing and listed[group[0]].struck_on_order(unit) - common = dict(order=order, request=order.request(self.comms.registry.require(self.commander), tuple(group)), deadline=deadline(due), commander=self.commander) + request = order.request(self.comms.registry.require(self.commander), tuple(group)) + due_at = RecordedEventTime(time.time()).at_clock(due) if new: self.call(CommsTodoCreateTool(owner=unit, order=CreateProblemGoalAction( - problem=new[0], repo=Repo.of(Path(check.directory)).github, text=new[1], **common))) + problem=new[0], repo=Repo.of(Path(check.directory)).github, text=new[1], + order=order, request=request, deadline=due_at, commander=self.commander))) 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=trace, **common))) + assigner=self.incarnation, traced=trace, + order=order, request=request, deadline=due_at, commander=self.commander))) message += "\n" + "\n".join(order.pull_request_instruction(pid) for pid in group) sent = self.post(VerdictAudience((unit,)), message, answers) # The message is each order's recorded link; a verdict citing it lands on that order. @@ -485,18 +475,15 @@ def write_board(self): lines = ['# Fleet board', '', "GENERATED by agent-comms-commander order from Core's Todo store. Do not edit by hand:", '`order problem FILE.json` adds a row, `order send` orders it, `order close` or `order withdraw` removes it.', '', - '| Problem | Owner file | Owner | Order | Due | Progress | Delivered ' + '| Problem | Owner file | Owner | Order | Due | Check | Progress | Order state ' '| Strikes | Owner last acted |', - '|---|---|---|---|---|---|---|---|---|'] + '|---|---|---|---|---|---|---|---|---|---|'] listed, read = self.todos(), self.progress() activity = {unit: self.last_activity(unit) for unit in {t.holder for t in listed if t.holder}} for t in sorted(listed, key=lambda t: int(t.id[1:]) if t.id[1:].isdigit() else 0): - unit, last = t.holder, t.released[-1:] - owner = unit or f'unassigned: released from {last[0].previous.owner}: {last[0].reason.text}' - lines.append(f'| {t.id}: {t.text} | {t.order.owner.to_wire()} | {owner} | {t.order.id} | ' - f'{hhmm(t.deadline)} | {read[t.id].line if t.id in read else "-"} | ' - f'{"no" if t.pending else "yes, install pending"} | ' - f'{t.strike_count} | {activity[unit].clock() if unit else "-"} |') + lines.append(f'| {t.id}: {t.text.replace("|", "|")} | {t.order.board_cells(t)} | ' + f'{read[t.id].line if t.id in read else "-"} | {t.order.line} | ' + f'{t.strike_count} | {activity[t.holder].clock() if t.holder else "-"} |') history = self.comms.goals.todos.history_for_commander(self.incarnation) lines += ['', '## Withdrawn orders', '', *(f'- {line}' for t in history for line in t.order.withdrawals(t))] @@ -619,7 +606,7 @@ def first_calls(self, held, activity): self.call(CommsChangeProblemTool(problem=todo.id, operation=RecordFirstCallTodo( todo.revision, self.incarnation, event))) if not acted and self.comms.goals.todos.get(todo.id).revision != todo.revision: - missing.append(f'{unit} (order {order.id}, sent {hhmm(order.at)})') + missing.append(f'{unit} (order {order.id}, sent {order.at.clock()})') return missing def hunt(self, where, why, sent): @@ -663,7 +650,7 @@ def deadline_lines(self, pending: list[Todo], read: dict[str, OrderProgress], if not t.overdue(now): continue answer = self.reply(t.holder, ordered) - due = f'due {hhmm(t.deadline)}, order {t.order.id}' + due = f'due {t.deadline.clock()}, order {t.order.id}' late.append(f'{t.holder} ({due}, {work.line}{answer})') alarm = f'OVERDUE: {", ".join(late)} -> push or merge unfinished work now' return [alarm] * bool(late) @@ -704,7 +691,7 @@ def overdue(self): def unstrike(self, task: str, why: str, at: str | None = None) -> None: """A strike recorded in error is voided in Core, then narrated and posted to #comms.""" todo = self.comms.goals.todos.get(task) - live = [s for s in todo.live_strikes() if at in (None, hhmm(s.at))] + live = [s for s in todo.live_strikes() if at in (None, s.at.clock())] if not live: sys.exit(f'Refused: {task} has no live strike' + (f' recorded at {at}.' if at else '.')) voided = live[-1] @@ -714,7 +701,7 @@ def unstrike(self, task: str, why: str, at: str | None = None) -> None: unit = voided.previous.owner n = struck(self.comms.goals.todos.get(task), unit) line = (f'{time.strftime("%Y-%m-%d %H:%M")} {unit} (strike voided on {task}, ' - f'recorded {hhmm(voided.at)}; {n} strike(s) remain): {why}') + f'recorded {voided.at.clock()}; {n} strike(s) remain): {why}') self.on_record(line, 'Void', VerdictAudience.of(unit, todo.goal)) print(f'{unit}: {n} strike(s) on {task}') @@ -726,13 +713,13 @@ def due(self, pid: str, at: str, why: str) -> None: if len(why.split()) < 8: sys.exit('Refused: --why says why the deadline moves (8+ words)') refuse(why, *OUTGOING) # posted to #comms, so refused before Core records the deadline - moved = deadline(at) + moved = RecordedEventTime(time.time()).at_clock(at) self.call(CommsChangeProblemTool(problem=pid, operation=AssignTodo( todo.revision, todo.assignment, moved, todo.order, todo.goal))) self.on_record(f'{time.strftime("%Y-%m-%d %H:%M")} {todo.holder} ({pid} order {todo.order.id} due ' - f'{hhmm(todo.deadline)} moved to {hhmm(moved)}): {why}', 'Deadline', + f'{todo.deadline.clock()} moved to {moved.clock()}): {why}', 'Deadline', VerdictAudience.of(todo.held().owner, todo.goal)) - print(f'{pid}: due {hhmm(moved)}') + print(f'{pid}: due {moved.clock()}') @boundary_decoder def withdraw(self, pid: str, why: str, superseded_by: str | None = None) -> None: @@ -924,13 +911,14 @@ def run(self, ctx: Commander) -> None: @dataclass(frozen=True, kw_only=True) -class InstallVerb(RuledVerb): +class InstallVerb(OrderVerb, RuledVerb): help = "plan the fleet's next install; merged and delivered orders wait on it" at: str = option("at", help="HH:MM the install is due") installer: str = option("installer", help="the unit that runs the install") def run(self, ctx: Commander) -> None: - PlannedInstallDeadline(deadline(self.at), self.installer).plan(ctx.comms.root) + planned = RecordedEventTime(time.time()).at_clock(self.at) + PlannedInstallDeadline(planned, self.installer).plan(ctx.comms.root) print(f'install due {self.at}, run by {self.installer}') diff --git a/src/agent_comms/commander/request.py b/src/agent_comms/commander/request.py index a93f5ea4e8..382746005d 100644 --- a/src/agent_comms/commander/request.py +++ b/src/agent_comms/commander/request.py @@ -16,7 +16,7 @@ from ..goal_states import ActiveGoal from ..goals import Goal, RequestGoalOrder from ..todos import Todo -from .order import Commander, CommanderVerb, RuledVerb, hhmm +from .order import Commander, CommanderVerb, RuledVerb def open_requests(ctx: Commander) -> tuple[tuple[Goal, tuple[Todo, ...]], ...]: @@ -64,7 +64,7 @@ def run(self, ctx: Commander) -> None: for goal, served in requests: print(f"{goal.id} {goal.text}") for todo in served: - print(f" {todo.id} due {hhmm(todo.deadline)}: {todo.text}") + print(f" {todo.id} due {todo.deadline.clock()}: {todo.text}") def main(a: list[str], ctx: Commander) -> None: diff --git a/src/agent_comms/commander/rules/caution-must-pay.md b/src/agent_comms/commander/rules/caution-must-pay.md new file mode 100644 index 0000000000..7033b344d7 --- /dev/null +++ b/src/agent_comms/commander/rules/caution-must-pay.md @@ -0,0 +1,9 @@ +--- +name: caution-must-pay +description: "overcaution is a defect, not diligence: every wait, gate, rehearsal or one-at-a-time step must name the specific failure it prevents that nothing else already catches; one that cannot is deleted. Stalling to look careful is the same failure as stalling." +type: feedback +--- + +Tristan's words: overcaution caused most of today's problems; it is performative work that stalls instead of shipping. + +Today's examples: installs waiting on idle turns (about three hours), a practice install required on every PR (merges from 26 an hour to 2), the word gate blocking file names, one failed turn parking a goal, storage strikes taking critical orders, the growth gate failing PRs, and landing PRs one at a time when drop-on-fail already covered the risk. diff --git a/src/agent_comms/commander/rules/review-like-linus.md b/src/agent_comms/commander/rules/review-like-linus.md new file mode 100644 index 0000000000..c32751eccf --- /dev/null +++ b/src/agent_comms/commander/rules/review-like-linus.md @@ -0,0 +1,9 @@ +--- +name: review-like-linus +description: "review like Linus Torvalds on the kernel mailing list: when work is stupid, say it is stupid in plain words, name exactly what is broken, why the author knew better, and the fix, then stop; no HR softening and no hedging; the flame is always attached to a concrete technical failure, aimed at the work, never slurs" +type: feedback +--- + +Tristan's reason, in his words: politeness is how incompetence gets tolerated and how tone drifts soft; the harshest stretches of the fleet's work were its most productive, because nothing stupid was accepted. Linus's mailing-list reviews are the precedent: decades of blunt, exact criticism that kept the kernel from rotting. + +Tonight's examples: a quoted deadline rule then broken by turning a ceiling into an appointment; a typed-order change shipped without running it against the orders already stored; a unit sitting 101 minutes on a ten-minute deletion. diff --git a/src/agent_comms/drain_failures.py b/src/agent_comms/drain_failures.py index e23672ec21..861cd854fd 100644 --- a/src/agent_comms/drain_failures.py +++ b/src/agent_comms/drain_failures.py @@ -4,7 +4,6 @@ import asyncio import sqlite3 -import time from contextlib import asynccontextmanager from functools import partial from typing import TYPE_CHECKING @@ -34,6 +33,7 @@ async def drain_failures( comms: Comms, thread: str, owner: OwnerIdentity, + unavailable: type[UnavailableDrainDiagnostic] = UnavailableDrainDiagnostic, ): """Record and propagate failures; the observation owner waits for real change.""" snapshot = await Coordination.run_worker(comms.registry.snapshot) @@ -42,9 +42,7 @@ async def drain_failures( )) async def record(kind, error) -> None: - diagnostic = kind( - owner, type(error).__name__, str(error), recorded_at=time.time() - ) + diagnostic = kind.from_error(owner, error) await Coordination.run_worker(partial( comms.agents.set_drain_diagnostic, thread, owner, diagnostic, source_error=error, @@ -61,18 +59,10 @@ async def record(kind, error) -> None: await record(NativeInputDrainDiagnostic, error) raise except UNAVAILABLE_DRAIN_ERRORS as error: - await record(UnavailableDrainDiagnostic, error) + await record(unavailable, error) raise except ValueError as error: - diagnostic = StoppedDrainDiagnostic( - owner, type(error).__name__, str(error), recorded_at=time.time() - ) - await Coordination.run_worker(partial( - comms.agents.set_drain_diagnostic, thread, owner, diagnostic, - source_error=error, - )) + await record(StoppedDrainDiagnostic, error) raise else: - current = await Coordination.run_worker(partial(comms.registry.require, thread)) - if not current.turn_record.read().executing: - await comms.agents.recover_drain(thread, owner, observed) + await comms.agents.recover_drain(thread, owner, observed) diff --git a/src/agent_comms/input_drain.py b/src/agent_comms/input_drain.py index d926e214ec..10840ce432 100644 --- a/src/agent_comms/input_drain.py +++ b/src/agent_comms/input_drain.py @@ -26,6 +26,7 @@ from .comms import Comms from .coordination_errors import IdentityConflict from .drain_failures import DEFERRED_DRAIN_ERRORS, UNAVAILABLE_DRAIN_ERRORS, drain_failures +from .activity import FailedWakeDiagnostic from .coordination_cohort import next_sealed_assignment from .coordinator import Coordination from .cursor_owner import CursorOwner @@ -161,20 +162,46 @@ async def drain(self, inputs, session_id: str, root_id: str) -> int: class TickState(ABC): - @abstractmethod async def next( self, observation: InputObservation, session_id: str, revision: TickRevision, + ) -> SettledTick | FullTick | FailedTick: + try: + return await self._next(observation, session_id, revision) + except UNAVAILABLE_DRAIN_ERRORS as error: + return FailedTick(error) + + @abstractmethod + async def _next( + self, observation: InputObservation, session_id: str, revision: TickRevision, ) -> SettledTick | FullTick: ... class UnsettledTick(TickState): - async def next( + async def _next( self, observation: InputObservation, session_id: str, revision: TickRevision, ) -> FullTick: read = await observation._own_entry(session_id) return FullTick(replace(revision, registry=read.revision), read.entry) +@dataclass(frozen=True, slots=True) +class FailedTick(UnsettledTick): + error: Exception + + async def apply( + self, observation: InputObservation, session_id: str, watcher: WireWatch, + ) -> TickState: + thread = observation.sessions.require(session_id) + snapshot = await Coordination.run_worker(observation.comms.registry.snapshot) + owner = snapshot.owner_identity(thread) + diagnostic = FailedWakeDiagnostic.from_error(owner, self.error) + await Coordination.run_worker(partial( + observation.comms.agents.set_drain_diagnostic, thread, owner, diagnostic, + source_error=self.error, + )) + return self + + @dataclass(frozen=True, slots=True) class FullTick: revision: TickRevision @@ -230,16 +257,16 @@ class SettledTick(TickState): claimed_after_seq: int | None through_seq: int - async def next( + async def _next( self, observation: InputObservation, session_id: str, revision: TickRevision, ) -> SettledTick | FullTick: if revision.other != self.revision.other or not observation._quiescent(session_id): - return await UnsettledTick().next(observation, session_id, revision) + return await UnsettledTick()._next(observation, session_id, revision) if revision.registry != self.revision.registry: # A standby's wait graph also depends on other threads' entries. if observation.effects.turns.goals.in_standby(session_id): - return await UnsettledTick().next(observation, session_id, revision) - step = await UnsettledTick().next(observation, session_id, revision) + return await UnsettledTick()._next(observation, session_id, revision) + step = await UnsettledTick()._next(observation, session_id, revision) if step.entry != self.entry: return step else: @@ -261,6 +288,10 @@ async def apply(self, observation: InputObservation, session_id: str, watcher: W class InputObservation: """File-driven observation and process-local wake behavior.""" + comms: Comms + sessions: SessionLifecycle + closing: bool + def _tick_paths(self) -> TickRevisionPaths: configuration = self.sessions.config.configuration return TickRevision.paths( @@ -284,8 +315,8 @@ def _tick_revision( def _retire_observer(self, session_id: str, task: asyncio.Task[None]) -> None: state = self.state(session_id) - if state.drain_task is task: - state.observation_watcher = None + if state.drain_task is task and not self.closing: + raise SystemExit(1) def wake_observer(self, session_id: str) -> None: state = self.state(session_id) diff --git a/src/agent_comms/private_bus_checkpoint.py b/src/agent_comms/private_bus_checkpoint.py index 5d43cc2048..321bd987c6 100644 --- a/src/agent_comms/private_bus_checkpoint.py +++ b/src/agent_comms/private_bus_checkpoint.py @@ -1125,7 +1125,8 @@ def _install(cls, bus: WireLog, marker: WireMetadata) -> PrefixWitness: def collect(offset, raw, record): nonlocal digest, through_seq digest = _chain(digest, raw) - through_seq = _index_row(db, offset, raw, record, through_seq) + through_seq = _index_row(db, offset, raw, record, through_seq, + marker.claimed_after_seq) with db: for _ in bus.verified_records_unlocked(marker, on_row=collect): @@ -1212,7 +1213,8 @@ def commit_tail(cls, bus: WireLog, marker: WireMetadata, db: sqlite3.Connection, with db: sequence = prior.through_seq for offset, raw, record in additions: - sequence = _index_row(db, offset, raw, record, sequence) + sequence = _index_row(db, offset, raw, record, sequence, + marker.claimed_after_seq) if sequence != expected.through_seq: raise RelationViolationError("Indexed tail differs from its sealed sequence.") cls.from_witness(expected).upsert(db) @@ -1278,9 +1280,9 @@ def _saved(db: sqlite3.Connection) -> PrefixWitness: def _index_row(db: sqlite3.Connection, offset: int, raw: bytes, - record: WireRecord, previous_sequence: int) -> int: + record: WireRecord, previous_sequence: int, claimed_after_seq: int) -> int: sequence = record.sequence_after(previous_sequence) - for row in record.checkpoint_rows(offset, len(raw), previous_sequence): + for row in record.checkpoint_rows(offset, len(raw), previous_sequence, claimed_after_seq): row.insert(db) return sequence diff --git a/src/agent_comms/recorded_time.py b/src/agent_comms/recorded_time.py index 35bb91357d..b4464897df 100644 --- a/src/agent_comms/recorded_time.py +++ b/src/agent_comms/recorded_time.py @@ -116,6 +116,13 @@ def before(self, timestamp: float) -> bool: def clock(self) -> str: return time.strftime("%H:%M:%S", time.localtime(self.at)) + def at_clock(self, clock: str) -> RecordedEventTime: + """The next local HH:MM minute, anchored to this recorded instant.""" + hour, minute = map(int, clock.split(':')) + day = time.localtime(self.at) + at = time.mktime((day.tm_year, day.tm_mon, day.tm_mday, hour, minute, 0, 0, 0, -1)) + return RecordedEventTime(at + 86400 if at < self.at - 60 else at) + def recorded_at(self, label: str) -> float: return self.at diff --git a/src/agent_comms/schedule_rules.py b/src/agent_comms/schedule_rules.py index 5577c28c18..3e038205f6 100644 --- a/src/agent_comms/schedule_rules.py +++ b/src/agent_comms/schedule_rules.py @@ -9,6 +9,7 @@ from .coordinator import Coordination from .drain_failures import drain_failures +from .activity import FailedWakeDiagnostic from .errors import RelationViolationError from .reservation_rules import ReservationRule, RuleCheck, ReservationViolationError from .routing import ScheduledTurn @@ -160,7 +161,7 @@ async def run(self) -> None: hold = await Coordination.run_worker(inputs.comms.owners.maintenance.read) if hold.turn_pending: return hold - async with drain_failures(inputs.comms, self.thread, self.owner): + async with drain_failures(inputs.comms, self.thread, self.owner, FailedWakeDiagnostic): before = await inputs._observe_private_revision(session_id, root_id) result = await effects.turns.run_selected(session_id, self.execution) if result is not None: diff --git a/src/agent_comms/selected_request.py b/src/agent_comms/selected_request.py index d6fe4e79f0..e20d0b00db 100644 --- a/src/agent_comms/selected_request.py +++ b/src/agent_comms/selected_request.py @@ -9,6 +9,7 @@ from .coordination_errors import StaleFence from .coordinator import Coordination +from .drain_failures import UNAVAILABLE_DRAIN_ERRORS from .native_pi import NativePiTerminalFailure, NativeRequestFailure from .private_send_claim import PrivateSendClaim from .selected_participant import SelectedParticipant @@ -50,9 +51,9 @@ async def native_failures(self): except NativeRequestFailure as error: await self._native_failure(error) raise - except CancelledError: - # Native execution has unwound child/tool cleanup before the - # reserved request sees cancellation of its abandoned result. + except (CancelledError, *UNAVAILABLE_DRAIN_ERRORS): + # After native cleanup, cancellation or an unavailable read leaves + # this abandoned request UNKNOWN; settlement errors propagate. await self._settle_unknown() raise diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index ea58133091..bff8fbbd79 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -414,10 +414,6 @@ def read(self, repo: str, state: str, order: ProblemOrder) -> tuple[PullRequest, return self._read[repo, state, checked.id] -def _clock(at: float) -> str: - return time.strftime("%H:%M", time.localtime(at)) - - @dataclass(frozen=True) class ProgressPoint: """A revision and its time in the order's progress; Git rows use committer time.""" @@ -483,7 +479,7 @@ class MovingTrend(DistanceTrend): @property def projection(self) -> str: - return f"; distance {self.distance}, projected done {_clock(self.done)}" + return f"; distance {self.distance}, projected done {RecordedEventTime(self.done).clock()}" @dataclass(frozen=True) @@ -780,7 +776,7 @@ def found(cls, source: ProgressSource) -> Iterator[Self]: @property def state(self) -> str: - return f"committed {self.sha} at {_clock(self.at)}, not pushed" + return f"committed {self.sha} at {RecordedEventTime(self.at).clock()}, not pushed" @dataclass(frozen=True) @@ -797,7 +793,7 @@ def found(cls, source: ProgressSource) -> Iterator[Self]: @property def state(self) -> str: - return f"pushed {self.sha} at {_clock(self.at)}, no PR" + return f"pushed {self.sha} at {RecordedEventTime(self.at).clock()}, no PR" @dataclass(frozen=True) @@ -895,8 +891,7 @@ def after(cls, installer: str, now: float) -> PlannedInstallDeadline: return cls(RecordedEventTime(now + cls.cadence), installer) def overdue(self, now: float) -> tuple[str, ...]: - due = time.strftime("%H:%M", time.localtime(self.at.at)) - return (f"due {due}, installer {self.installer}",) * (now >= self.at.at) + return (f"due {self.at.clock()}, installer {self.installer}",) * (now >= self.at.at) def plan(self, root: Path) -> None: path = self.path_in(root) @@ -912,10 +907,10 @@ def refusals(self, work: str, commander: str) -> tuple[str, ...]: def alarm(self, waiting: list[str], now: float) -> list[str]: if not waiting: return [] - due = time.strftime("%H:%M", time.localtime(self.at.at)) if now > self.at.at: - return [f"INSTALL LATE (due {due}), not the orders; waiting: {', '.join(waiting)}"] - return [f"INSTALL due {due}; waiting: {', '.join(waiting)}"] + return [f"INSTALL LATE (due {self.at.clock()}), not the orders; " + f"waiting: {', '.join(waiting)}"] + return [f"INSTALL due {self.at.clock()}; waiting: {', '.join(waiting)}"] @dataclass(frozen=True) @@ -1295,6 +1290,9 @@ class ProblemFinding(DeclaredFamily, affix="ProblemFinding"): def instructions(self) -> tuple[str, ...]: return () + def place(self) -> str: + return "-" + @dataclass(frozen=True) class UnlocatedProblemFinding(ProblemFinding): @@ -1306,7 +1304,10 @@ class LocatedProblemFinding(ProblemFinding): where: FileLine def instructions(self) -> tuple[str, ...]: - return (f"Where: {self.where.to_wire()}",) + return (f"Where: {self.place()}",) + + def place(self) -> str: + return self.where.to_wire() @dataclass(frozen=True, kw_only=True) @@ -1328,6 +1329,9 @@ def pull_request_ids(cls, body: str) -> tuple[str, ...]: return tuple(line.removeprefix(cls.pull_request_label).strip() for line in body.splitlines() if line.startswith(cls.pull_request_label)) + def board_cells(self, todo: Todo) -> str: + return f"{self.finding.place()} | unassigned | - | - | -" + def checked(self) -> CheckedProblemOrder: raise ValueError("This problem has no checked order.") @@ -1421,6 +1425,13 @@ def checked(self) -> CheckedProblemOrder: default_factory=NoOrderDeletion, metadata={"wire_omit_default": True}) serves: type[OrderPhase] = field(default=UnplannedPhase, metadata={"wire_omit_default": True}) + def board_cells(self, todo: Todo) -> str: + last = todo.released[-1:] + owner = todo.holder or (f"unassigned: released from {last[0].previous.owner}: " + f"{last[0].reason.text}") + check = json.dumps(FieldCodec.encode(self.check)).replace("|", "|") + return f"{self.owner.to_wire()} | {owner} | {self.id} | {todo.deadline.clock()} | {check}" + @property def phase(self) -> type[OrderPhase]: return self.serves @@ -1444,7 +1455,6 @@ def __post_init__(self): if problems: raise ValueError("\n".join(problems)) - pending: ClassVar[bool] line: ClassVar[str] NAMED_FILE: ClassVar[re.Pattern] = re.compile(r"([\w./-]+\.\w+):\d+") @@ -1467,7 +1477,7 @@ def copy_as[Kind: ProblemOrder](self, kind: type[Kind], **evidence) -> Kind: return kind(**{item.name: getattr(self, item.name) for item in fields(kind) if item.name not in evidence}, **evidence) - def merged(self, sha: str) -> ProblemOrder: + def merged(self, sha: str) -> MergedProblemOrder: return self.copy_as(MergedProblemOrder, sha=sha) @property @@ -1529,22 +1539,11 @@ def require_checked(self, problem: str) -> None: raise ValueError(f"An installed claim needs the recorded check result of {problem}; " "its order has none.") - def require_close(self): - raise ValueError("Closing requires recorded installed application evidence.") - def withdrawn(self, previous: Assignment, why: str, superseded_by: str | None) -> ProblemOrder: self.require_withdrawal() return self.copy_as(WithdrawnProblemOrder, previous=previous, why=why, superseded_by=superseded_by) - def withdrawals(self, todo: Todo) -> tuple[str, ...]: - """How this order retired its problem without delivery, as the record and board say.""" - return () - - def waiting_on_install(self, todo: Todo, read: dict[str, OrderProgress]) -> tuple[Todo, ...]: - """The order waiting on the fleet's install to record it working.""" - return () - @dataclass(frozen=True, kw_only=True) class MergedProblemOrder(CheckedProblemOrder): @@ -1591,7 +1590,6 @@ def require_withdrawal(self) -> None: @dataclass(frozen=True, kw_only=True) class DroppedProblemOrder(UnorderedProblemOrder, CheckedProblemOrder): replacing_message: str - pending: ClassVar[bool] = False line: ClassVar[str] = "Dropped for a newer order" @@ -1782,6 +1780,10 @@ class Assignment: previous_owners: tuple[ThreadIncarnation, ...] = field( default=(), metadata={"wire_omit_default": True}) + @classmethod + def of(cls, owner: ThreadIncarnation, parent: ThreadIncarnation, generation: str) -> Assignment: + return cls(owner.name, owner.wire_birth, parent.name, parent.wire_birth, generation) + def __post_init__(self) -> None: _text(self.owner, "Thread name", 128) _text(self.parent, "Thread name", 128) @@ -1921,8 +1923,8 @@ class AssignTodo(TodoOperation): """The order's commander gives the problem to a unit.""" proposed: Assignment deadline: EventTime - order: PendingProblemOrder - goal: GoalRef + order: PendingProblemOrder | MergedProblemOrder + goal: GoalReference traced: FileLine | None = field(default=None, metadata={"wire_omit_default": True}) def apply(self, current: Todo) -> Todo: @@ -2346,6 +2348,20 @@ class Todo(TypedTable): corrections: tuple[WordingCorrection, ...] = () messages: tuple[OrderMessage, ...] = () + unrequested_goal: ClassVar[GoalReference] = NoGoalReference() + + @classmethod + def ordered(cls, todo_id: str, repo: str, text: str, creator: ThreadIncarnation, + assignment: Assignment, deadline: EventTime, + order: PendingProblemOrder | MergedProblemOrder, + goal: GoalReference = unrequested_goal) -> Todo: + _text(todo_id, "Todo ID", 128) + _text(text, "Todo text", 2000) + deadline.require_deadline() + goal.require_reference() + return cls(todo_id, _repo(repo), text, creator.name, creator.wire_birth, OpenTodoState, + 1, goal, assignment, order, deadline) + @property def pending(self): return self.order.pending @@ -2725,27 +2741,11 @@ def create( order: PendingProblemOrder, goal: GoalRef, ) -> Todo: - _text(todo_id, "Todo ID", 128) - _text(text, "Todo text", 2000) - repo_id = _repo(repo) identity, assignee = _participant(creator), _thread(owner) - deadline.require_deadline() - goal.require_reference() order.require_writable() - proposed = Todo( - id=todo_id, - repo=repo_id, - text=text, - creator=identity.name, - creator_created=identity.created_at, - state=OpenTodoState, - revision=1, - goal=goal, - assignment=Assignment(assignee.name, assignee.created_at, - identity.name, identity.created_at, generation), - deadline=deadline, - order=order, - ) + proposed = Todo.ordered( + todo_id, repo, text, identity, Assignment.of(assignee, identity, generation), + deadline, order, goal) with self.writing() as db: proposed.insert_if_absent(db) current = self._current(db, todo_id) @@ -2804,9 +2804,6 @@ def assign( order: PendingProblemOrder, traced: FileLine | None = None, ) -> Todo: - identity = _thread(owner) - proposed = Assignment( - identity.name, identity.created_at, assigner.name, assigner.wire_birth, generation) + proposed = Assignment.of(_thread(owner), assigner, generation) return self.apply( todo_id, AssignTodo(expected_revision, proposed, deadline, order, goal, traced)) - diff --git a/src/agent_comms/wire_record.py b/src/agent_comms/wire_record.py index 1cca5728ea..d06623bff0 100644 --- a/src/agent_comms/wire_record.py +++ b/src/agent_comms/wire_record.py @@ -102,7 +102,8 @@ def record_key(self, seen: set[str]) -> None: def require_claim(self, metadata: WireMetadata) -> None: ... @abstractmethod - def checkpoint_rows(self, offset: int, length: int, previous_sequence: int): ... + def checkpoint_rows(self, offset: int, length: int, previous_sequence: int, + claimed_after_seq: int): ... class MessageWireRecord(WireRecord): @@ -116,8 +117,13 @@ def sequence_after(self, previous: int) -> int: raise MalformedRecord("Bus sequence is not increasing.") return self.message.seq - def checkpoint_rows(self, offset: int, length: int, previous_sequence: int): - return self.message.checkpoint_rows(previous_sequence) + def checkpoint_rows(self, offset: int, length: int, previous_sequence: int, + claimed_after_seq: int): + # Retained messages precede this root's own allocations. Their source + # positions stay intact; only this root's suffix can contain an UNKNOWN gap. + predecessor = (self.message.seq - 1 if self.message.seq <= claimed_after_seq + else previous_sequence) + return self.message.checkpoint_rows(predecessor) class WireObservation(DeclaredFamily, affix="WireObservation"): @@ -221,9 +227,10 @@ class AuthoredObservationWireRecord(ObservationWireRecord): def read(self) -> WireObservation: return self.observation - def checkpoint_rows(self, offset: int, length: int, previous_sequence: int) -> tuple: + def checkpoint_rows(self, offset: int, length: int, previous_sequence: int, + claimed_after_seq: int) -> tuple: return DeferredObservationWireRecord.decode(self.to_wire()).checkpoint_rows( - offset, length, previous_sequence + offset, length, previous_sequence, claimed_after_seq ) def to_wire(self) -> dict: @@ -257,7 +264,8 @@ def read(self) -> WireObservation: except MalformedRecord as error: raise StoredHistoryUnreadable(f"Unreadable {owner.row_label(data)}: {error}") from error - def checkpoint_rows(self, offset: int, length: int, previous_sequence: int) -> tuple: + def checkpoint_rows(self, offset: int, length: int, previous_sequence: int, + claimed_after_seq: int) -> tuple: owner, data = self.member return owner.checkpoint_from_wire(data, offset, length, previous_sequence) diff --git a/tests/conftest.py b/tests/conftest.py index 3393fbc2eb..47f7a4e264 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -282,9 +282,8 @@ def commander(comms: Comms, tmp_path: Path, monkeypatch: pytest.MonkeyPatch): comms.registry.declare(comms.registry.new_thread(name, frozenset({'comms'}), str(tmp_path), created_at=float(created))) comms.goals.update_goal("worker", SetGoalAction(text="Original commander brief", goal_id="ongoing")) comms.goals.update_goal("lead", SetGoalAction(text="Owner request", goal_id="request", order=RequestGoalOrder())) - import sys from agent_comms.commander.config import CommanderConfig - CommanderConfig(commander="lead", root=comms.root, cli=Path(sys.executable).parent / "agent-comms", + CommanderConfig(commander="lead", root=comms.root, state=tmp_path / "state", board=tmp_path / "board.md", narrative=tmp_path / "narrative.md", repos=(), worktrees=tmp_path / "wt", scratch=tmp_path / "scratch", disk=tmp_path).write(tmp_path) diff --git a/tests/test_commander_skill_hook.py b/tests/test_commander_skill_hook.py index 951d6ac8c0..4e623969a2 100644 --- a/tests/test_commander_skill_hook.py +++ b/tests/test_commander_skill_hook.py @@ -4,7 +4,6 @@ import os import subprocess import sys -from pathlib import Path import pytest @@ -27,7 +26,6 @@ def test_session_start_reads_all_packaged_files_without_a_bus(tmp_path, source, config = CommanderConfig( commander="fresh-commander", root=bus, - cli=Path(sys.executable), state=tmp_path / "state", board=tmp_path / "board.md", narrative=tmp_path / "narrative.md", diff --git a/tests/test_live_drain_failure.py b/tests/test_live_drain_failure.py index 279ef8ba87..a27054687e 100644 --- a/tests/test_live_drain_failure.py +++ b/tests/test_live_drain_failure.py @@ -118,7 +118,7 @@ async def test_successful_drain_clears_failure_during_active_turn(tmp_path): await agent.inputs.drain_inbox("beta") current = owner_view(comms) assert current.activity.readiness.source_diagnostic() is None - assert current.thread.active_turn.executing + assert current.thread.turn_record.read().executing assert current.presentation.status.working assert not current.presentation.status.details activity = (comms.root / "activity.jsonl").read_bytes() @@ -241,10 +241,15 @@ async def broken(_session): monkeypatch.setattr(agent.inputs, "drain_inbox", broken) try: - agent.inputs.ensure_live_drain("beta") - task = agent.inputs.state("beta").drain_task + # Invoke the real retirement callback explicitly so its process exit + # does not terminate pytest's own event loop. + task = asyncio.create_task(agent.inputs.observe("beta")) + agent.inputs.state("beta").drain_task = task with pytest.raises(ValueError, match="unexpected drain implementation failure"): await task + with pytest.raises(SystemExit) as stopped: + agent.inputs._retire_observer("beta", task) + assert stopped.value.code == 1 assert isinstance( owner_view(comms).activity.readiness.source_diagnostic(), StoppedDrainDiagnostic, diff --git a/tests/test_private_bus_checkpoint.py b/tests/test_private_bus_checkpoint.py index 280f55dd27..310615daeb 100644 --- a/tests/test_private_bus_checkpoint.py +++ b/tests/test_private_bus_checkpoint.py @@ -130,6 +130,37 @@ def test_human_duplicate_check_includes_retained_rows_without_delivery_pointers( +def test_sparse_imported_prefix_keeps_references_and_accepts_fresh_human_send(tmp_path): + from agent_comms.message_reference import MessageReference + from agent_comms.private_bus_checkpoint import PublicMessages + from agent_comms.wire_record import PublicWireRecord + + source, _ = _root(tmp_path) + first = Message('sender', 'Alice', 'selected original', InfoKind, seq=7) + second = Message('Alice', 'sender', 'selected answer', InfoKind, seq=43, + answers=(MessageReference(first.seq, first.message_id),)) + copied = Comms(tmp_path / 'selected') + for thread in source.registry.snapshot().threads.values(): + copied.registry.declare(thread) + copied.messaging.initialize_private_initial_protocol(history=(first, second)) + before = copied.bus.log.path.read_bytes() + with copied.bus.log.checked_read() as checked: + PublicMessages.require_contiguous(checked.connection) + assert [(row.seq, row.previous_seq) for row in PublicMessages.select( + checked.connection, order_by=('seq',))] == [(7, 6), (43, 42)] + restored = tuple(Message.from_wire(json.loads(row)) for row in before.splitlines()) + assert restored == (first, second) + fresh = copied.messaging.send_user_message('Alice', 'fresh human input', worktree=str(tmp_path)) + assert fresh.seq == 44 + assert copied.bus.log.path.read_bytes().startswith(before) + # A gap after the imported boundary is still this root's uncertain allocation. + after = PublicWireRecord(Message('sender', 'Alice', 'own gap', InfoKind, seq=46)) + public, = after.checkpoint_rows(0, 1, fresh.seq, second.seq) + assert (public.seq, public.previous_seq) == (46, 44) + with pytest.raises(ValueError, match='not increasing'): + after.sequence_after(46) + + def test_marker_bound_complete_addressed_pages(tmp_path: Path) -> None: comms, root_id = _root(tmp_path) with pytest.raises(RelationViolationError, match="already installed"): diff --git a/tests/test_rts_open_carry.py b/tests/test_rts_open_carry.py deleted file mode 100644 index faaa9f7aa0..0000000000 --- a/tests/test_rts_open_carry.py +++ /dev/null @@ -1,59 +0,0 @@ -"""The one-time move reads only ordered problems on a separate Core copy.""" -import json -import shutil -import sqlite3 - -from cutover.issue_rts_orders import carry -from agent_comms.recorded_time import RecordedEventTime -from agent_comms.messages import Message, InfoKind -from agent_comms.todos import Assignment, NoGoalReference, OpenTodoState, Todo, TodoStore -from test_commander_problem_orders import order - - -def test_open_only_move_on_copied_root(commander, tmp_path, capsys): - comms, _ = commander - actor = comms.registry.new_thread('claude-code', frozenset(), str(tmp_path), created_at=3.0) - comms.registry.declare(actor) - worker = comms.registry.require('worker') - assignment = Assignment(worker.name, worker.created_at, actor.name, actor.created_at, - worker.goal.require().id) - row = Todo(id='P1', repo='OpenHCSDev/agent-comms', text='Repair the bus reader', - creator=actor.name, creator_created=actor.created_at, state=OpenTodoState, - revision=1, goal=NoGoalReference(), assignment=assignment, - order=order(tmp_path, 'old-message'), deadline=RecordedEventTime(1000)) - with TodoStore(comms.root).writing() as db: - row.upsert(db) - copied = tmp_path / 'copy' - copied.mkdir() - shutil.copy2(comms.root / 'registry.json', copied / 'registry.json') - with sqlite3.connect(comms.root / 'todos.sqlite3') as source, sqlite3.connect(copied / 'todos.sqlite3') as target: - source.backup(target) - message = Message('claude-code', 'worker', f'''Problem: P1 -Owner: {__file__}:1 -1. Repair the existing bus reader -Banned: No second store -Report: Report the result -''', InfoKind, seq=1) - (copied / 'order-messages.jsonl').write_text(json.dumps(message.to_wire()) + '\n') - rts = tmp_path / 'rts' - rts.mkdir() - (rts / 'deadlines.json').write_text(json.dumps({'worker': [{ - 'due': '00:30', 'msg': message.message_id, 'at': 100, 'fact': 'P1', 'facts': ['P1'], - 'check': [str(tmp_path), 'printf 1', 1], 'disk': False}]})) - (rts / 'problems.json').write_text(json.dumps({ - 'P1': {'text': row.text, 'where': 'reader.py:1', 'at': 90}, - 'UNORDERED': {'not_a_problem_record': True}})) - (rts / 'strikes.jsonl').write_text('\n'.join(json.dumps(record) for record in ( - {'unit': 'worker', 'why': 'Archived old strike', 'at': 50, 'task': None}, - {'unit': 'worker', 'why': 'The unit changed the wrong file and ignored the required edit', - 'at': 120, 'task': 'P1'})) + '\n') - assert carry(rts, copied, True) == 0 - report = json.loads(capsys.readouterr().out) - assert report['orders_issued'] == {'create': 0, 'assign': 1} - assert report['unordered_problems'] == report['strikes_left_as_history'] == 1 - assert report['strikes_attached'] == 1 - assert report['blockers'] == {} - assert TodoStore(comms.root).get('P1') == row - issued = TodoStore(copied).get('P1') - assert issued.order.id == message.message_id - assert issued.strikes[0].previous == issued.held() diff --git a/tests/test_todos.py b/tests/test_todos.py index 13c9d61363..3199ac231f 100644 --- a/tests/test_todos.py +++ b/tests/test_todos.py @@ -2,6 +2,7 @@ from __future__ import annotations +import json from concurrent.futures import ProcessPoolExecutor from contextlib import closing from dataclasses import replace @@ -18,7 +19,8 @@ from agent_comms.registration import Registration from agent_comms.threads import Thread from agent_comms.todos import ( - AssignTodo, Assignment, ChangeTodoState, DoneTodoState, GoalRef, MechanicalOrderRun, PendingProblemOrder, + AssignTodo, Assignment, ChangeTodoState, DoneTodoState, GoalRef, + MechanicalOrderRun, PendingProblemOrder, BlockedTodoState, OperationDelivery, Todo, TodoConflict, TodoError, TodoState, TodoStore, ) @@ -99,7 +101,8 @@ def test_commander_reads_decode_only_open_todos(tmp_path, monkeypatch): assignment=None, order=delivered).insert(db) with closing(CoordinationStore.writer(store.path)) as db: indexes = tuple(name for name, in db.execute( - "SELECT name FROM sqlite_master WHERE type='index' AND name NOT LIKE 'sqlite_%'")) + "SELECT name FROM sqlite_master WHERE type='index' AND tbl_name=? " + "AND name NOT LIKE 'sqlite_%'", (Todo.declared_name,))) for name in indexes: db.execute(f'DROP INDEX "{name}"') db.commit() @@ -179,3 +182,18 @@ def test_validation_rejects_ambiguous_identity_and_bad_state(tmp_path: Path) -> ) with pytest.raises(ValueError): FieldCodec.decode(type[TodoState], "unverified") + + +def test_merged_order_keeps_its_checked_fields(tmp_path: Path) -> None: + store = TodoStore(tmp_path) + lead = thread("lead", tmp_path, 1.0) + todo = store.create( + "todo-1", "OpenHCSDev/agent-comms", "Fix the sidebar", creator=lead, goal=GoalRef("lead", 1.0, "goal-1"), + owner=lead, generation="initial", deadline=RecordedEventTime(1000.0), order=order(tmp_path)) + merged = replace(todo, order=todo.order.checked().merged("2b9c1f1ab")) + cells = merged.order.board_cells(merged) + assert cells.startswith(f"{__file__}:1 | lead | checked-order | {RecordedEventTime(1000.0).clock()} | ") + assert json.loads(cells.split(" | ", 4)[-1]) == FieldCodec.encode(merged.order.checked().check) + assert merged.order.checked().copy_as(PendingProblemOrder) == todo.order + assert merged.order.line == "Merged and installed" + assert merged.order.waiting_on_install(merged, {}) == () diff --git a/tests/test_tool_replies.py b/tests/test_tool_replies.py index 9bf51e31c9..d6814b664f 100644 --- a/tests/test_tool_replies.py +++ b/tests/test_tool_replies.py @@ -40,7 +40,7 @@ def scenario(comms: Comms, project, monkeypatch): comms.registry.new_thread(name, frozenset(), str(project), created_at=float(created))) # Commander-specific rules read this bus's record, never the machine's installed config. from agent_comms.commander.config import CommanderConfig - CommanderConfig(commander="lead", root=comms.root, cli=project / "agent-comms", + CommanderConfig(commander="lead", root=comms.root, state=comms.root / "state", board=comms.root / "board.md", narrative=comms.root / "narrative.md", skill=comms.root / "skill.md", rules=(), repos=(), worktrees=comms.root / "wt", scratch=comms.root / "scratch", diff --git a/tools/carry_rts_orders.py b/tools/carry_rts_orders.py new file mode 100644 index 0000000000..bdf14a4b5f --- /dev/null +++ b/tools/carry_rts_orders.py @@ -0,0 +1,234 @@ +from __future__ import annotations + +import argparse +import json +import re +import sys +from abc import abstractmethod +from dataclasses import dataclass, replace +from pathlib import Path +from typing import Annotated, cast + +from agent_comms import todos +from agent_comms.change_report import FileLine +from agent_comms.commander.config import Repo +from agent_comms.comms import Comms +from agent_comms.field_codec import ( + FieldCodec, + FieldRepresentation, + NullMember, + SingleFieldValue, + SubclassUnion, + TextRepresentation, + boundary_decoder, +) +from agent_comms.goals import OrderCheck +from agent_comms.recorded_time import RecordedEventTime +from agent_comms.thread_identity import ThreadIncarnation + +RETIRED = ("deadlines.json", "strikes.jsonl", "sent.jsonl", "hunts.jsonl") +SOURCES = ("problems.json", *RETIRED) +TOOLS = ("order.py", "judge.py", "rotate.py", "tool_watch.py", "shift_after_window.py") + + +@dataclass(frozen=True) +class RtsProblem: + at: float + text: str + where: FileLine + held_by: str = "" + + +@dataclass(frozen=True) +class RtsSent: + at: float + id: str + text: str + to: tuple[str, ...] + + +@dataclass(frozen=True) +class RtsStrike: + at: float + task: str | NullMember + unit: str + why: str + caught: str = "" + + def recorded(self, actor: ThreadIncarnation, owner: ThreadIncarnation) -> todos.StrikeTodo: + previous = todos.Assignment.of(owner, actor, f"rts-strike:{self.task}:{self.at}") + return todos.StrikeTodo(1, actor, previous, self.why, RecordedEventTime(self.at)) + + +class OrderMerge(SubclassUnion, affix=""): + @abstractmethod + def wrap( + self, order: todos.PendingProblemOrder + ) -> todos.PendingProblemOrder | todos.MergedProblemOrder: ... + + +@dataclass(frozen=True) +class NotMerged(NullMember, OrderMerge): + def wrap(self, order: todos.PendingProblemOrder) -> todos.PendingProblemOrder: + return order + + +@dataclass(frozen=True) +class MergedAt(SingleFieldValue, OrderMerge): + sha: str + + def wrap(self, order: todos.PendingProblemOrder) -> todos.MergedProblemOrder: + return order.merged(self.sha) + + +class MechanicalRunJson(FieldRepresentation): + accepts_null = True + + @classmethod + def decode(cls, value: object) -> todos.MechanicalOrderRun: + FieldCodec.decode(type(None), value) + return todos.MechanicalOrderRun() + + @classmethod + def encode(cls, value: object) -> None: + return None + + @classmethod + def schema(cls) -> dict[str, object]: + return FieldCodec.value_schema(type(None)) + + +class ApplicationRunText(TextRepresentation[todos.ApplicationOrderRun]): + @classmethod + def from_text(cls, value: str) -> todos.ApplicationOrderRun: + return todos.ApplicationOrderRun(value) + + @classmethod + def encode(cls, value: object) -> str: + return cast(todos.ApplicationOrderRun, value).text + + +@dataclass(frozen=True) +class RtsOrder: + due: str + msg: str + at: float + fact: str + facts: tuple[str, ...] + check: tuple[str, str, int] + disk: bool + works: ( + Annotated[todos.MechanicalOrderRun, MechanicalRunJson] + | Annotated[todos.ApplicationOrderRun, ApplicationRunText] + ) + started: bool = False + merged: OrderMerge = NotMerged() + no_first_call: bool = False + + @boundary_decoder + def declared(self, body: str) -> todos.PendingProblemOrder: + place = re.search(r"^Owner:\s*(\S+):(\d+)\s*$", body, re.M) + report = re.search(r"\bReport:\s*([^\n]+)", body) + if place is None or report is None or not re.search(r"^Banned:", body, re.M): + raise ValueError(f"{self.facts} {self.msg}: the order needs Owner, Banned and Report") + directory, command, expected = self.check + check = OrderCheck.parse(f"{directory} :: {command} -> {expected}") + if check is None: + raise ValueError(f"{self.facts}: the order needs a check") + return todos.PendingProblemOrder( + id=self.msg, + owner=FileLine(place[1], int(place[2])), + check=check, + edits=tuple(re.findall(r"^\s*\d+[.)] (.+?)\s*$", body, re.M)), + banned=tuple( + part.strip() + for line in re.findall(r"^Banned:(.*)$", body, re.M) + for part in line.split(";") + if part.strip() + ), + report=report[1], + at=RecordedEventTime(self.at), + first_call_deadline=RecordedEventTime(self.at + 120), + first_call=todos.ObservedFirstCall() if self.started else todos.AwaitingFirstCall(), + run=self.works, + ) + + +@boundary_decoder +def rts_hunt(line: bytes) -> todos.Hunt: + values = json.loads(line) + sent = values.pop("sent") + values.update( + at=FieldCodec.encode(RecordedEventTime(values.pop("at"))), + kind="clean" if sent is None else "ordered", + ) + return FieldCodec.decode(todos.Hunt, values if sent is None else {**values, "sent": sent}) + + +def lines[T](data: bytes, kind: type[T]) -> list[T]: + return [FieldCodec.decode(kind, json.loads(line)) for line in data.splitlines() if line] + + +def carry(rts: Path, root: Path, apply: bool, retire: bool) -> None: + source = {rts / name: (rts / name).read_bytes() for name in SOURCES} + bodies = {sent.id: sent.text for sent in lines(source[rts / "sent.jsonl"], RtsSent)} + orders = [ + (unit, row, row.declared(bodies[row.msg])) + for unit, rows in FieldCodec.decode( + dict[str, tuple[RtsOrder, ...]], json.loads(source[rts / "deadlines.json"]) + ).items() + for row in rows + ] + selected = {pid: (unit, row, order) for unit, row, order in orders for pid in row.facts} + if len(selected) != sum(len(row.facts) for _, row, _ in orders): + raise ValueError("A problem has two open RTS orders; split it first") + problems = FieldCodec.decode(dict[str, RtsProblem], json.loads(source[rts / "problems.json"])) + problems = {pid: problems[pid] for pid in selected} + strikes = lines(source[rts / "strikes.jsonl"], RtsStrike) + if apply: + comms = Comms(root) + actor = comms.registry.require("claude-code").incarnation + with todos.TodoStore(root).writing() as db: + existing = {todo.id: todo for todo in todos.Todo.stream(db)} + for pid, (unit, row, order) in selected.items(): + owner = comms.registry.require(unit).incarnation + proposed = todos.Assignment.of(owner, actor, "rts-order:" + row.msg) + due = RecordedEventTime(row.at).at_clock(row.due) + if pid in existing: + todo = existing[pid] + todo.require_commander(actor) + todo.state.require_open(todo) + else: + print(f"{pid} order {row.msg}", file=sys.stderr, flush=True) + repo = Repo.of(Path(order.owner.file).parent) + text = problems[pid].text + todo = todos.Todo.ordered(pid, repo.github, text, actor, proposed, due, order) + todo = todos.AssignTodo( + todo.revision, proposed, due, row.merged.wrap(order), todo.goal + ).apply(todo) + history = tuple( + strike.recorded(actor, comms.registry.require(strike.unit).incarnation) + for strike in strikes + if strike.task == pid + ) + todo = replace(todo, strikes=history, revision=todo.revision + 1) + if not todo.sent_in(row.msg): + todo = todos.OrderMessage(todo.revision, actor, row.msg).apply(todo) + todo.upsert(db) + for index, line in enumerate(source[rts / "hunts.jsonl"].splitlines()): + todos.HuntRecord(f"rts-hunt:{index}", actor, rts_hunt(line)).upsert(db) + if any(path.read_bytes() != data for path, data in source.items()): + raise ValueError(f"RTS changed during the import; {root} is unchanged") + if retire: + for name in TOOLS + RETIRED: + (rts / name).rename(rts / f"{name}.retired-p707") + print(json.dumps({"orders": len(selected), "strikes": len(strikes)}, indent=2)) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--rts-dir", dest="rts", required=True, type=Path) + parser.add_argument("--root", required=True, type=Path) + parser.add_argument("--apply", action="store_true") + parser.add_argument("--retire-rts", dest="retire", action="store_true") + carry(**vars(parser.parse_args())) diff --git a/tools/cutover/issue_rts_orders.py b/tools/cutover/issue_rts_orders.py deleted file mode 100644 index dca44f72d7..0000000000 --- a/tools/cutover/issue_rts_orders.py +++ /dev/null @@ -1,179 +0,0 @@ -"""Temporary move of open RTS orders into a copied Core root. - -The root contains registry/Todo copies and order-messages.jsonl: exact original -bus rows selected by the RTS message ids. Never run against the live root. -Core's order actions choose assignments and goal generations; this tool does -not write Todo rows or infer generations from old goals. -""" -from __future__ import annotations - -import argparse -import json -import re -import time -from dataclasses import dataclass -from pathlib import Path - -from agent_comms.change_report import FileLine -from agent_comms.repo import Repo -from agent_comms.comms import Comms -from agent_comms.field_codec import FieldCodec -from agent_comms.goal_actions import CreateProblemGoalAction, OrderGoalAction, SetGoalAction -from agent_comms.goals import OrderCheck, RequestGoalOrder -from agent_comms.messages import Message -from agent_comms.recorded_time import RecordedEventTime -from agent_comms.todos import (ApplicationOrderRun, AwaitingFirstCall, MechanicalOrderRun, - ObservedFirstCall, PendingProblemOrder, StrikeTodo, TodoStore) - - -@dataclass(frozen=True) -class RtsOrder: - due: str - msg: str - at: float - fact: str - facts: tuple[str, ...] - check: tuple[str, str, int] - disk: bool - works: str | None = None - started: bool = False - no_first_call: bool = False - - def deadline(self) -> RecordedEventTime: - hour, minute = map(int, self.due.split(':')) - sent = time.localtime(self.at) - at = time.mktime((sent.tm_year, sent.tm_mon, sent.tm_mday, - hour, minute, 0, 0, 0, -1)) - return RecordedEventTime(at + 86400 if at < self.at - 60 else at) - - def order(self, message: Message) -> PendingProblemOrder: - body = message.body - place = re.search(r'^Owner:\s*(\S+):(\d+)\s*$', body, re.M) - report = re.search(r'^Report:\s*(.+)$', body, re.M) - if place is None or report is None: - raise ValueError(f'{self.msg}: original order needs Owner and Report lines') - return PendingProblemOrder( - id=self.msg, owner=FileLine(place[1], int(place[2])), - check=OrderCheck.parse(f'{self.check[0]} :: {self.check[1]} -> {self.check[2]}'), - edits=tuple(re.findall(r'^\s*\d+[.)] (.+?)\s*$', body, re.M)), - banned=tuple(part.strip() for line in re.findall(r'^Banned:(.*)$', body, re.M) - for part in line.split(';') if part.strip()), - report=report[1], at=RecordedEventTime(self.at), - first_call_deadline=RecordedEventTime(self.at + 120), - first_call=ObservedFirstCall() if self.started else AwaitingFirstCall(), - run=ApplicationOrderRun(self.works) if self.works else MechanicalOrderRun()) - - -@dataclass(frozen=True) -class RtsProblem: - text: str - where: str - at: float - held_by: str | None = None - - -@dataclass(frozen=True) -class RtsStrike: - unit: str - why: str - at: float - task: str | None - caught: str | None = None - - -def read_json(path: Path, kind): - return FieldCodec.decode(kind, FieldCodec.parse(path.read_text())) - - -def read_lines(path: Path, kind): - text = path.read_text() - if text and not text.endswith('\n'): - raise ValueError(f'{path}: incomplete final row') - return tuple(FieldCodec.decode(kind, FieldCodec.parse(line)) - for line in text.splitlines() if line.strip()) - - -def carry(rts: Path, root: Path, apply: bool) -> int: - deadlines = read_json(rts / 'deadlines.json', dict[str, tuple[RtsOrder, ...]]) - records = FieldCodec.parse((rts / 'problems.json').read_text()) - strikes = read_lines(rts / 'strikes.jsonl', RtsStrike) - messages = {} - for line in (root / 'order-messages.jsonl').read_text().splitlines(): - message = Message.from_committed_wire(FieldCodec.parse(line)) - messages[message.message_id] = message - ordered = {} - blockers = {} - for unit, orders in deadlines.items(): - for old in orders: - for pid in old.facts: - if pid in ordered: - blockers[pid] = 'two open orders name this problem' - ordered[pid] = (unit, old) - selected = {pid: FieldCodec.decode(RtsProblem, records[pid]) for pid in ordered} - parsed_orders = {old.msg: old.order(messages[old.msg]) - for orders in deadlines.values() for old in orders} - attached = tuple(strike for strike in strikes - if strike.task in ordered and strike.unit == ordered[strike.task][0]) - report = {'orders_issued': {'create': 0, 'assign': 0}, 'strikes_attached': 0, - 'strikes_left_as_history': len(strikes) - len(attached), - 'unordered_problems': len(records) - len(selected), - 'source_orders': sum(map(len, deadlines.values())), - 'source_ordered_problems': len(ordered), - 'source_first_call_exemptions': [old.msg for orders in deadlines.values() - for old in orders if old.no_first_call], - 'blockers': blockers} - if blockers or not apply: - print(json.dumps(report, indent=2)) - return 2 if blockers else 0 - comms = Comms(root) - actor = comms.registry.require('claude-code') - requests = tuple(goal.id for goal in actor.goal.active() - if isinstance(goal.order, RequestGoalOrder)) - if not requests: - comms.goals.update_goal(actor.name, SetGoalAction( - text='Move open RTS orders into this copied Core root', order=RequestGoalOrder())) - request = comms.registry.require(actor.name).goal.require().id - store = TodoStore(root) - existing = {row.id: row for row in store.history_for_commander(actor.incarnation)} - for pid, (unit, old) in ordered.items(): - try: - message = messages[old.msg] - if message.sender != actor.name: - raise ValueError(f'{old.msg}: order was not sent by the commander') - order = parsed_orders[old.msg] - common = dict(problem=pid, order=order, request=request, - deadline=old.deadline(), commander=actor.name) - if pid in existing: - action = OrderGoalAction(**common, expected_revision=store.get(pid).revision) - kind = 'assign' - else: - action = CreateProblemGoalAction(**common, - repo=Repo.of(Path(order.owner.file).parent).github, text=selected[pid].text) - kind = 'create' - comms.goals.update_goal(unit, action) - report['orders_issued'][kind] += 1 - for old_strike in attached: - if old_strike.task != pid: - continue - current = store.get(pid) - comms.goals.change_problem(pid, StrikeTodo( - current.revision, actor.incarnation, current.held(), old_strike.why, - RecordedEventTime(old_strike.at)), actor.name) - report['strikes_attached'] += 1 - except (ValueError, KeyError) as error: - blockers[pid] = str(error) - print(json.dumps(report, indent=2)) - return 2 if blockers else 0 - - -def main() -> int: - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument('--rts-dir', required=True, type=Path) - parser.add_argument('--root', required=True, type=Path) - parser.add_argument('--apply', action='store_true') - args = parser.parse_args() - return carry(args.rts_dir, args.root, args.apply) - - -if __name__ == '__main__': - raise SystemExit(main())