diff --git a/src/agent_comms/channel_coding_tools.py b/src/agent_comms/channel_coding_tools.py index b9569945ca..56658f06ba 100644 --- a/src/agent_comms/channel_coding_tools.py +++ b/src/agent_comms/channel_coding_tools.py @@ -69,7 +69,7 @@ def goal_permission(self) -> TurnGoalPermission: ... async def reserve_room(self, call: CodingCall) -> RoomReservation: registry = self.comms.registry thread = self.thread - projection = StorageProjection(registry.storage_rule, thread) + projection = StorageProjection(registry.storage_rule, thread, self.comms) effects = await asyncio.to_thread(projection.check_call, call) work = tuple(dict.fromkeys(kind for effect in effects for kind in effect.work)) if work: diff --git a/src/agent_comms/cli_commands.py b/src/agent_comms/cli_commands.py index 00ebfcd2ce..7f067c70e8 100644 --- a/src/agent_comms/cli_commands.py +++ b/src/agent_comms/cli_commands.py @@ -506,7 +506,7 @@ def apply(cls, command: PreparedCheckoutCliCommand, ctx: Comms, thread: Thread, grant = command.grant(ctx, thread) grant.require() repository = command.source(ctx, grant) - projection = StorageProjection(ctx.registry.storage_rule, thread) + projection = StorageProjection(ctx.registry.storage_rule, thread, ctx) if destination.exists(): CheckoutStorageEffect(destination, repository).check(projection) Checkout(destination).prepare() @@ -527,7 +527,7 @@ class RemoveCheckoutOperation(CheckoutOperation): def apply(cls, command: PreparedCheckoutCliCommand, ctx: Comms, thread: Thread, destination: Path) -> None: RemoveCheckoutStorageEffect(destination).check( - StorageProjection(ctx.registry.storage_rule, thread)) + StorageProjection(ctx.registry.storage_rule, thread, ctx)) Checkout(destination).retire() diff --git a/src/agent_comms/commander_frame_file.py b/src/agent_comms/commander_frame_file.py index ba4ed6a680..ac855c5a91 100644 --- a/src/agent_comms/commander_frame_file.py +++ b/src/agent_comms/commander_frame_file.py @@ -1,5 +1,8 @@ from __future__ import annotations +from pathlib import Path + +from agent_comms.service_passes import ServicePassStore from agent_comms.locked_store import PatternFile, RuntimeFile @@ -12,3 +15,8 @@ class CommanderFrameFile(RuntimeFile, PatternFile): __slots__ = () filename = "commander_frames/{lookup}.json" + + @classmethod + def source_paths(cls, root: Path) -> tuple[Path, ...]: + """Frame readers hydrate service health too; both files wake their open view.""" + return (*sorted(cls.matching_in(root)), ServicePassStore.path_in(root)) diff --git a/src/agent_comms/commander_hud.py b/src/agent_comms/commander_hud.py index 87c53567b2..77eda11310 100644 --- a/src/agent_comms/commander_hud.py +++ b/src/agent_comms/commander_hud.py @@ -5,7 +5,7 @@ import asyncio import time from abc import ABC, abstractmethod -from dataclasses import dataclass, field, replace +from dataclasses import dataclass, field from pathlib import Path from typing import TYPE_CHECKING, Annotated @@ -18,15 +18,14 @@ from agent_comms.commander.vibe_check import OverloadDegradation from agent_comms.commander_frame_file import CommanderFrameFile from agent_comms.commander_service import CommanderService -from agent_comms.coordinator import Coordination from agent_comms.errors import ToolRefusedError -from agent_comms.field_codec import DisclosureIdentity, OrderId, WireTransient, projected +from agent_comms.field_codec import DisclosureIdentity, OrderId, projected from agent_comms.heartbeat_frame import HeartbeatFrame from agent_comms.locked_store import JsonDocument from agent_comms.plan_phases import Plan, PlanPhase from agent_comms.order_story import OrderStory -from agent_comms.recorded_time import EventTime, RecordedEventTime, UnrecordedEventTime -from agent_comms.service_passes import ServicePassStore +from agent_comms.recorded_time import RecordedEventTime +from agent_comms.service_passes import BoardUnavailable, ServicePassStore, ServiceReading from agent_comms.thread_identity import ThreadIncarnation from agent_comms.todos import OrderProgress, ProgressPoint, ProgressRead, ProgressSource, PullRequest from agent_comms.tool_result import ToolResult @@ -256,7 +255,23 @@ def line(self) -> str: @dataclass(frozen=True) -class CommanderHud(ToolResult): +class CommanderLoad(ServiceReading): + """Retained load with the same recorded service evidence as its published frame.""" + + load: OverloadDegradation + + def retained_status(self) -> str: + if self.services.heartbeat_current(self.heartbeat): + return 'Retained board rows and load:' + return 'STALE retained board rows and load:' + + @projected(view='wire') + def header(self) -> tuple[str, ...]: + return (*self.service_status, self.retained_status(), str(self.load)) + + +@dataclass(frozen=True) +class CommanderHud(CommanderLoad): """One disposable view of native records, shared by comms_hud and the commander board. TodoStore owns orders, review timers and strikes; OrderEditCountRead owns edit counts; @@ -268,14 +283,9 @@ class CommanderHud(ToolResult): commander: ThreadIncarnation units: tuple[CommanderUnit, ...] phases: tuple[CommanderPhase, ...] - load: OverloadDegradation forks: tuple[ThreadIncarnation, ...] usage: CodexUsage board: tuple[BoardEntry, ...] = () - heartbeat: Annotated[EventTime, WireTransient()] = field( - default_factory=UnrecordedEventTime, compare=False, kw_only=True, - ) - """Read-time projection of ServicePassStore's clock; never written into the frame.""" @classmethod def read(cls, comms: Comms, commander: str) -> CommanderHud: @@ -327,9 +337,9 @@ def of(cls, comms: Comms, commander: str, frame: HeartbeatFrame, if snapshot.status(fork.name).receiving) board = tuple(entry for packet in CommanderUpdate.latest(comms, owner) for entry in packet.board(comms, commander)) - return cls(owner, tuple(units), phases, - OverloadDegradation.from_orders(held, len(timers), sum(edits.values())), - forks, CodexUsage.read(), board) + return cls(commander=owner, units=tuple(units), phases=phases, + load=OverloadDegradation.from_orders(held, len(timers), sum(edits.values())), + forks=forks, usage=CodexUsage.read(), board=board) def rows(self) -> tuple[HudRow, ...]: """Committed board order, actionable cards, live phases, then folded quiet units.""" @@ -370,8 +380,10 @@ def brief(self) -> tuple[str, ...]: @projected(view="wire") def frame(self) -> tuple[str, ...]: """The load line, then one line per row in the order rows() declares.""" - return (str(self.load), *self.heartbeat.aged_line("Tristan", time.time()), - *(row.line() for row in self.rows())) + return (*self.header, *(row.line() for row in self.rows())) + + def load_reading(self) -> CommanderLoad: + return CommanderLoad(self.load, services=self.services, heartbeat=self.heartbeat) @@ -386,26 +398,23 @@ class CommanderFrame(CommanderFrameFile, JsonDocument[CommanderHud]): """ path: Path + reading: ServiceReading = field(default_factory=ServiceReading, compare=False, kw_only=True) record_type = CommanderHud def empty(self) -> CommanderHud: - raise ToolRefusedError(f"Tristan frame missing: {self.path}") + raise BoardUnavailable(self.reading, f'Tristan frame missing: {self.path}') @classmethod def current(cls, comms: Comms, commander: str) -> CommanderHud: """Read the heartbeat's publication, never repeat its forge and history work.""" owner = comms.registry.require(commander).incarnation.require_recorded() passes = ServicePassStore.at(comms.root).read() - completed = passes.heartbeat_completed_at(owner) - if not completed.since(time.time() - 3 * Coordination.watch_period): - raise ToolRefusedError( - f"Tristan stale for {owner.name} since {completed.clock()}: {passes.stall()}" - ) - hud = cls(cls.path_in(comms.root, lookup=str(owner.lookup))).read() + health = ServiceReading().with_services(passes, owner) + hud = cls(cls.path_in(comms.root, lookup=str(owner.lookup)), reading=health).read() if hud.commander != owner: - raise ToolRefusedError(f"Tristan frame belongs to another incarnation of {owner.name}") - return replace(hud, heartbeat=completed) + raise BoardUnavailable(health, f'Tristan frame belongs to another incarnation of {owner.name}') + return hud.with_services(passes, owner) @classmethod def replace_all(cls, root: Path, huds: tuple[CommanderHud, ...]) -> None: @@ -422,7 +431,10 @@ def replace_all(cls, root: Path, huds: tuple[CommanderHud, ...]) -> None: @classmethod def read_all(cls, root: Path) -> tuple[CommanderHud, ...]: passes = ServicePassStore.at(root).read() - return tuple(sorted((replace(hud, heartbeat=passes.heartbeat_completed_at( - hud.commander.require_recorded())) - for path in cls.matching_in(root) for hud in (cls(path).read(),)), + health = ServiceReading(services=passes) + huds = tuple(sorted((hud.with_services(passes, hud.commander.require_recorded()) + for path in cls.matching_in(root) for hud in (cls(path, reading=health).read(),)), key=lambda hud: hud.commander.name)) + if not huds and passes.failing: + raise BoardUnavailable(health, 'no published commander frame.') + return huds diff --git a/src/agent_comms/heartbeat_frame.py b/src/agent_comms/heartbeat_frame.py index 551982e768..d8cc735265 100644 --- a/src/agent_comms/heartbeat_frame.py +++ b/src/agent_comms/heartbeat_frame.py @@ -13,7 +13,7 @@ from agent_comms.repo import Repo from agent_comms.standing_rules import OwedAnswer, StandingRule from agent_comms.thread_identity import ThreadIncarnation -from agent_comms.turn_digests import EditWorkplaces +from agent_comms.storage_rules import EditWorkplaces from agent_comms.todos import ( ClosedHeadReading, HeadReviewRecord, ProgressRead, StoredClosedHead, StoredHeadReview, ) @@ -76,5 +76,5 @@ def capture(cls, comms: Comms, commander: str, inputs: CommanderInputs, OwedAnswer.read_scan(comms, commander, inputs), closed, reviews, - EditWorkplaces.capture(comms, snapshot, inputs.live, plan.todos, inputs.pulls), + EditWorkplaces.capture(comms, snapshot, inputs.live, inputs.pulls), ) diff --git a/src/agent_comms/order_tools.py b/src/agent_comms/order_tools.py index e3d4a4fa1c..ec2894b064 100644 --- a/src/agent_comms/order_tools.py +++ b/src/agent_comms/order_tools.py @@ -33,11 +33,14 @@ JudgedCall, ToolCallVerdict, ) -from .commander.vibe_check import OverloadDegradation -from .commander_hud import CommanderHud +from .commander_hud import CommanderHud, CommanderLoad +from .service_passes import ServicePassStore from .comms import Comms from .errors import ToolRefusedError from .field_codec import FieldCodec, MessageId, OrderId, RecordProblems, WireName +from .problem_number import ProblemNumber +from .tool_declaration import EditCall +from .work_branches import WorkBranchError from .goal_actions import CreateProblemGoalAction, OrderGoalAction from .goals import Goal, OrderCheck from .head_review import HeadReviewOrder, PreparedHeadReview, PullRequestHead @@ -69,6 +72,7 @@ TurnView, TurnsSince, UnjudgedEdits, + UnavailableOrderEdits, ) @@ -228,8 +232,17 @@ class CommsUnjudgedTool(ThreadReadTool): description = "Read original edit calls since each open order, excluding that order's recorded verdicts." def apply(self, comms: Comms, caller: ToolCaller) -> UnjudgedEdits: - return UnjudgedEdits(tuple(edit for read in OrderEditsRead.for_commander( - comms, caller.subject(self.thread)) for edit in read.read())) + owner = comms.registry.require(caller.subject(self.thread)).incarnation.require_recorded() + passes = ServicePassStore.at(comms.root).read() + edits: list[EditCall] = [] + unavailable: list[UnavailableOrderEdits] = [] + for read in OrderEditsRead.for_commander(comms, owner.name): + try: + edits.extend(read.read()) + except (WorkBranchError, OSError) as error: + unavailable.append(UnavailableOrderEdits( + ProblemNumber.parse(read.todo.id), read.thread.incarnation, error)) + return UnjudgedEdits(tuple(edits), tuple(unavailable)).with_services(passes, owner) @dataclass(frozen=True, kw_only=True) @@ -244,10 +257,10 @@ def reply(self, read: dict[str, JsonValue], caller: ToolCaller) -> dict[str, Jso @dataclass(frozen=True, kw_only=True) class CommsCommanderLoadTool(BoardReadTool): label = "Measured commander load" - description = "Read open orders, review timers and unjudged edits used by the overload rule." + description = "Read retained commander load with recorded service failures and its original completion clock." - def apply(self, comms: Comms, caller: ToolCaller) -> OverloadDegradation: - return CommanderHud.read(comms, caller.subject(self.thread)).load + def apply(self, comms: Comms, caller: ToolCaller) -> CommanderLoad: + return CommanderHud.read(comms, caller.subject(self.thread)).load_reading() @dataclass(frozen=True, kw_only=True) @@ -255,7 +268,8 @@ class CommsHudTool(ThreadReadTool): label = "Commander HUD" description = ( "Read the commander's latest published Tristan frame: orders, review timers, " - "strikes, edits, PRs, plan phases and load. A missing or stale Tristan frame is refused." + "strikes, edits, PRs, plan phases and load. Recorded failures precede retained data; " + "stale rows stay explicitly stale, and missing frames remain unavailable." ) def apply(self, comms: Comms, caller: ToolCaller) -> CommanderHud: @@ -325,7 +339,8 @@ class CommsStandingRulesTool(CommsHudTool): label = "Read broken standing rules" description = ( "Read the published Tristan board and HUD. Tristan owns " - "enforcement; this read never repeats it. A missing or stale Tristan frame is refused." + "enforcement; this read never repeats it. Recorded failures precede stale retained rows; " + "a missing frame remains explicitly unavailable." ) diff --git a/src/agent_comms/presentation.py b/src/agent_comms/presentation.py index 3a21c7715a..90613f2fca 100644 --- a/src/agent_comms/presentation.py +++ b/src/agent_comms/presentation.py @@ -895,7 +895,7 @@ def external_transcript_paths(views: HistoryViews) -> tuple[Path, ...]: def commander_frame_paths(views: HistoryViews) -> tuple[Path, ...]: - return tuple(sorted(CommanderFrameFile.matching_in(views.root))) + return CommanderFrameFile.source_paths(views.root) def expiry_period_changed(before: int, now: int) -> bool: @@ -990,7 +990,7 @@ class WireRevision(ReadState): # Activity-derived views change once per expiry period, not per tick. expiry_tick: Annotated[ int, - WatchViews((ActivityView,)), + WatchViews((ActivityView, CommanderFramesView)), Differs(expiry_period_changed), ] diff --git a/src/agent_comms/service_passes.py b/src/agent_comms/service_passes.py index b34d2dc30b..2270061e94 100644 --- a/src/agent_comms/service_passes.py +++ b/src/agent_comms/service_passes.py @@ -5,9 +5,13 @@ import time from collections.abc import Iterable from dataclasses import dataclass, field, replace -from typing import ClassVar +from typing import Annotated, ClassVar, Self +from .coordinator import Coordination from .locked_store import DocumentChange, DurableFile, LockedStore +from .errors import ToolRefusedError +from .field_codec import WireTransient, projected +from .tool_result import ToolResult from .recorded_time import EventTime, RecordedEventTime, UnrecordedEventTime from .store_files import DependentStateTier, SingleFileStore from .thread_identity import RecordedThreadIncarnation, ThreadIncarnation @@ -20,6 +24,12 @@ class FailingPass: since: float error: str + def failure_line(self, watch: str) -> str: + return (f'FIX service pass {watch} failing since {RecordedEventTime(self.since).clock()}' + f' ({max(0, int(time.time() - self.since))}s elapsed):' + f' {self.error}. is anyone fixing it, or letting it pile up?' + ' -> fix the owner of the failing read this wake') + @dataclass(frozen=True) class ServicePasses: @@ -27,25 +37,58 @@ class ServicePasses: commander_completed: dict[RecordedThreadIncarnation, float] = field(default_factory=dict) """Last observation completed for each commander, including quiet observations.""" + def failure_lines(self) -> tuple[str, ...]: + return tuple(failure.failure_line(watch) for watch, failure in sorted(self.failing.items())) + def heartbeat_completed_at(self, commander: RecordedThreadIncarnation) -> EventTime: """The original completion clock, or its declared unrecorded state.""" if commander in self.commander_completed: return RecordedEventTime(self.commander_completed[commander]) return UnrecordedEventTime() - def stall(self) -> str: - """Why passes stopped completing: every recorded failure with its error, or, with none - recorded, that no watch is running at all.""" - return "; ".join( - f"{watch} failing since {RecordedEventTime(failure.since).clock()}: {failure.error}" - for watch, failure in self.failing.items() - ) or "no failure recorded, so no heartbeat is running" + def heartbeat_current(self, heartbeat: EventTime) -> bool: + """Only the commander's completion clock determines heartbeat freshness.""" + return heartbeat.since(time.time() - 3 * Coordination.watch_period) def completed_since(self, commander: RecordedThreadIncarnation, since: float) -> bool: """A pass for this commander completed at or after ``since``.""" return self.heartbeat_completed_at(commander).since(since) +@dataclass(frozen=True) +class ServiceReading(ToolResult): + """Read-time service evidence shared by board, load and independent edit replies.""" + + services: Annotated[ServicePasses, WireTransient()] = field( + default_factory=ServicePasses, compare=False, kw_only=True) + heartbeat: Annotated[EventTime, WireTransient()] = field( + default_factory=UnrecordedEventTime, compare=False, kw_only=True) + + def with_services(self, passes: ServicePasses, owner: RecordedThreadIncarnation) -> Self: + return replace(self, services=passes, heartbeat=passes.heartbeat_completed_at(owner)) + + @projected(view='wire') + def service_status(self) -> tuple[str, ...]: + clock = self.heartbeat.aged_line('Tristan last completed', time.time()) + failures = self.services.failure_lines() + if not failures and not self.services.heartbeat_current(self.heartbeat): + failures = ('no failure recorded, so no heartbeat is running',) + return (*failures, *(clock or ('Tristan last completed: unrecorded',))) + + def require_current(self) -> None: + """A disclosure without a health header cannot serve stale cards silently.""" + if not self.services.heartbeat_current(self.heartbeat): + raise BoardUnavailable(self, 'the commander heartbeat is stale') + + +class BoardUnavailable(ToolRefusedError): + """A missing or mismatched publication retains the service evidence that explains it.""" + + def __init__(self, reading: ServiceReading, reason: str) -> None: + self.reading = reading + super().__init__('\n'.join((*reading.service_status, f'Board unavailable: {reason}'))) + + @dataclass(frozen=True, slots=True) class ServicePassStore(SingleFileStore, DurableFile, LockedStore[ServicePasses], DependentStateTier): """The connector's log reaches nobody; the commander's standing rules read this file.""" diff --git a/src/agent_comms/shell_effects.py b/src/agent_comms/shell_effects.py index 9da819f982..3b808e828c 100644 --- a/src/agent_comms/shell_effects.py +++ b/src/agent_comms/shell_effects.py @@ -73,7 +73,7 @@ class WriteStorageEffect(StorageEffect): path: Path def check(self, projection: StorageProjection) -> None: - projection.rule.require_path(projection.thread, self.path) + projection.require_write(self.path) @dataclass(frozen=True) diff --git a/src/agent_comms/standing_rules.py b/src/agent_comms/standing_rules.py index b8073b43ef..1f15a3274c 100644 --- a/src/agent_comms/standing_rules.py +++ b/src/agent_comms/standing_rules.py @@ -35,7 +35,7 @@ from agent_comms.recorded_time import EventTime, RecordedEventTime from agent_comms.repository_name import RepositoryName from agent_comms.repo import GitCommandError -from agent_comms.service_passes import ServicePassStore +from agent_comms.service_passes import FailingPass, ServicePasses, ServicePassStore from agent_comms.storage_rules import MemoryUse, StorageRule, UnitStorage from agent_comms.child_process import ProcessIdentity, ResidentMemory from agent_comms.root_runtime import RootRuntimeStore @@ -500,7 +500,7 @@ def prepare(cls, comms: Comms, commander: str, holds: HeldOrders, pulls.checkout(directory).path)) except (OSError, GitCommandError, WorkBranchError, TimeoutExpired) as error: - pulls.unread(todo, error) + pulls.unread((todo,), error) for member in cls.members_with(cls): member.prepare_captured(tuple(prepared), pulls) @@ -1208,26 +1208,26 @@ def line(self) -> str: @dataclass(frozen=True) -class ServicePassFailing(IndependentStandingRule, DatedStandingRule, FamilyScan): +class ServicePassFailing(FailingPass, IndependentStandingRule, DatedStandingRule, FamilyScan): """A Core service watch whose newest pass failed; the connector's log reaches nobody.""" watch: str - since: float error: str = field(compare=False) @classmethod def violations(cls, comms: Comms, commander: str) -> tuple[Self, ...]: - failing = ServicePassStore.at(comms.root).read().failing + return cls.from_passes(ServicePassStore.at(comms.root).read()) + + @classmethod + def from_passes(cls, passes: ServicePasses) -> tuple[Self, ...]: return tuple(cls(watch=watch, since=failed.since, error=failed.error) - for watch, failed in sorted(failing.items())) + for watch, failed in sorted(passes.failing.items())) def began(self) -> float: return self.since def line(self) -> str: - return (f"FIX service pass {self.watch} failing since {RecordedEventTime(self.since).clock()}:" - f" {self.error}. is anyone fixing it, or letting it pile up?" - " -> fix the owner of the failing read this wake") + return self.failure_line(self.watch) @dataclass(frozen=True) class BoardDisagrees(UndatedStandingRule): diff --git a/src/agent_comms/storage_rules.py b/src/agent_comms/storage_rules.py index 6164449fa1..9c60b8ffde 100644 --- a/src/agent_comms/storage_rules.py +++ b/src/agent_comms/storage_rules.py @@ -6,10 +6,11 @@ import shutil import time from abc import abstractmethod -from collections.abc import Iterator, Sequence +from collections.abc import Iterable, Iterator, Mapping, Sequence from dataclasses import dataclass, field, replace from functools import cached_property from pathlib import Path +from subprocess import TimeoutExpired from typing import TYPE_CHECKING, ClassVar, Self from agent_comms.declared_family import DeclaredFamily @@ -20,12 +21,13 @@ MalformedRecord, ) from agent_comms.locked_store import DurableFile +from agent_comms.errors import TodoConflict, TodoError from agent_comms.problem_number import ProblemNumber from agent_comms.repo import Checkout, GitCommandError, Repo from agent_comms.repository_name import RepositoryName from agent_comms.storage_room import RoomAllocation from agent_comms.thread_identity import ThreadIncarnation -from agent_comms.work_branches import WorkBranches +from agent_comms.work_branches import WorkBranchError, WorkBranches from .shell_effects import StorageDenied, StorageEffect from .thread_storage import ( @@ -36,7 +38,10 @@ if TYPE_CHECKING: from agent_comms.registration import Registration - from agent_comms.todos import Todo, TodoStore + from agent_comms.comms import Comms + from agent_comms.registry_document import RegistrySnapshot + from agent_comms.todos import ProgressRead, Todo, TodoStore + from agent_comms.held_orders import LiveHold from .native_tool_call import CodingCall @@ -96,14 +101,29 @@ def owns_declared(self, rule: StorageRule, owner: Thread, path: Path, repository: RepositoryName) -> bool: return self.owns_present(rule, owner, path) - def problems(self) -> tuple[str, ...]: - """Only a problem allocation reserves an edit boundary for a problem.""" + @classmethod + def for_work(cls, rule: StorageRule, owner: Thread, problem: str, + primary: RepositoryName, companions: tuple[RepositoryName, ...], + generation: GoalId) -> tuple[Path, ...]: + """Read/launch allocations reserve no editable work; declared members supply it.""" return () - def source_repositories(self, primary: RepositoryName, + @classmethod + def source_repositories(cls, primary: RepositoryName, companions: tuple[RepositoryName, ...]) -> tuple[RepositoryName, ...]: return () + def editable_roots(self, rule: StorageRule, owner: Thread, store: TodoStore, + repository: RepositoryName) -> frozenset[Path]: + return frozenset() + + def reserves_boundary(self) -> bool: + return False + + def edit_boundary(self, rule: StorageRule, owner: Thread, path: Path, + store: TodoStore) -> frozenset[Path]: + return frozenset() + @dataclass(frozen=True) class HomeCheckoutAllocation(CheckoutAllocation): @@ -117,16 +137,43 @@ def owns_present(self, rule: StorageRule, owner: Thread, path: Path) -> bool: return path in owner.home_checkouts -class ProblemCheckoutAllocation(CheckoutAllocation): +class EditableCheckoutAllocation(CheckoutAllocation): + """Declared editable members acquire Git identity once at the source-write boundary.""" + + def reserves_boundary(self) -> bool: + return True + + def edit_boundary(self, rule: StorageRule, owner: Thread, path: Path, + store: TodoStore) -> frozenset[Path]: + try: + repository = Repo.of(path).github + if not self.owns_declared(rule, owner, path, repository): + return frozenset() + return frozenset(root for root in self.editable_roots(rule, owner, store, repository) + if root == path) + except (GitCommandError, TodoError) as error: + raise StorageDenied(f'Cannot authorize source write in {path}: {error}') from error + + +class ProblemCheckoutAllocation(EditableCheckoutAllocation): """A checkout allocated to one problem; ``verb`` is the command that creates it.""" verb: ClassVar[str] problem: str - def problems(self) -> tuple[str, ...]: - return (self.problem,) + def editable_roots(self, rule: StorageRule, owner: Thread, store: TodoStore, + repository: RepositoryName) -> frozenset[Path]: + current = store.get(self.problem) + current.state.require_open(current) + assignment = current.held() + current.require_assignee(owner.incarnation, assignment.generation) + companions = current.order.source_repositories() + if repository not in self.source_repositories(current.repo, companions): + return frozenset() + return rule.edit_boundaries(owner, current.id, current.repo, companions, assignment.generation) - def source_repositories(self, primary: RepositoryName, + @classmethod + def source_repositories(cls, primary: RepositoryName, companions: tuple[RepositoryName, ...]) -> tuple[RepositoryName, ...]: return (primary,) @@ -151,6 +198,12 @@ def allocated(cls, problem: str, suffix: str) -> Iterator[CheckoutAllocation]: if not suffix: yield cls(problem) + @classmethod + def for_work(cls, rule: StorageRule, owner: Thread, problem: str, + primary: RepositoryName, companions: tuple[RepositoryName, ...], + generation: GoalId) -> tuple[Path, ...]: + return (rule.order_checkout(rule.roots_of(owner), problem),) + def owns_present(self, rule: StorageRule, owner: Thread, path: Path) -> bool: return path == rule.order_checkout(rule.roots_of(owner), self.problem) @@ -179,7 +232,15 @@ class RepositoryCheckoutAllocation(ProblemCheckoutAllocation, RepositoryAllocati problem: str suffix: str - def source_repositories(self, primary: RepositoryName, + @classmethod + def for_work(cls, rule: StorageRule, owner: Thread, problem: str, + primary: RepositoryName, companions: tuple[RepositoryName, ...], + generation: GoalId) -> tuple[Path, ...]: + return tuple(rule.repository_checkout(rule.roots_of(owner), problem, repository) + for repository in companions) + + @classmethod + def source_repositories(cls, primary: RepositoryName, companions: tuple[RepositoryName, ...]) -> tuple[RepositoryName, ...]: return companions @@ -194,14 +255,43 @@ def owns_declared(self, rule: StorageRule, owner: Thread, path: Path, @dataclass(frozen=True) -class GoalCheckoutAllocation(RepositoryAllocation): +class GoalCheckoutAllocation(RepositoryAllocation, EditableCheckoutAllocation): goal: GoalId suffix: str + @classmethod + def source_repositories(cls, primary: RepositoryName, + companions: tuple[RepositoryName, ...]) -> tuple[RepositoryName, ...]: + return tuple(dict.fromkeys((primary, *companions))) + + def editable_roots(self, rule: StorageRule, owner: Thread, store: TodoStore, + repository: RepositoryName) -> frozenset[Path]: + roots = frozenset(path.resolve() for goal in owner.goal.matching(self.goal) + for path in self.for_goal(rule, owner, goal.id, repository)) + for current in store.for_goal(owner.incarnation, self.goal): + companions = current.order.source_repositories() + if repository in (current.repo, *companions): + roots |= rule.edit_boundaries(owner, current.id, current.repo, companions, self.goal) + return roots + + @classmethod + def for_work(cls, rule: StorageRule, owner: Thread, problem: str, + primary: RepositoryName, companions: tuple[RepositoryName, ...], + generation: GoalId) -> tuple[Path, ...]: + return tuple(path for repository in cls.source_repositories(primary, companions) + for path in cls.for_goal(rule, owner, generation, repository)) + + @classmethod + def for_goal(cls, rule: StorageRule, owner: Thread, generation: GoalId, + repository: RepositoryName) -> tuple[Path, ...]: + return (rule.repository_checkout(rule.roots_of(owner), generation, repository),) + @classmethod def read(cls, owner: Thread, path: Path, problem: str, suffix: str) -> Iterator[CheckoutAllocation]: - if path == owner.project_path or not suffix: + # An allocated checkout can also be the current project. Its Goal identity, + # not the fact it is home, determines whether it grants editable work. + if not suffix: return try: goal = FieldCodec.decode(GoalId, problem) @@ -265,6 +355,13 @@ def repository_checkout(self, roots: StorageRoots, problem: str, checkout = self.order_checkout(roots, problem) return checkout.with_name(f'{checkout.name}--{repository.name}') + def edit_boundaries(self, owner: Thread, problem: str, primary: RepositoryName, + companions: tuple[RepositoryName, ...], generation: GoalId) -> frozenset[Path]: + """Physical editable roots follow declared allocations, even before materialization.""" + return frozenset(path.resolve() + for member in CheckoutAllocation.members_with(CheckoutAllocation) + for path in member.for_work(self, owner, problem, primary, companions, generation)) + def problem_checkouts(self, roots: StorageRoots, problem: str) -> tuple[Path, ...]: """Every checkout allocated to one problem: its check repository's, then each other's.""" checkout = self.order_checkout(roots, problem) @@ -403,15 +500,6 @@ def require_destination(self, thread: Thread, path: Path) -> None: if path.is_relative_to(self.scratch_base) and not path.is_relative_to(scratch): self.deny(f'foreign scratch destination {path}; registered scratch is {scratch}') - def require_path(self, thread: Thread, path: Path) -> None: - """A write in a present checkout follows that checkout's allocation.""" - self.require_destination(thread, path) - base = self.checkout_base - if path.is_relative_to(base) and path != base: - checkout = base / path.relative_to(base).parts[0] - if not self.owns_present_checkout(thread, checkout): - self.deny(f'{checkout} is not a checkout of {thread.name}') - def order_text(self, thread: Thread, pythons: Sequence[str]) -> str: """The rule uses the same declared paths as storage enforcement.""" checkout = self.order_checkout(self.roots_of(thread), "P") @@ -490,11 +578,33 @@ class StorageProjection: rule: StorageRule thread: Thread + comms: Comms + + @cached_property + def snapshot(self) -> RegistrySnapshot: + return self.comms.registry.snapshot() @cached_property def storage(self) -> UnitStorage: return UnitStorage.read(self.rule, self.thread) + def require_write(self, path: Path) -> None: + self.rule.require_destination(self.thread, path) + owners = self.snapshot.threads_for(self.thread.incarnation.require_recorded()) + if not owners: + self.rule.deny('The source writer incarnation is no longer registered.') + owner, = owners + resolved = path.resolve() + sources = Checkout.git_roots(resolved, *resolved.parents) + store = self.comms.goals.todos + allowed = frozenset(boundary for root in sources + for allocation in CheckoutAllocation.for_path(self.rule, owner, root) + for boundary in allocation.edit_boundary(self.rule, owner, root, store)) + if (sources or path.is_relative_to(self.rule.checkout_base) + or resolved.is_relative_to(owner.project_path.resolve())) and not any( + resolved.is_relative_to(root) for root in allowed): + self.rule.deny(f'{path} is a read/launch source, not allocated editable work of {owner.name}.') + def check_call(self, call: CodingCall) -> tuple[StorageEffect, ...]: """Check the call's declared effects against this same lazy owner snapshot.""" effects = call.storage_effects(self.thread.worktree.path) @@ -505,3 +615,191 @@ def check(self, effects: Sequence[StorageEffect]) -> None: """Preflight every effect of one call against this one snapshot.""" for effect in effects: effect.check(self) + + +@dataclass(frozen=True) +class EditWorkplaces: + """Allocation edit ownership plus retained edit-review evidence, captured for one operation. + + Review-only check/PR sources preserve original calls; they never grant source writes. + Zero or multiple review candidates never select a held order. Records stay put. + """ + + rule: StorageRule + owners: Mapping[ThreadIncarnation, Thread] + allocated: Mapping[Path, frozenset[str]] + orders: Mapping[str, Todo] = field(default_factory=dict, kw_only=True) + review_sources: Mapping[Path, frozenset[str]] = field(default_factory=dict, kw_only=True) + + @staticmethod + def owner_threads(snapshot: RegistrySnapshot) -> dict[ThreadIncarnation, Thread]: + return {thread.incarnation: thread for thread in snapshot.threads.values()} + + @classmethod + def for_assignment(cls, root: Path, snapshot: RegistrySnapshot, held: tuple[Todo, ...], + proposed: tuple[Todo, ...]) -> Self: + """Capture declared physical boundaries without forge reads or a Todo write lock.""" + rule = StorageRule.load(root) + orders = {todo.id: todo for todo in (*held, *proposed)} + allocated: dict[Path, set[str]] = {} + for todo in orders.values(): + companions = todo.order.source_repositories() + for assignment in todo.current_holder.current_assignments(): + for owner in assignment.holder_in(snapshot): + for path in rule.edit_boundaries(owner, todo.id, todo.repo, companions, + assignment.generation): + allocated.setdefault(path, set()).add(todo.id) + return cls(rule, cls.owner_threads(snapshot), + {path: frozenset(problems) for path, problems in allocated.items()}, + orders=orders) + + def roots_for(self, problem: str) -> frozenset[Path]: + return frozenset(path for path, problems in self.allocated.items() if problem in problems) + + def require_owner(self, todo: Todo, snapshot: RegistrySnapshot) -> None: + assignment = todo.held() + before = frozenset((thread.incarnation, self.rule.roots_of(thread)) + for thread in self.owners.values() + if thread.incarnation == assignment.owner_identity) + after = frozenset((thread.incarnation, self.rule.roots_of(thread)) + for thread in assignment.holder_in(snapshot)) + if not before or before != after: + raise TodoConflict(todo) + + def require_current(self, todo: Todo, snapshot: RegistrySnapshot) -> None: + if todo.id not in self.orders: + raise TodoConflict(todo) + todo.require_result(self.orders[todo.id]) + if self.roots_for(todo.id): + self.require_owner(todo, snapshot) + + @classmethod + def capture(cls, comms: Comms, snapshot: RegistrySnapshot, holds: Iterable[LiveHold], + pulls: ProgressRead + ) -> dict[ThreadIncarnation, EditWorkplaces]: + """Acquire source/forge facts once, then match edits without Git or network calls.""" + holds = tuple(holds) + if not holds: + return {} + prepared = cls.for_assignment(comms.root, snapshot, tuple(hold.todo for hold in holds), ()) + repositories = snapshot.commander_project.require().repos + registered = frozenset(repo.github for repo in repositories) + rule = prepared.rule + declared = {todo.id: todo.order.source_repositories() for todo in prepared.orders.values()} + trees: dict[RepositoryName, dict[str, tuple[str, ...]]] = {} + for repo in repositories: + branches = trees.setdefault(repo.github, {}) + for branch, worktrees in pulls.worktrees(repo).items(): + branches[branch] = (*branches.get(branch, ()), *worktrees) + listed = {Path(path) for branches in trees.values() + for worktrees in branches.values() for path in worktrees} + explicit = {order.check.directory for hold in holds + for order in hold.todo.order.declarations} + links = tuple(dict.fromkeys(link for hold in holds + for order in hold.todo.order.declarations + for link in order.pull_requests)) + requests = {link: row + for repo in repositories + for row in pulls.rows(repo.github, tuple( + link.number for link in links if link.repo == repo.github)) + for link in links if link.repo == repo.github and link.number == row.number} + progress = pulls.progress(tuple(hold.todo for hold in holds), comms.registry) + source_orders: dict[Path, dict[str, Todo]] = {} + for hold in holds: + todo = hold.todo + paths = {order.check.directory for order in todo.order.declarations} + paths.update(prepared.roots_for(todo.id)) + paths.update(Path(path) for order in todo.order.declarations + for link in order.pull_requests if link in requests + for path in trees[link.repo].get(requests[link].branch, ())) + for path in paths: + source_orders.setdefault(path, {})[todo.id] = todo + # Qualify only captured Git roots, never nearest parents of individual events. + # An alias cannot manufacture another problem allocation around the same source. + present = Checkout.git_roots(*listed, *source_orders) + actual: dict[Path, Repo] = {} + for path in present: + affected = source_orders.get(path, {}) + readable: tuple[Path, ...] = (path,) if affected else () + for todo in affected.values(): + readable = progress[todo.id].readable_paths(readable) + for source in readable: + if source.is_absolute() and Checkout(source).resolved().path == source: + try: + repository = Repo.of(source) + if repository.github in registered: + # A declared source can be a separate clone. Qualify its own + # worktree facts within the same per-source failure boundary. + extra = (pulls.worktrees(repository) + if source in explicit - listed else {}) + actual[source] = repository + branches = trees[repository.github] + for branch, worktrees in extra.items(): + if str(source) in worktrees: + branches[branch] = (*branches.get(branch, ()), str(source)) + except (OSError, GitCommandError, WorkBranchError, TimeoutExpired) as error: + pulls.unread(affected.values(), error) + owners = prepared.owners + # The allocation cut is the only problem ownership map. Qualify its source + # identity once; absent Git never enables a generic check-root fallback. + allocated = { + path: frozenset( + problem for problem in problems if path in actual + for assignment in prepared.orders[problem].current_holder.current_assignments() + for owner in assignment.holder_in(snapshot) + for allocation in CheckoutAllocation.for_path(rule, owner, path) + if actual[path].github in allocation.source_repositories( + prepared.orders[problem].repo, declared[problem]) + and allocation.owns_declared(rule, owner, path, actual[path].github)) + for path, problems in prepared.allocated.items() + } + workplaces: dict[ThreadIncarnation, EditWorkplaces] = {} + for holder in dict.fromkeys(hold.holder.incarnation for hold in holds): + sources: dict[Path, set[str]] = {} + for hold in holds: + if hold.holder.incarnation != holder: + continue + todo = hold.todo + linked = tuple(link for order in todo.order.declarations + for link in order.pull_requests) + for order in todo.order.declarations: + path = order.check.directory + # Explicit pre-PR source is valid only for its primary actual repository, + # and ceases being a generic workplace once that source has a PR link. + if (path in actual and actual[path].github == todo.repo + and not any(link.repo == todo.repo for link in linked)): + sources.setdefault(path, set()).add(todo.id) + for link in linked: + if link not in requests: + continue # Unregistered repositories cannot qualify a source boundary. + for worktree_path in trees[link.repo].get(requests[link].branch, ()): + root = Path(worktree_path) + if root in actual and actual[root].github == link.repo: + sources.setdefault(root, set()).add(todo.id) + workplaces[holder] = cls(rule, owners, allocated, review_sources={ + path: frozenset(problems) for path, problems in sources.items()}) + return workplaces + + def problems_for(self, path: str) -> frozenset[str]: + original = Path(path) + if not original.is_absolute(): + return frozenset() + # Resolve the edited path once, not its missing leaf with Git. Lexical membership + # rejects aliases into a checkout; resolved membership rejects symlink escapes. + resolved = original.resolve() + parts = (original.relative_to(self.rule.checkout_base).parts + if original.is_relative_to(self.rule.checkout_base) else ()) + boundary = self.rule.checkout_base.joinpath(*parts[:1]) + reserved = boundary in self.allocated or any(allocation.reserves_boundary() + for thread in self.owners.values() + for prefix in (self.rule.order_checkout(self.rule.roots_of(thread), ''),) + if boundary.parent == prefix.parent and boundary.name.startswith(prefix.name) + for allocation in CheckoutAllocation.for_path(self.rule, thread, boundary)) + # Even a missing/released/unknown allocation reserves its lexical boundary. Only + # captured actual Git identity may supply an owner; absence never enables spill. + candidates = ({boundary: self.allocated.get(boundary, frozenset())} + if reserved else self.review_sources) + return frozenset(problem for root, problems in candidates.items() + if original.is_relative_to(root) and resolved.is_relative_to(root) + for problem in problems) + diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index a3e7130b1f..1624374d78 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -134,7 +134,7 @@ from agent_comms.repo import Checkout, GitCommandError, Repo from agent_comms.repository_name import RepositoryName from agent_comms.slop import DeclarationDerivation, Slop -from agent_comms.storage_rules import StorageRule, UnitStorage +from agent_comms.storage_rules import EditWorkplaces, StorageRule, UnitStorage from agent_comms.thread_identity import ( CoreWatcherIncarnation, RecordedThreadIncarnation, @@ -1565,9 +1565,10 @@ def __init__(self, todos: Sequence[Todo] = ()) -> None: PullRequests.__init__(self, todos) self._progress: dict[str, OrderProgress] = {} - def unread(self, todo: Todo, error: Exception) -> None: - directory = todo.order.checked().check.directory - self._progress[todo.id] = UnreadProgress(f"Source read in {directory}: {error}") + def unread(self, todos: Iterable[Todo], error: Exception) -> None: + for todo in todos: + directory = todo.order.checked().check.directory + self._progress[todo.id] = UnreadProgress(f"Source read in {directory}: {error}") def progress(self, held: Sequence[Todo], registry: Registration) -> dict[str, OrderProgress]: """Each held order's progress, read once per frame however many readers ask.""" @@ -2230,6 +2231,10 @@ def corrected(self, correction: CorrectOrderTodo) -> ProblemOrder: default_factory=UnlocatedProblemFinding ) + def source_repositories(self) -> tuple[RepositoryName, ...]: + return tuple(dict.fromkeys(repository for declaration in self.declarations + for repository in declaration.source_repositories())) + @property def declarations(self) -> tuple[CheckedProblemOrder, ...]: return () @@ -3371,9 +3376,14 @@ class HuntRecord(TodoStoreRows, TypedTable): hunt: Hunt -class SharedWorktreeRefusal(TodoError): +class SharedWorktreeRefusal(TodoError, RejectedEnforcementError): """Two different order holders cannot write into the same physical worktree.""" + reason: ClassVar[str] = 'the proposed holder shares an occupied edit boundary' + + def explanation(self) -> str: + return str(self) + def __init__(self, problem: str, holding_problem: str, holder: ThreadIncarnation, holding_holder: ThreadIncarnation, worktree: Path) -> None: self.problem = problem @@ -3583,20 +3593,24 @@ def running_holders(self, snapshot: RegistrySnapshot) -> tuple[Thread, ...]: return tuple(owner for owner in self.holder_in(snapshot) if owner.process_alive) def require_available_in(self, db: sqlite3.Connection, problem: str, incoming: IncomingWork, - snapshot: RegistrySnapshot) -> None: - """Validate this holder against open work using the transaction's registry snapshot; - the incoming problems move with this holder, so their current holders share nothing.""" + snapshot: RegistrySnapshot, workplaces: EditWorkplaces) -> None: + """Compare prepared physical edit ownership; launch/check paths confer no occupancy.""" + proposed = workplaces.orders[problem] + prepared = proposed.held() + if (self.owner_identity, self.generation) != (prepared.owner_identity, prepared.generation): + raise TodoConflict(proposed) + workplaces.require_owner(proposed, snapshot) incoming.require_free(db, self.owner_identity, snapshot) - worktrees = {path for owner in self.holder_in(snapshot) for path in owner.home_checkouts} + worktrees = workplaces.roots_for(problem) for held in TodoStore._open(db, "assignment IS NOT NULL", ()): previous = held.held() if held.id in incoming.problems or previous.owner_identity == self.owner_identity: continue - for other in previous.holder_in(snapshot): - for shared in worktrees.intersection(other.home_checkouts): - raise SharedWorktreeRefusal( - problem, held.id, self.owner_identity, other.incarnation, shared - ) + workplaces.require_current(held, snapshot) + for shared in worktrees.intersection(workplaces.roots_for(held.id)): + raise SharedWorktreeRefusal( + problem, held.id, self.owner_identity, previous.owner_identity, shared + ) @property def parent_identity(self) -> ThreadIncarnation: @@ -3738,6 +3752,10 @@ def __post_init__(self) -> None: def prepare(self, goals: Goals, current: Todo) -> Self: return self + def proposed_work(self, store: TodoStore, problem: str) -> tuple[Todo, ...]: + """Only assignment members prepare an incoming allocation boundary.""" + return () + def record(self, goals: Goals, problem: str) -> Todo: """Write this operation; publish its effects only after the canonical row was written.""" changed = goals.apply_problem(problem, self) @@ -3852,11 +3870,14 @@ def authored_by(self, actor: ThreadIncarnation) -> Self: return replace(self, actor=actor, proposed=replace(self.proposed, parent=actor.name, parent_created=actor.wire_birth)) + def proposed_work(self, store: TodoStore, problem: str) -> tuple[Todo, ...]: + return (replace(store.get(problem), assignment=self.proposed, order=self.order),) + def validate(self, db: sqlite3.Connection, current: Todo, store: TodoStore) -> None: self.proposed.require_available_in( db, current.id, IncomingWork( self.actor, self.group | {current.id}, frozenset({self.order.id})), - RegistryStore.at(store.root).snapshot()) + RegistryStore.at(store.root).snapshot(), store.prepared_edits()) OwnerRequest.require(db, id=self.request).require_open(current.creator_identity) def apply(self, current: Todo) -> Todo: @@ -4937,13 +4958,17 @@ class RestoreHolderTodo(TodoCommand): actor: ThreadIncarnation previous: Assignment + def proposed_work(self, store: TodoStore, problem: str) -> tuple[Todo, ...]: + return (replace(store.get(problem), assignment=self.previous),) + def validate(self, db: sqlite3.Connection, current: Todo, store: TodoStore) -> None: snapshot = RegistryStore.at(store.root).snapshot() if not self.previous.running_holders(snapshot): raise TodoConflict(current) self.previous.require_available_in(db, current.id, IncomingWork( self.actor, frozenset({current.id}), - frozenset(declared.id for declared in current.order.declarations)), snapshot) + frozenset(declared.id for declared in current.order.declarations)), snapshot, + store.prepared_edits()) def changed(self, current: Todo) -> Todo: self.actor.require_may_change(current) @@ -5970,6 +5995,18 @@ def writing(self) -> Iterator[sqlite3.Connection]: f"{self.path}: {type(error).__name__}: {error}" ) from error + def preparing_edits(self, proposed: tuple[Todo, ...]) -> TodoStore: + """Prepare physical allocations outside SQL; the prepared store lives for this write only.""" + if not proposed: + return self + snapshot = RegistryStore.at(self.root).snapshot() + held = self.held_orders() + workplaces = EditWorkplaces.for_assignment(self.root, snapshot, held, proposed) + return PreparedTodoStore(self.root, workplaces) + + def prepared_edits(self) -> EditWorkplaces: + raise TodoError('Assignment allocation evidence was not prepared before the write.') + def allocate_problem(self) -> str: with self.writing() as db: return str(ProblemAllocation.allocate(db)) @@ -6216,10 +6253,11 @@ def create( proposed = Todo.ordered( todo_id, repo, text, identity, Assignment.of(assignee, assigner, generation), review_at, order, request) - with self.writing() as db: + prepared = self.preparing_edits((proposed,)) + with prepared.writing() as db: proposed.held().require_available_in( db, todo_id, IncomingWork(assigner, frozenset({todo_id}), frozenset({order.id})), - RegistryStore.at(self.root).snapshot()) + RegistryStore.at(self.root).snapshot(), prepared.prepared_edits()) OwnerRequest.require(db, id=request).require_open(identity) proposed.insert_if_absent(db) current = self._current(db, todo_id) @@ -6247,9 +6285,12 @@ def apply(self, todo_id: str, command: TodoCommand) -> Todo: def apply_together(self, commands: Sequence[ProblemCommand]) -> tuple[Todo, ...]: """Write the commands under one Todo transaction, in order: each validates against the rows the commands before it wrote, and any refusal writes none of them.""" - with self.writing() as db: + proposed = tuple(todo for command in commands + for todo in command.operation.proposed_work(self, command.problem)) + prepared = self.preparing_edits(proposed) + with prepared.writing() as db: for command in commands: - command.write(db, self) + command.write(db, prepared) return self._snapshot_rows( db, tuple(self._current(db, command.problem) for command in commands)) @@ -6303,6 +6344,16 @@ def assign( for expected in problems)) +@dataclass(frozen=True, slots=True) +class PreparedTodoStore(TodoStore): + """One mutation's immutable allocation cut, not another persisted edit ownership authority.""" + + workplaces: EditWorkplaces + + def prepared_edits(self) -> EditWorkplaces: + return self.workplaces + + @dataclass(frozen=True, slots=True) class BoardTodoHistory(TodoStore): """One read's immutable aggregates; explicit story requests still use the store.""" diff --git a/src/agent_comms/turn_digests.py b/src/agent_comms/turn_digests.py index 46bbb2c55b..c8d29c7f3b 100644 --- a/src/agent_comms/turn_digests.py +++ b/src/agent_comms/turn_digests.py @@ -6,8 +6,7 @@ from collections.abc import Iterable, Iterator, Mapping from dataclasses import dataclass, field, replace from pathlib import Path -from subprocess import TimeoutExpired -from typing import TYPE_CHECKING, ClassVar, Self +from typing import TYPE_CHECKING, Annotated, ClassVar, Self from pydantic import JsonValue @@ -16,17 +15,18 @@ from .declared_family import DeclaredFamily from .errors import ToolRefusedError from .external_transcript import ExternalTranscript, StoreRead -from .field_codec import FieldCodec +from .field_codec import FieldCodec, WireTransient, projected from .held_orders import HeldOrders, LiveHold from .order_work import OrderWork -from .repo import Checkout, GitCommandError, Repo -from .storage_rules import CheckoutAllocation, StorageRule +from .problem_number import ProblemNumber +from .storage_rules import EditWorkplaces from .thread_execution import ConversationPreparation from .thread_identity import ThreadIncarnation from .threads import Thread -from .todos import ProgressRead, PullRequestLink, Todo, TodoStore +from .todos import NoCommitProgress, OrderProgress, ProgressRead, Todo, TodoStore from .tool_declaration import EditCall from .tool_result import ToolResult +from .service_passes import ServiceReading from .transcript_events import IncomingTranscript, TurnDigest from .turn_item import NativeTurnId from .work_branches import WorkBranchError @@ -34,8 +34,6 @@ if TYPE_CHECKING: from .comms import Comms from .registry_document import RegistrySnapshot - from .repository_name import RepositoryName - from .todos import OrderProgress from .transcript_sources import TranscriptSource @@ -75,8 +73,26 @@ def to_wire(self) -> JsonValue: @dataclass(frozen=True) -class UnjudgedEdits(ToolResult): +class UnavailableOrderEdits(ToolResult): + """One original-edit source failed at its foreign I/O boundary; other reads survive.""" + + problem: ProblemNumber + owner: ThreadIncarnation + cause: Annotated[WorkBranchError | OSError, WireTransient()] + + @projected(view='wire') + def error(self) -> str: + return str(self.cause) + + +@dataclass(frozen=True) +class UnjudgedEdits(ServiceReading): edits: tuple[EditCall, ...] + unavailable: tuple[UnavailableOrderEdits, ...] = () + + @projected(view='wire') + def complete(self) -> bool: + return not self.unavailable class TurnBound(DeclaredFamily): @@ -231,6 +247,11 @@ class OrderRead[Result](TranscriptRead[Result]): class OrderEditRead[Result](OrderRead[Result]): """A held order's edits: the ones its workplaces own that no verdict names.""" workplaces: EditWorkplaces + progress: OrderProgress = NoCommitProgress() + + def read(self) -> Result: + self.progress.require_read() + return super().read() @classmethod def for_commander(cls, comms: Comms, commander: str) -> Iterator[Self]: @@ -238,7 +259,7 @@ def for_commander(cls, comms: Comms, commander: str) -> Iterator[Self]: held = HeldOrders.of(comms.goals.todos, snapshot, commander) history = comms.goals.todos.history_for_commander(snapshot.require(commander).incarnation) pulls = ProgressRead(history) - workplaces = EditWorkplaces.capture(comms, snapshot, held.live, history, pulls) + workplaces = EditWorkplaces.capture(comms, snapshot, held.live, pulls) yield from cls.judgeable( comms.root, held.live, pulls.progress(held.todos, comms.registry), workplaces) @@ -248,7 +269,8 @@ def judgeable(cls, root: Path, holds: Iterable[LiveHold], workplaces: Mapping[ThreadIncarnation, EditWorkplaces]) -> Iterator[Self]: """The reads a verdict can still clear: an order whose PR merged has no head left to judge, so its edits are nobody's review and its transcript is never read.""" - return (read for read in cls.for_orders(root, holds, workplaces) + return (replace(read, progress=progress[read.todo.id]) + for read in cls.for_orders(root, holds, workplaces) if progress[read.todo.id].judgeable) @classmethod @@ -277,159 +299,6 @@ def edits_from_turns(self, turns: tuple[TurnDigest, ...]) -> tuple[EditCall, ... return self.owned(reversed(edits.values())) -@dataclass(frozen=True) -class EditWorkplaces: - """One read's derived boundaries. Allocations take precedence over generic source roots; - zero or multiple candidates never select a held order. Original edit records stay put.""" - - rule: StorageRule - owners: Mapping[Path, Thread] - allocated: Mapping[Path, frozenset[str]] - sources: Mapping[Path, frozenset[str]] - - @classmethod - def capture(cls, comms: Comms, snapshot: RegistrySnapshot, holds: Iterable[LiveHold], - history: tuple[Todo, ...], pulls: ProgressRead - ) -> dict[ThreadIncarnation, EditWorkplaces]: - """Acquire source/forge facts once, then match edits without Git or network calls.""" - holds = tuple(holds) - if not holds: - return {} - repositories = snapshot.commander_project.require().repos - registered = frozenset(repo.github for repo in repositories) - rule = StorageRule.load(comms.root) - todos = {todo.id: todo for todo in history} - declared = { - todo.id: tuple(dict.fromkeys(repository for order in todo.order.declarations - for repository in order.source_repositories())) - for todo in history - } - trees: dict[RepositoryName, dict[str, tuple[str, ...]]] = {} - for repo in repositories: - branches = trees.setdefault(repo.github, {}) - for branch, worktrees in pulls.worktrees(repo).items(): - branches[branch] = (*branches.get(branch, ()), *worktrees) - listed = {Path(path) for branches in trees.values() - for worktrees in branches.values() for path in worktrees} - explicit = {order.check.directory for hold in holds - for order in hold.todo.order.declarations} - links = tuple(dict.fromkeys(link for hold in holds - for order in hold.todo.order.declarations - for link in order.pull_requests)) - requests = {PullRequestLink(repo.github, row.number): row - for repo in repositories - for row in pulls.rows(repo.github, tuple( - link.number for link in links if link.repo == repo.github))} - progress = pulls.progress(tuple(hold.todo for hold in holds), comms.registry) - source_orders: dict[Path, dict[str, Todo]] = {} - for hold in holds: - todo = hold.todo - roots = rule.roots_of(hold.holder) - paths = {order.check.directory for order in todo.order.declarations} - paths.add(rule.order_checkout(roots, todo.id)) - paths.update(rule.repository_checkout(roots, todo.id, repo) - for repo in declared[todo.id]) - paths.update(Path(path) for order in todo.order.declarations - for link in order.pull_requests if link in requests - for path in trees[link.repo].get(requests[link].branch, ())) - for path in paths: - source_orders.setdefault(path, {})[todo.id] = todo - # Qualify only captured Git roots, never nearest parents of individual events. - # An alias cannot manufacture another problem allocation around the same source. - present = Checkout.git_roots(*listed, *source_orders) - actual: dict[Path, Repo] = {} - for path in present: - affected = source_orders.get(path, {}) - readable = (path,) if affected else () - for todo in affected.values(): - readable = progress[todo.id].readable_paths(readable) - for source in readable: - if source.is_absolute() and Checkout(source).resolved().path == source: - try: - repository = Repo.of(source) - if repository.github in registered: - # A declared source can be a separate clone. Qualify its own - # worktree facts within the same per-source failure boundary. - extra = (pulls.worktrees(repository) - if source in explicit - listed else {}) - actual[source] = repository - branches = trees[repository.github] - for branch, worktrees in extra.items(): - if str(source) in worktrees: - branches[branch] = (*branches.get(branch, ()), str(source)) - except (OSError, GitCommandError, WorkBranchError, TimeoutExpired) as error: - for todo in affected.values(): - pulls.unread(todo, error) - owners = {rule.order_checkout(rule.roots_of(thread), ''): thread - for thread in snapshot.threads.values()} - allocations: dict[Path, tuple[tuple[Thread, CheckoutAllocation], ...]] = {} - for path in present: - candidates = tuple((thread, allocation) - for prefix, thread in owners.items() - if path.parent == prefix.parent and path.name.startswith(prefix.name) - for allocation in CheckoutAllocation.for_path(rule, thread, path) - if allocation.problems()) - if candidates: - allocations[path] = candidates - # Every recognized allocation reserves its boundary, even without registered Git proof. - allocated: dict[Path, frozenset[str]] = {path: frozenset() for path in allocations} - for path, repository in actual.items(): - if path in allocations: - allocated[path] = frozenset( - problem for thread, allocation in allocations[path] - for problem in allocation.problems() - if problem in todos - and repository.github in allocation.source_repositories( - todos[problem].repo, declared[problem]) - and allocation.owns_declared(rule, thread, path, repository.github)) - workplaces: dict[ThreadIncarnation, EditWorkplaces] = {} - for holder in dict.fromkeys(hold.holder.incarnation for hold in holds): - sources: dict[Path, set[str]] = {} - for hold in holds: - if hold.holder.incarnation != holder: - continue - todo = hold.todo - linked = tuple(link for order in todo.order.declarations - for link in order.pull_requests) - for order in todo.order.declarations: - path = order.check.directory - # Explicit pre-PR source is valid only for its primary actual repository, - # and ceases being a generic workplace once that source has a PR link. - if (path in actual and actual[path].github == todo.repo - and not any(link.repo == todo.repo for link in linked)): - sources.setdefault(path, set()).add(todo.id) - for link in linked: - if link not in requests: - continue # Unregistered repositories cannot qualify a source boundary. - for worktree_path in trees[link.repo].get(requests[link].branch, ()): - root = Path(worktree_path) - if root in actual and actual[root].github == link.repo: - sources.setdefault(root, set()).add(todo.id) - workplaces[holder] = cls(rule, owners, allocated, { - path: frozenset(problems) for path, problems in sources.items()}) - return workplaces - - def problems_for(self, path: str) -> frozenset[str]: - original = Path(path) - if not original.is_absolute(): - return frozenset() - # Resolve the edited path once, not its missing leaf with Git. Lexical membership - # rejects aliases into a checkout; resolved membership rejects symlink escapes. - resolved = original.resolve() - parts = (original.relative_to(self.rule.checkout_base).parts - if original.is_relative_to(self.rule.checkout_base) else ()) - boundary = self.rule.checkout_base.joinpath(*parts[:1]) - reserved = any(allocation.problems() - for prefix, thread in self.owners.items() - if boundary.parent == prefix.parent and boundary.name.startswith(prefix.name) - for allocation in CheckoutAllocation.for_path(self.rule, thread, boundary)) - # Even a missing/released/unknown allocation reserves its lexical boundary. Only - # captured actual Git identity may supply an owner; absence never enables spill. - candidates = ({boundary: self.allocated.get(boundary, frozenset())} - if reserved else self.sources) - return frozenset(problem for root, problems in candidates.items() - if original.is_relative_to(root) and resolved.is_relative_to(root) - for problem in problems) class OrderEditsRead(OrderEditRead[tuple[EditCall, ...]]): diff --git a/src/agent_comms/ui_model/observation.py b/src/agent_comms/ui_model/observation.py index b517aac113..42de3b8a77 100644 --- a/src/agent_comms/ui_model/observation.py +++ b/src/agent_comms/ui_model/observation.py @@ -445,6 +445,7 @@ def read(self, comms: Comms) -> CommanderUnitDetail: if not self.commander.current(snapshot) or not self.unit.current(snapshot): raise ToolRefusedError("The selected commander or unit incarnation is no longer current") hud = CommanderFrame.current(comms, self.commander.name) + hud.require_current() for card in hud.units: if card.unit == self.unit: return card.detail(comms, self.commander, self.orders) diff --git a/src/agent_comms/work_dispatch.py b/src/agent_comms/work_dispatch.py index 842b5d6f59..c246cf230e 100644 --- a/src/agent_comms/work_dispatch.py +++ b/src/agent_comms/work_dispatch.py @@ -74,7 +74,7 @@ def start(self, comms: Comms) -> Thread: Repo.git(repository.checkout, "fetch", "origin", "main") start = CheckoutStart.of(repository.checkout, f"auto/{name}", "origin/main") CreateCheckoutStorageEffect(checkout, repository, start).check( - StorageProjection(storage, candidate) + StorageProjection(storage, candidate, comms) ) Checkout.add(repository.checkout, checkout, start) location = candidate.runtime.new_location(comms.root, checkout) diff --git a/tests/test_commander_forks.py b/tests/test_commander_forks.py index 91b327a6b2..3a7f58ab6a 100644 --- a/tests/test_commander_forks.py +++ b/tests/test_commander_forks.py @@ -40,7 +40,8 @@ from agent_comms.repository_name import RepositoryName from agent_comms.session_file import RegistrySessionLocation from agent_comms.storage_rules import StorageRule -from agent_comms.turn_digests import EditWorkplaces, OrderEditsRead, OrderEditCountRead, TurnsSince +from agent_comms.storage_rules import EditWorkplaces +from agent_comms.turn_digests import OrderEditsRead, OrderEditCountRead, TurnsSince from agent_comms.installer.backup import StoreBackup from agent_comms.maintenance_barrier import ( MaintenanceBarrier, @@ -253,8 +254,8 @@ def test_registered_fork_sends_parent_problem_with_fork_actor(fork_bus, tmp_path assert ctx.held_orders() == (saved,) assert ctx.call(CommsTodoListTool(thread=ctx.commander)).todos == (saved,) load = ctx.call(CommsCommanderLoadTool(thread=ctx.commander)) - assert load.owed == 1 and load.timers_up == 0 and load.unjudged == 0 - assert "unjudged edits" in str(load) + assert load.load.owed == 1 and load.load.timers_up == 0 and load.load.unjudged == 0 + assert "unjudged edits" in str(load.load) 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 @@ -337,7 +338,7 @@ def test_edits_under_a_merged_order_are_nobodys_review(fork_bus): open_order = {hold.todo.id: NoCommitProgress() for hold in holds} merged = {hold.todo.id: MergedProgress("a" * 40) for hold in holds} workplaces = EditWorkplaces.capture( - comms, comms.registry.snapshot(), holds, tuple(hold.todo for hold in holds), ProgressRead()) + comms, comms.registry.snapshot(), holds, ProgressRead()) assert [read.read() for read in OrderEditCountRead.judgeable( comms.root, holds, open_order, workplaces)] == [3] assert tuple(OrderEditCountRead.judgeable(comms.root, holds, merged, workplaces)) == () @@ -361,7 +362,7 @@ def edit_workplaces(comms): snapshot = comms.registry.snapshot() holds = HeldOrders.of(comms.goals.todos, snapshot, "lead").live history = comms.goals.todos.history_for_commander(snapshot.require("lead").incarnation) - return holds, EditWorkplaces.capture(comms, snapshot, holds, history, ProgressRead(history)) + return holds, EditWorkplaces.capture(comms, snapshot, holds, ProgressRead(history)) def save_edit_order(comms, todo): @@ -481,7 +482,7 @@ def test_canonical_primary_companion_and_released_roots_take_precedence(edit_sou selected = workplaces[owner.incarnation] assert selected.problems_for(str(primary / "new.py")) == frozenset(("P40",)) assert selected.problems_for(str(companion / "new.py")) == frozenset(("P40",)) - assert selected.problems_for(str(released_path / "new.py")) == frozenset(("P41",)) + assert selected.problems_for(str(released_path / "new.py")) == frozenset() # A registered foreign repository in an undeclared companion allocation stays unowned. current = store.get("P40") save_edit_order(comms, replace(current, order=replace( diff --git a/tests/test_commander_notifications.py b/tests/test_commander_notifications.py index 7ebb45d67b..b8c9e10a5d 100644 --- a/tests/test_commander_notifications.py +++ b/tests/test_commander_notifications.py @@ -155,7 +155,7 @@ def failed_pass(comms, error): def retained_failure(comms): stored = ServicePassStore.at(comms.root).read().failing[BrokenHeartbeat.__name__] - return ServicePassFailing(BrokenHeartbeat.__name__, stored.since, stored.error) + return ServicePassFailing(watch=BrokenHeartbeat.__name__, since=stored.since, error=stored.error) def aged(pings): @@ -188,7 +188,7 @@ def enforce(self, _comms, _commander): recorded_turn(comms, (True, True)) states = state_cards(comms, Spinning) rules = (RefusedRule("refused", 1.0, "review pending"), - ServicePassFailing("independent", 1.0, "another source")) + ServicePassFailing(watch="independent", since=1.0, error="another source")) beat(comms, rules, states=states) board, = notifications(comms) assert board.task.not_applied and len(board.task.handled) == 2 @@ -254,7 +254,7 @@ def everything_to_lead(comms): def test_an_independent_heartbeat_message_never_replaces_the_board(notification_bus): comms = notification_bus - failing = ServicePassFailing("following-rule", time.time(), "independent source") + failing = ServicePassFailing(watch="following-rule", since=time.time(), error="independent source") beat(comms, (failing,)) board, = notifications(comms) raw = independent_instruction(comms, "Heartbeat pass failed: an old raw alarm") @@ -270,8 +270,8 @@ def test_an_independent_heartbeat_message_never_replaces_the_board(notification_ def test_a_later_notice_is_the_baseline_and_keeps_the_lines_still_on_it(notification_bus): comms = notification_bus - kept = ServicePassFailing("kept-rule", time.time(), "still failing") - cleared = ServicePassFailing("cleared-rule", time.time(), "now fixed") + kept = ServicePassFailing(watch="kept-rule", since=time.time(), error="still failing") + cleared = ServicePassFailing(watch="cleared-rule", since=time.time(), error="now fixed") beat(comms, (kept, cleared)) decisions, = notifications(comms) beat(comms, (kept,)) @@ -285,7 +285,7 @@ def test_a_later_notice_is_the_baseline_and_keeps_the_lines_still_on_it(notifica def test_a_raw_message_between_failures_does_not_restate_the_board(notification_bus): comms = notification_bus - rules = (ServicePassFailing("independent", 1.0, "another source"),) + rules = (ServicePassFailing(watch="independent", since=1.0, error="another source"),) beat(comms, rules) board, = notifications(comms) failed_pass(comms, RuntimeError("down")) @@ -302,7 +302,7 @@ def test_a_raw_message_between_failures_does_not_restate_the_board(notification_ def test_a_recreated_commander_inherits_no_board(notification_bus): comms = notification_bus - failing = ServicePassFailing("following-rule", time.time(), "independent source") + failing = ServicePassFailing(watch="following-rule", since=time.time(), error="independent source") beat(comms, (failing,)) board, = notifications(comms) worktree = comms.registry.require("lead").worktree @@ -362,15 +362,16 @@ def test_frame_puts_hands_first_and_folds_finished_phases_per_plan(notification_ comms = notification_bus lead = comms.registry.require("lead").incarnation worker = comms.registry.require("worker").incarnation - hud = CommanderHud(lead, (), ( + hud = CommanderHud(commander=lead, units=(), phases=( CommanderPhase(RetainedStorePhase, True, False, ()), CommanderPhase(OneOrderOwnerPhase, True, False, ()), CommanderPhase(DesignReconciledPhase, False, False, (worker,)), CommanderPhase(RecordPlanePhase, False, True, ()), CommanderPhase(ReviewQueuePhase, True, False, (worker,)), - ), OverloadDegradation.from_orders((), 0, 0), (worker,), CodexUsage(UnreadableUsage(reason='not requested'))) + ), load=OverloadDegradation.from_orders((), 0, 0), forks=(worker,), + usage=CodexUsage(UnreadableUsage(reason='not requested'))) assert hud.frame == ( - str(hud.load), + *hud.header, "commander_native/record_plane done=False ready=True holders=", "commander_native/review_queue done=True ready=False holders=worker", "commander_native/design_reconciled done=False ready=False holders=worker", @@ -395,9 +396,9 @@ def test_frame_keeps_committed_board_order_and_folds_units_needing_no_hand(tmp_p BoardEntry("judge", "JUDGE #1", now - 240, now + 3600), ) hud = CommanderHud( - RecordedThreadIncarnation("lead", 3.0), (awaiting, stuck), (), - OverloadDegradation.from_orders((), 0, 0), (), - CodexUsage(UnreadableUsage(reason="not requested")), board, + commander=RecordedThreadIncarnation("lead", 3.0), units=(awaiting, stuck), phases=(), + load=OverloadDegradation.from_orders((), 0, 0), forks=(), + usage=CodexUsage(UnreadableUsage(reason="not requested")), board=board, ) CommanderFrame.replace_all(tmp_path, (hud,)) @@ -421,7 +422,7 @@ def test_quiet_completion_is_per_commander_even_when_a_later_recipient_fails(not comms = notification_bus lead = comms.registry.require("lead") missing = replace(comms.registry.require("worker"), name="unregistered-recipient") - failing = ServicePassFailing("following-rule", time.time(), "independent source") + failing = ServicePassFailing(watch="following-rule", since=time.time(), error="independent source") frame = capture_frame(comms, "lead", ProgressRead()) frames = tuple(CommanderObservation(thread, frame, (failing,), ()) for thread in (lead, missing)) @@ -450,7 +451,7 @@ def test_recreated_watcher_reads_original_committed_unit_observation(notificatio comms = notification_bus states = state_cards(comms, UnitState) assert states and not any(state.attention() for state in states) - failing = ServicePassFailing("following-rule", time.time(), "independent source") + failing = ServicePassFailing(watch="following-rule", since=time.time(), error="independent source") beat(comms, (failing,), states=states) first, = notifications(comms) assert failing.line() in first.body @@ -496,7 +497,7 @@ def enforce(self, _comms, _commander): def test_rule_packets_list_every_new_line_and_never_restate_one(notification_bus): comms = notification_bus - rules = tuple(ServicePassFailing(f"watch-{index}", 1.0, "failed source read") + rules = tuple(ServicePassFailing(watch=f"watch-{index}", since=1.0, error="failed source read") for index in range(7)) beat(comms, rules) first, = notifications(comms) @@ -536,7 +537,7 @@ def test_resolution_is_passive_and_same_rule_can_recur(notification_bus): def test_latest_observation_follows_commander_and_service_rename_aliases(notification_bus): comms = notification_bus - beat(comms, (ServicePassFailing("watch", 1.0, "failed source read"),)) + beat(comms, (ServicePassFailing(watch="watch", since=1.0, error="failed source read"),)) first, = notifications(comms) comms.threads._rename_thread("lead", "renamed-lead") comms.threads._rename_thread(CommanderHeartbeat.sender, "renamed-heartbeat") diff --git a/tests/test_edit_allocations.py b/tests/test_edit_allocations.py new file mode 100644 index 0000000000..d618343116 --- /dev/null +++ b/tests/test_edit_allocations.py @@ -0,0 +1,154 @@ +"""Real order prepared and native write preflight share physical allocation edit ownership.""" + +from dataclasses import replace + +import pytest + +from agent_comms.errors import TodoConflict +from agent_comms.field_codec import GoalId, OrderId +from agent_comms.goal_actions import CreateProblemGoalAction +from agent_comms.order_tools import CommsTodoCreateTool +from agent_comms.repo import Checkout +from agent_comms.shell_effects import StorageDenied, WriteStorageEffect +from agent_comms.storage_rules import StorageProjection, StorageRule +from agent_comms.thread_execution import ExternalThreadExecution +from agent_comms.thread_model import DeclaredModel +from agent_comms.threads import Thread +from agent_comms.todos import Assignment, SharedWorktreeRefusal +from agent_comms.tool_caller import ThreadCaller + +pytest_plugins = ('test_commander_forks',) + + +@pytest.fixture +def allocation_bus(fork_bus): + comms, source, template = fork_bus + for name in ('first', 'second'): + comms.registry.declare(Thread( + name, frozenset({'comms'}), Checkout(source.checkout), + model=DeclaredModel('anthropic/claude-opus-4-6'), execution=ExternalThreadExecution)) + return comms, source, template + + +def issue(comms, source, template, name, problem): + lead = comms.registry.require('lead').incarnation + action = CreateProblemGoalAction( + problem=problem, repo=source.github, text='Repair actual allocated source edit ownership', + order=replace(template, id=OrderId.mint()), + request=comms.goals.todos.get('P40').request.require(), + review_at=template.at, commander=lead.name, assigner=lead) + ThreadCaller('lead').invoke(CommsTodoCreateTool(order=action, owner=name), comms) + return comms.goals.todos.get(problem) + + +def test_shared_launch_base_admits_distinct_unmaterialized_allocations_and_not_source_writes(allocation_bus): + comms, source, template = allocation_bus + first = issue(comms, source, template, 'first', 'P41') + second = issue(comms, source, template, 'second', 'P42') + one, two = (comms.registry.require(name) for name in ('first', 'second')) + assert set(one.home_checkouts).intersection(two.home_checkouts) + rule = StorageRule.load(comms.root) + first_path = rule.order_checkout(rule.roots_of(one), first.id) + second_path = rule.order_checkout(rule.roots_of(two), second.id) + assert not first_path.exists() and not second_path.exists() + assert first.held().owner_identity == one.incarnation + assert second.held().owner_identity == two.incarnation + # The shared launch/check repository is still not writable edit ownership for either holder. + for owner in (one, two): + with pytest.raises(StorageDenied): + StorageProjection(rule, owner, comms).check(( + WriteStorageEffect(source.checkout / 'declaration.py'),)) + first_path.parent.mkdir(parents=True, exist_ok=True) + source.git(source.checkout, 'worktree', 'add', '-q', '-b', 'allocated-first', str(first_path)) + StorageProjection(rule, one, comms).check(( + WriteStorageEffect(first_path / 'declaration.py'),)) + # Moving into an allocated home does not erase its declared edit permission. + one = comms.registry.declare(replace(one, worktree=Checkout(first_path))) + StorageProjection(rule, one, comms).check(( + WriteStorageEffect(first_path / 'declaration.py'),)) + with pytest.raises(StorageDenied): + StorageProjection(rule, two, comms).check(( + WriteStorageEffect(first_path / 'declaration.py'),)) + + +def test_physical_alias_refuses_public_creation_without_partial_goal_or_todo(allocation_bus): + comms, source, template = allocation_bus + first = issue(comms, source, template, 'first', 'P41') + rule = StorageRule.load(comms.root) + one, two = (comms.registry.require(name) for name in ('first', 'second')) + original = rule.order_checkout(rule.roots_of(one), first.id) + alias = rule.order_checkout(rule.roots_of(two), 'P42') + original.mkdir(parents=True) + alias.symlink_to(original, target_is_directory=True) + before = two.goal + with pytest.raises(SharedWorktreeRefusal) as refused: + issue(comms, source, template, 'second', 'P42') + assert refused.value.worktree == original.resolve() + assert refused.value.holding_problem == first.id + assert refused.value.holder == two.incarnation + assert refused.value.holding_holder == one.incarnation + assert comms.goals.todos.get(first.id) == first + assert comms.registry.require('second').goal == before + assert 'P42' not in {todo.id for todo in comms.goals.todos.list(source.github)} + + +def test_prepared_cut_refuses_a_replaced_assignment_generation(allocation_bus): + comms, source, template = allocation_bus + original = issue(comms, source, template, 'first', 'P41') + owner = comms.registry.require('second').incarnation + actor = comms.registry.require('lead').incarnation + proposed = replace(original, assignment=Assignment.of(owner, actor, GoalId.mint())) + store = comms.goals.todos + prepared = store.preparing_edits((proposed,)) + other_generation = replace(proposed.held(), generation=GoalId.mint()) + from agent_comms.todos import IncomingWork + with prepared.writing() as db: + with pytest.raises(TodoConflict): + other_generation.require_available_in( + db, original.id, IncomingWork(actor, frozenset({original.id}), + frozenset({original.order.id})), + comms.registry.snapshot(), prepared.prepared_edits()) + assert store.get(original.id) == original + + +def test_same_owner_incoming_group_can_share_its_own_physical_allocation(allocation_bus): + from agent_comms.todos import IncomingWork + comms, source, template = allocation_bus + first = issue(comms, source, template, 'first', 'P41') + second = issue(comms, source, template, 'second', 'P42') + owner = comms.registry.require('first') + actor = comms.registry.require('lead').incarnation + proposed = replace(second, assignment=Assignment.of(owner.incarnation, actor, first.held().generation)) + rule = StorageRule.load(comms.root) + original = rule.order_checkout(rule.roots_of(owner), first.id) + alias = rule.order_checkout(rule.roots_of(owner), second.id) + original.mkdir(parents=True) + alias.symlink_to(original, target_is_directory=True) + prepared = comms.goals.todos.preparing_edits((proposed,)) + with prepared.writing() as db: + proposed.held().require_available_in( + db, second.id, IncomingWork(actor, frozenset({first.id, second.id}), + frozenset({first.order.id, second.order.id})), + comms.registry.snapshot(), prepared.prepared_edits()) + + +def test_recreated_holder_cannot_use_the_prepared_birth(allocation_bus): + from agent_comms.todos import IncomingWork + comms, source, template = allocation_bus + current = issue(comms, source, template, 'first', 'P41') + old = comms.registry.require('second') + actor = comms.registry.require('lead').incarnation + proposed = replace(current, assignment=Assignment.of(old.incarnation, actor, GoalId.mint())) + prepared = comms.goals.todos.preparing_edits((proposed,)) + comms.registry.remove(old.name) + comms.registry.declare(Thread( + old.name, old.tags, Checkout(source.checkout), + model=old.model, execution=ExternalThreadExecution)) + assert comms.registry.require(old.name).incarnation != old.incarnation + with prepared.writing() as db: + with pytest.raises(TodoConflict): + proposed.held().require_available_in( + db, current.id, IncomingWork(actor, frozenset({current.id}), + frozenset({current.order.id})), + comms.registry.snapshot(), prepared.prepared_edits()) + assert comms.goals.todos.get(current.id) == current diff --git a/tests/test_recorded_board_health.py b/tests/test_recorded_board_health.py new file mode 100644 index 0000000000..e3cb6375ff --- /dev/null +++ b/tests/test_recorded_board_health.py @@ -0,0 +1,163 @@ +"""Retained board evidence never disguises a recorded failed service pass.""" + +import asyncio +import time +from contextlib import nullcontext, suppress +from dataclasses import replace + +import pytest + +from agent_comms.commander_heartbeat import CommanderHeartbeat, CommanderObservation +from agent_comms.coordinator import Coordination +from agent_comms.commander_hud import CommanderFrame +from agent_comms.errors import ToolRefusedError +from agent_comms.field_codec import ToolCallId +from agent_comms.locked_store import DocumentChange +from agent_comms.order_tools import CommsCommanderLoadTool, CommsHudTool, CommsUnjudgedTool +from agent_comms.recorded_time import RecordedEventTime +from agent_comms.service_passes import ServicePassStore +from agent_comms.todos import ProgressRead +from agent_comms.tool_caller import ThreadCaller +from agent_comms.ui_model.observation import ReadCommanderUnit +from agent_comms.ui_model.unit_state import UnitState +from commander_heartbeat_fixture import ObservedHeartbeat, capture_frame + +pytest_plugins = ('test_commander_forks',) + + +@pytest.fixture +def published_board(fork_bus, installed_runtime): + comms, source, _ = fork_bus + from test_commander_forks import record_edits + record_edits(comms, source, 1) + parent = comms.registry.require('lead') + frame = capture_frame(comms, parent.name, ProgressRead()) + states = UnitState.current(comms, parent.name, frame) + observation = CommanderObservation(parent, frame, (), states) + asyncio.run(ObservedHeartbeat((observation,)).run(lambda: nullcontext(comms))) + return comms + + +def test_recorded_failure_precedes_stale_board_and_load_without_changing_retained_data(published_board): + comms = published_board + owner = comms.registry.require('lead').incarnation.require_recorded() + services = ServicePassStore.at(comms.root) + original = CommanderFrame.current(comms, 'lead') + old_completion = time.time() - 3600 + services.update(lambda passes: DocumentChange.to(replace( + passes, commander_completed=passes.commander_completed | {owner: old_completion}))) + services.failed('CommanderHeartbeat', KeyError('original branch read failed')) + failure = services.read().failing['CommanderHeartbeat'] + hud = ThreadCaller('lead').invoke(CommsHudTool(), comms) + load = ThreadCaller('lead').invoke(CommsCommanderLoadTool(), comms) + assert hud.units == original.units and hud.load == original.load + assert hud.heartbeat == RecordedEventTime(old_completion) + assert load.load == original.load and load.heartbeat == hud.heartbeat + for lines in (hud.frame, load.header): + assert 'original branch read failed' in lines[0] + assert 'elapsed' in lines[0] + assert 'last completed' in lines[1] + assert any('STALE retained' in line for line in lines) + edits = ThreadCaller('lead').invoke(CommsUnjudgedTool(), comms) + assert 'original branch read failed' in edits.service_status[0] + assert edits.complete and edits.unavailable == () + assert len(edits.edits) == 1 and edits.edits[0].id == ToolCallId('parent-edit-0') + assert services.read().failing['CommanderHeartbeat'] == failure + assert services.read().commander_completed[owner] == old_completion + services.failed('CommanderHeartbeat', ValueError('newest failure')) + assert services.read().failing['CommanderHeartbeat'].since == failure.since + assert 'newest failure' in CommanderFrame.current(comms, 'lead').frame[0] + published_before = tuple((path, path.read_bytes()) for path in CommanderFrame.matching_in(comms.root)) + + async def recover(): + task = asyncio.create_task(Coordination.watch_worker( + CommanderHeartbeat(), lambda: nullcontext(comms), services)) + try: + async with asyncio.timeout(15): + while 'CommanderHeartbeat' in services.read().failing: + await asyncio.sleep(0.01) + finally: + task.cancel() + with suppress(asyncio.CancelledError): + await task + + asyncio.run(recover()) + recovered = CommanderFrame.current(comms, 'lead') + assert recovered.services.failing == {} + assert recovered.heartbeat.since(old_completion) + assert any(path.read_bytes() != before for path, before in published_before) + assert recovered.heartbeat == RecordedEventTime(services.read().commander_completed[owner]) + + +def test_stopped_heartbeat_without_failure_explains_stale_board_and_refuses_details(published_board): + comms = published_board + owner = comms.registry.require('lead').incarnation.require_recorded() + services = ServicePassStore.at(comms.root) + services.update(lambda passes: DocumentChange.to(replace( + passes, commander_completed=passes.commander_completed | {owner: time.time() - 3600}))) + hud = CommanderFrame.current(comms, 'lead') + assert not hud.services.failing + assert hud.header[0] == 'no failure recorded, so no heartbeat is running' + assert 'STALE retained' in hud.retained_status() + card = hud.units[0] + request = ReadCommanderUnit( + root=str(comms.root), commander=owner, unit=card.unit, + orders=tuple(dict.fromkeys(order.order for order in card.orders))) + with pytest.raises(ToolRefusedError, match='no failure recorded, so no heartbeat is running'): + request.read(comms) + + +def test_failing_watch_does_not_make_a_fresh_heartbeat_stale(published_board): + comms = published_board + services = ServicePassStore.at(comms.root) + owner = comms.registry.require('lead').incarnation.require_recorded() + services.heartbeat_passed((owner,), (owner,)) + services.failed('another-watch', OSError('independent watch failed')) + hud = CommanderFrame.current(comms, 'lead') + assert 'independent watch failed' in hud.header[0] + assert hud.retained_status() == 'Retained board rows and load:' + hud.require_current() + + +def test_missing_frame_reports_recorded_failure_without_fabricating_a_board(fork_bus): + comms, _, _ = fork_bus + services = ServicePassStore.at(comms.root) + services.failed('CommanderHeartbeat', KeyError('no readable publication')) + with pytest.raises(ToolRefusedError) as refused: + CommanderFrame.current(comms, 'lead') + lines = str(refused.value).splitlines() + assert 'no readable publication' in lines[0] + assert 'Board unavailable:' in lines[-1] + assert not CommanderFrame.matching_in(comms.root) + with pytest.raises(ToolRefusedError, match='no readable publication'): + CommanderFrame.read_all(comms.root) + + +def test_one_unreadable_source_preserves_the_other_original_edit(published_board, fork_bus, tmp_path): + from agent_comms.repo import Checkout + from agent_comms.storage_rules import StorageRule + from agent_comms.thread_execution import ExternalThreadExecution + from agent_comms.thread_model import DeclaredModel + from agent_comms.threads import Thread + from test_edit_allocations import issue + + comms, source, template = fork_bus + worker = comms.registry.declare(Thread( + 'unreadable-worker', frozenset({'comms'}), Checkout(source.checkout), + model=DeclaredModel('anthropic/claude-opus-4-6'), execution=ExternalThreadExecution)) + rule = StorageRule.load(comms.root) + checkout = rule.order_checkout(rule.roots_of(worker), 'P41') + checkout.parent.mkdir(parents=True, exist_ok=True) + source.git(source.checkout, 'worktree', 'add', '-q', '-b', 'unreadable-edit-source', str(checkout)) + issue(comms, source, replace(template, check=replace(template.check, directory=checkout)), + worker.name, 'P41') + metadata = checkout / '.git' + retained_metadata = tmp_path / 'unreadable-source-metadata' + metadata.rename(retained_metadata) + try: + result = ThreadCaller('lead').invoke(CommsUnjudgedTool(), comms) + finally: + retained_metadata.rename(metadata) + assert not result.complete + assert len(result.unavailable) == 1 and str(result.unavailable[0].problem) == 'P41' + assert tuple(edit.id for edit in result.edits) == (ToolCallId('parent-edit-0'),) diff --git a/tests/test_start.py b/tests/test_start.py index fdffe8353e..0888bd148e 100644 --- a/tests/test_start.py +++ b/tests/test_start.py @@ -57,4 +57,4 @@ def test_start_and_allocating_tools_keep_the_real_free_space_floor(tmp_path, mon with rule.room_for_start(thread, comms.root): pytest.fail("A stopped owner must not start below the volume floor") with pytest.raises(StorageDenied, match="10 GiB free"): - StorageProjection(rule, thread).check((AllocateStorageEffect(),)) + StorageProjection(rule, thread, comms).check((AllocateStorageEffect(),)) diff --git a/tests/test_storage_rules.py b/tests/test_storage_rules.py index b8959a5b94..cc03fcfab0 100644 --- a/tests/test_storage_rules.py +++ b/tests/test_storage_rules.py @@ -122,22 +122,23 @@ def test_removed_worktree_updates_membership_and_another_home_cannot_claim_its_p assert rule.owns_present_checkout(thread, home) assert not rule.owns_present_checkout(other, old) with pytest.raises(StorageDenied): - StorageProjection(rule, other).check(( + StorageProjection(rule, other, comms).check(( CreateCheckoutStorageEffect(new, Repo.of(home), DetachedStart('HEAD')),)) - StorageProjection(rule, thread).check(( + StorageProjection(rule, thread, comms).check(( CreateCheckoutStorageEffect(new, Repo.of(home), DetachedStart('HEAD')),)) - StorageProjection(rule, thread).check((RemoveCheckoutStorageEffect(old),)) + StorageProjection(rule, thread, comms).check((RemoveCheckoutStorageEffect(old),)) shutil.rmtree(old) assert rule.checkouts_of(thread) == (home,) - StorageProjection(rule, thread).check(( + StorageProjection(rule, thread, comms).check(( CreateCheckoutStorageEffect(new, Repo.of(home), DetachedStart('HEAD')),)) assert comms.registry.require('worker').storage_roots == thread.storage_roots def test_three_small_checkouts_are_admitted(checkout_comms: Comms): - rule = checkout_comms.registry.storage_rule + comms = checkout_comms + rule = comms.registry.storage_rule thread = checkout_comms.registry.require('worker') - projection = StorageProjection(rule, thread) + projection = StorageProjection(rule, thread, comms) for problem in ('P1', 'P2', 'P3'): path = rule.order_checkout(rule.roots_of(thread), problem) start = DetachedStart('HEAD') @@ -179,10 +180,10 @@ def test_cannot_remove_current_project_but_can_remove_another_checkout(tmp_path) assert thread.project_path == current assert UnitStorage.read(comms.registry.storage_rule, thread).home == Checkout(current) with pytest.raises(StorageDenied): - StorageProjection(comms.registry.storage_rule, comms.registry.require('worker')).check(( + StorageProjection(comms.registry.storage_rule, comms.registry.require('worker'), comms).check(( RemoveCheckoutStorageEffect(project_link.resolve()),)) assert current.is_dir() - StorageProjection(comms.registry.storage_rule, comms.registry.require('worker')).check(( + StorageProjection(comms.registry.storage_rule, comms.registry.require('worker'), comms).check(( RemoveCheckoutStorageEffect(other),)) subprocess.run(('git', '-C', str(repository), 'worktree', 'remove', str(other)), check=True) assert not other.exists() @@ -194,11 +195,11 @@ def test_cannot_remove_current_project_but_can_remove_another_checkout(tmp_path) 'git worktree remove --force target', 'python -m venv env', 'uv sync', 'git -C ~/wt/other--P3 reset --hard', 'touch ~/wt/x-check/file']) -def test_forbidden_shell_operations_use_declared_rule(tmp_path, command): +def test_forbidden_shell_operations_use_declared_rule(tmp_path, command, comms): rule = StorageRule() thread = Thread('worker', frozenset(), Checkout(tmp_path), storage_roots=rule.initial_roots('worker')) with pytest.raises(SelectedToolDenied): - StorageProjection(rule, thread).check(shell_storage_effects(command, tmp_path)) + StorageProjection(rule, thread, comms).check(shell_storage_effects(command, tmp_path)) def test_programs_do_not_claim_to_be_effect_values(): diff --git a/tests/test_unowned_checkout_work.py b/tests/test_unowned_checkout_work.py index 0e450c7152..d5fb20afc6 100644 --- a/tests/test_unowned_checkout_work.py +++ b/tests/test_unowned_checkout_work.py @@ -199,7 +199,7 @@ def test_a_concurrent_unstrike_does_not_end_the_heartbeat_pass(struck_unowned_bu fresh = c.goals.todos.apply(current.id, VoidStrikeTodo( current.revision, c.registry.require('lead').incarnation, strike, 'The cited failure was incorrect and the commander takes back this strike.')) - following = ServicePassFailing('following-rule', time.time(), 'independent source') + following = ServicePassFailing(watch='following-rule', since=time.time(), error='independent source') run_heartbeat_rules(c, (stale, following)) assigned = c.goals.todos.get('P1') @@ -226,7 +226,7 @@ def enforce(self, comms: Comms, commander: str) -> None: current = c.goals.todos.get('P1') stale = ConflictingServicePass('stale-rule', time.time(), 'changed row', current) - following = ServicePassFailing('following-rule', time.time(), 'independent source') + following = ServicePassFailing(watch='following-rule', since=time.time(), error='independent source') with LaunchRoot.from_path(c.root).selected(): run_heartbeat_rules(c, (stale, following)) update, = CommanderUpdate.latest(c, c.registry.require('lead').incarnation) @@ -256,7 +256,7 @@ def redecided(self, comms: Comms, commander: str, previous.revision, c.registry.require('lead').incarnation, strike, 'The cited failure was incorrect and the commander takes back this strike.')) stale = ChangingRow(previous, current) - following = ServicePassFailing('following-rule', time.time(), 'independent source') + following = ServicePassFailing(watch='following-rule', since=time.time(), error='independent source') with LaunchRoot.from_path(c.root).selected(): rejected = StandingRule.enforce_all(c, 'lead', (stale, following)) error, = rejected.values()