diff --git a/docs/maintenance-holder.md b/docs/maintenance-holder.md new file mode 100644 index 0000000000..7a0139ee92 --- /dev/null +++ b/docs/maintenance-holder.md @@ -0,0 +1,60 @@ +# Process-owned maintenance (P281) + +The barrier stores the holder's PID and kernel birth time through the existing +`ProcessIdentity`/`FieldCodec` implementation. Names are not identities. READY +records describe completed work and do not depend on the old operator staying +alive. + +A non-READY read checks that exact process. Recovery compares the observed +record under the existing registry lock, writes `dead_holder_released` with the +dead PID/birth time, and refuses the detecting input. The following input uses +READY behavior. The asynchronous ingress borrows the asynchronous registry +lock; neither path upgrades an admitted shared wire descriptor. + +Live installs run in a separate session with private file-backed output, so +losing the calling session does not kill the installer or break its output +pipe. The worker records its own identity. Rehearsals remain synchronous. + +`TodoStore.writer_connection` admits all todo writers before opening SQLite, +including commander first-call, hunt, overdue and reconciliation writes. + +One registered carry converts version-1 READY witnesses. An old active record +contains only a name: its holder's death cannot be established. The carry +refuses that record rather than fabricating a PID or retaining a name fallback. +The runtime reads only version 2. + +## Delivery checklist + +- [x] Implementation checkpoint pushed on `fix/p281-maintenance-holder`. +- [x] Production AST inventory: 381 Core modules and 39 tools; no parse omissions. +- [x] Migrate all discovered producer and test callers off string operators. +- [x] Eleven focused real-process/ingress/todo/carry checks passed. +- [ ] Tools ratchet and src ratchet against `origin/main`. +- [x] Copied-bus live-phase proof with existing Python, without a build, + environment creation or `uv sync`. +- [x] Calling-session death left the worker alive; SIGKILL of its job mid-drain + gave a named `held` refusal, then an accepted and retained second message. + The release record retained the dead PID/birth time; original bus unchanged. +- [x] Draft PR #2072 opened through the commander's canonical PR command. + +## Installation wiring follow-up + +Normal and unchanged live installs now use the selected interpreter to refresh +`RuntimePublication`. That existing owner publishes console/skill links and +calls the declared `Harness.configure` implementations; repeated refreshes are +safe. Consumers no longer skip refresh merely because the pointer is current. + +Actual installed programs, with a private HOME and candidate publication code, +verified every declared hook on PATH/configuration, including P275 SessionStart +and P285 UserPromptSubmit. SessionStart exited zero with all 45 packaged files; +repeated refresh preserved user settings without duplicate hooks. Production +configuration and services were not changed. + +This does not yet fence external bridge watches or session hooks across mutable +checkout/schema updates. The reported transitional-reader failure remains +unfixed; publication success is not acceptance of that separate lifecycle gap. + +The inventory includes declarations, inheritance, imports, writes, decisions +and references throughout both production roots. Generated test child programs +are additionally located by source search; AST does not resolve their runtime +imports or prove behavior. diff --git a/plans/commander-native-plan.md b/plans/commander-native-plan.md index 05042371b2..4fbae94e3a 100644 --- a/plans/commander-native-plan.md +++ b/plans/commander-native-plan.md @@ -61,7 +61,7 @@ | 28 | Measured load and self-report | `capacity.py` unit count | `commander.capacity.Load.of` reports orders, overdue, edits and forks; self-report comparison unverified | `control_plane` | | 29 | No handoff of finished, checked work | none; scratch transfer moved P507 after its fix was pushed and checked | Transfer refuses delivered checked work; that owner merges; check is reused | `record_plane` | | 30 | A proof is bound to the build it ran on | none; screenshots and logs are bare paths | Build run records runtime and Core/Toad revisions; close requires delivered commit | `record_plane` | -| 31 | An order states its intent as a deletion target | none; P520 renamed a decision rather than deleting it | Order owns deletion family/site/count; delivery refuses nonzero or renamed decisions | `record_plane` | +| 31 | An order states its intent as a deletion target | Current order command with a declared FamilyDeletion | CommsTodoCreateTool and CommsOrderDoneTool check the delivered family count | `record_plane` | | 32 | Where units' time goes, classified | none; activity.log lacks durations and turn boundaries | `UnitTime`: model, build/check/test/shell, dependency/commander waits and idle; classify large waits | `commander_page`, `control_plane` | | 33 | Order progress beside its deadline; the unit never sees a due time | `order.py board`, `overdue`, `due`, `send`; `rotate.py` | Derived `OrderProgress`; merged work draws no strike; first commit 10 min, later progress 15 min STALL; units see no due time | `one_order_owner`, `control_plane` | | 34 | Several commanders on one machine and bus, one per project | none; a second commander.json install overwrites the first | Registry commander/project ownership; caller from --from; scoped orders, strikes, load, board, alarms and forks; raw-send guard; duplicate project commander refused | `one_order_owner`, `cutover` | diff --git a/src/agent_comms/cli_bridges/codex_protocol.py b/src/agent_comms/cli_bridges/codex_protocol.py index 7033caeab3..9c15543908 100644 --- a/src/agent_comms/cli_bridges/codex_protocol.py +++ b/src/agent_comms/cli_bridges/codex_protocol.py @@ -742,7 +742,7 @@ def publication(self) -> RuntimeUpdate: if self.completedAtMs is None: raise ValueError(f'Item {self.item.id} has not finished') event_time = self.started_at.continued_from(self.completedAtMs / 1000) - return self.item.completed(self.turnId, self.item.events(event_time.transcript_timestamp())) + return self.item.completed(self.turnId, self.item.events(event_time.recorded_at('A Codex item'))) class ItemsPage(WireModel): diff --git a/src/agent_comms/cli_protocol.py b/src/agent_comms/cli_protocol.py index ff7c949703..9d83f38dce 100644 --- a/src/agent_comms/cli_protocol.py +++ b/src/agent_comms/cli_protocol.py @@ -17,7 +17,7 @@ from .maintenance_barrier import MaintenanceHeldError from .messages import Message from .field_codec import TimestampText -from .recorded_time import MissingTranscriptTimestampError +from .recorded_time import UnrecordedTimeError from .external_transcript import RequestedEarlierPage if TYPE_CHECKING: @@ -45,7 +45,7 @@ class TimestampedWire(WireModel): @classmethod def require_source_time(cls, value): if isinstance(value, dict) and value.get('timestamp') is None: - raise MissingTranscriptTimestampError(f'{cls.__name__} has no recorded timestamp') + raise UnrecordedTimeError(f'{cls.__name__} has no recorded timestamp') return value @field_validator('timestamp', mode='before') diff --git a/src/agent_comms/commander/capacity.py b/src/agent_comms/commander/capacity.py index dfa109aca5..415d057940 100644 --- a/src/agent_comms/commander/capacity.py +++ b/src/agent_comms/commander/capacity.py @@ -22,14 +22,17 @@ class Load(ToolResult): def of(cls, comms: Comms, commander: str) -> Load: snapshot, todos, now = comms.registry.snapshot(), comms.goals.todos, time.time() owner = snapshot.require(commander) - owed = tuple(t for t in todos.held_for_commander(owner.incarnation) if t.pending) + held = todos.held_for_commander(owner.incarnation) + owed = tuple(t for t in held if t.pending) forks = tuple(sorted( thread.name for thread in snapshot.threads.values() if thread.parent == owner.name and snapshot.status(thread.name).receiving and todos.held_for_commander(thread.incarnation) )) - return cls(len(owed), sum(t.overdue(now) for t in owed), - OrderEditCountRead.count(comms, owner.name), forks) + unjudged = sum( + read.read() for read in OrderEditCountRead.for_orders(comms.root, snapshot, held) + ) + return cls(len(owed), sum(t.overdue(now) for t in owed), unjudged, forks) @projected(view="wire") def total(self) -> float: diff --git a/src/agent_comms/commander/order.py b/src/agent_comms/commander/order.py index 6441f23709..1834b631b3 100644 --- a/src/agent_comms/commander/order.py +++ b/src/agent_comms/commander/order.py @@ -742,8 +742,8 @@ def due(self, pid: str, at: str, why: str) -> None: moved = RecordedEventTime(time.time()).at_clock(at) self.call(CommsChangeProblemTool(problem=pid, operation=AssignTodo( todo.revision, todo.assignment, moved, todo.order, todo.goal))) - self.on_record(f'{time.strftime("%Y-%m-%d %H:%M")} {todo.holder} ({pid} order {todo.order.id} due ' - f'{todo.deadline.clock()} moved to {moved.clock()}): {why}', 'Deadline', + self.on_record(f'{time.strftime("%Y-%m-%d %H:%M")} {todo.holder} ({pid} order {todo.order.id} ' + f'deadline changed): {why}', 'Deadline', VerdictAudience.of(todo.held().owner, todo.goal)) print(f'{pid}: due {moved.clock()}') diff --git a/src/agent_comms/commander_parity/deletion.py b/src/agent_comms/commander_parity/deletion.py new file mode 100644 index 0000000000..78834661c6 --- /dev/null +++ b/src/agent_comms/commander_parity/deletion.py @@ -0,0 +1,261 @@ +"""Deletion intent parity through real orders, pushed commits and native delivery checks.""" + +from __future__ import annotations + +import contextlib +import io +import json +import shlex +import sys +import tempfile +from abc import ABC, abstractmethod +from dataclasses import dataclass, replace +from pathlib import Path + +from ..change_report import FileLine +from ..child_process import CountPrinted, ExitedOutcome +from ..commander.order import OrderFile, deadline +from ..field_codec import FieldCodec +from ..goal_actions import CreateProblemGoalAction +from ..goals import CountCheckResult, OrderCheck +from ..plan_phases import RecordPlanePhase +from ..progress_claims import MergedClaim +from ..repo import Repo +from ..todos import ( + CheckedCommitDelivery, + CommitDelivery, + DeliveredProblemOrder, + FamilyDeletion, + MechanicalOrderRun, + Todo, +) +from ..tool_caller import ThreadCaller +from ..tools import CommsOrderDoneTool, CommsTodoCreateTool, ProblemOrdered +from . import ParityComparison, ParityRow +from .lifecycle import LifecycleObservation, LifecycleRun, MeasuredCommander + + +@dataclass(frozen=True) +class DeletionComparison(ParityComparison): + current: LifecycleObservation + native: LifecycleObservation + + @property + def passed(self) -> bool: + return ( + self.current.passed and self.native.passed and self.current.checks == self.native.checks + ) + + def describe(self) -> str: + return f"current: {self.current.describe()}; native: {self.native.describe()}" + + @classmethod + def read(cls) -> DeletionComparison: + return cls(CurrentDeletionPath.read(), NativeDeletionPath.read()) + + +class DeletionPath(ABC): + """Both adapters exercise the same scenario on independent LifecycleRun copies.""" + + @classmethod + @abstractmethod + def send(cls, run: LifecycleRun, document: OrderFile) -> Todo: ... + + @classmethod + @abstractmethod + def delivery_command(cls, run: LifecycleRun, todo: Todo, commit: str) -> tuple[str, ...]: ... + + @staticmethod + def commit(repository: Path, text: str, message: str) -> str: + (repository / "family.py").write_text(text) + Repo.git(repository, "add", "family.py") + Repo.git(repository, "commit", "-qm", message) + Repo.git(repository, "push", "-q", "origin", "main") + return Repo.git(repository, "rev-parse", "HEAD").strip() + + @classmethod + def read(cls) -> LifecycleObservation: + with LifecycleRun.open() as original: + # Keep source checkouts out of the copied worker's storage accounting. + # Limits remain unchanged; all detached checks use this persistent owned base. + rule = original.commander.comms.registry.storage_rule + with tempfile.TemporaryDirectory( + prefix="parity-deletion-", dir=rule.checkout_base + ) as checkouts: + root = original.commander.root + (root / "storage.json").write_text( + json.dumps(FieldCodec.encode(replace(rule, checkout_base=Path(checkouts)))) + ) + (root / "commander.json").write_text(original.commander.model_dump_json()) + run = replace(original, commander=MeasuredCommander.read_root(root)) + return cls.exercise(run) + + @classmethod + def exercise(cls, run: LifecycleRun) -> LifecycleObservation: + repository = Path(run.commander.comms.registry.require(run.worker).worktree) + remote = repository.parent / "deletion-origin.git" + Repo.git(repository, "init", "-q", "-b", "main") + Repo.git(repository, "config", "user.name", "Deletion parity") + Repo.git(repository, "config", "user.email", "deletion-parity@localhost") + Repo.git(repository, "init", "--bare", "-q", "--initial-branch=main", str(remote)) + Repo.git(repository, "remote", "add", "origin", str(remote)) + family = "class RetiredDecision:\n pass\n\n" + cls.commit( + repository, + family + "def decisions():\n return [RetiredDecision()]\n", + "Declare the obsolete decision family and its caller", + ) + count = OrderCheck( + str(repository), + shlex.join( + ( + sys.executable, + "-c", + "import ast;from pathlib import Path;" + "print(sum(isinstance(node,ast.ClassDef) for node in " + "ast.walk(ast.parse(Path('family.py').read_text()))))", + ) + ), + CountCheckResult(0), + ) + deletion = FamilyDeletion( + "obsolete decision family", FileLine(str(repository / "family.py"), 1), count + ) + initial = OrderFile.read(run.document.read_text()) + order = replace( + initial.order, + check=OrderCheck( + str(repository), + shlex.join((sys.executable, "-c", "import family;assert not family.decisions()")), + ), + run=MechanicalOrderRun(), + deletes=deletion, + ) + document = replace(initial, order=order) + run.document.write_text(json.dumps(FieldCodec.encode(document))) + output = io.StringIO() + with contextlib.redirect_stdout(output): + before = cls.send(run, document) + partial = cls.commit( + repository, + family + "def decisions():\n return []\n", + "Repair the caller but leave one obsolete family site", + ) + refused = run.invoke_command( + cls.delivery_command(run, before, partial), env=run.python_environment() + ) + after_refusal = run.commander.comms.goals.todos.get(before.id) + remaining = count.recorded(partial, str(repository)) + complete = cls.commit( + repository, + "def decisions():\n return []\n", + "Delete the last obsolete family declaration", + ) + accepted = run.invoke_command( + cls.delivery_command(run, before, complete), env=run.python_environment() + ) + after_acceptance = run.commander.comms.goals.todos.get(before.id) + zero = count.recorded(complete, str(repository)) + primary = order.check.recorded(partial, str(repository)) + delivered = after_acceptance.order + checks = ( + ("native order retained deletion intent", before.order.checked().deletes == deletion), + ( + "partial commit passed primary check", + bool(primary) and order.check.result.distance(primary[0]) == CountPrinted(0), + ), + ( + "partial commit retained one family site", + bool(remaining) and count.result.distance(remaining[0]) == CountPrinted(1), + ), + ("nonzero delivery refused", refused.outcome == ExitedOutcome(1)), + ("refusal preserved pending order", before == after_refusal and after_refusal.pending), + ( + "zero count checked on delivered commit", + bool(zero) and count.result.distance(zero[0]) == CountPrinted(0), + ), + ( + "zero delivery accepted", + accepted.outcome == ExitedOutcome(0) and not after_acceptance.pending, + ), + ( + "accepted delivery names the zero-site commit", + isinstance(delivered, DeliveredProblemOrder) + and isinstance(delivered.delivery, CheckedCommitDelivery) + and delivered.delivery.delivered_commit == complete, + ), + ("accepted delivery retained deletion intent", delivered.checked().deletes == deletion), + ) + return LifecycleObservation( + CommsOrderDoneTool.__qualname__, + len((refused, accepted)), + checks, + output.getvalue() + f"partial={partial}; zero={complete}; " + f"refused={refused.outcome.summary}: " + f"{refused.stdout.decode(errors='replace')}{refused.stderr.decode(errors='replace')}; " + f"accepted={accepted.outcome.summary}: " + f"{accepted.stdout.decode(errors='replace')}{accepted.stderr.decode(errors='replace')}", + ) + + +class CurrentDeletionPath(DeletionPath): + @classmethod + def send(cls, run: LifecycleRun, document: OrderFile) -> Todo: + return run.create() + + @classmethod + def delivery_command(cls, run: LifecycleRun, todo: Todo, commit: str) -> tuple[str, ...]: + return ( + sys.executable, + "-c", + "import sys;from agent_comms.commander.order import Commander,DoneVerb;" + "DoneVerb(unit=sys.argv[2],message=sys.argv[3],sha=sys.argv[4]).run(" + "Commander.read_root(sys.argv[1]))", + str(run.commander.root), + run.worker, + todo.order.checked().id, + commit, + ) + + +class NativeDeletionPath(DeletionPath): + @classmethod + def send(cls, run: LifecycleRun, document: OrderFile) -> Todo: + result = run.commander.call( + CommsTodoCreateTool( + owner=run.worker, + order=CreateProblemGoalAction( + problem="P1", + repo=Repo.of(Path(document.order.check.directory)).github, + text=document.defect, + order=document.order, + request=document.order.request( + run.commander.comms.registry.require(run.commander.commander), ("P1",) + ), + deadline=deadline(run.due), + commander=run.commander.commander, + ), + ) + ) + if not isinstance(result, ProblemOrdered): + raise TypeError("Native creation returned no ordered problem") + return result.todo + + @classmethod + def delivery_command(cls, run: LifecycleRun, todo: Todo, commit: str) -> tuple[str, ...]: + request = CommsOrderDoneTool( + problem=todo.id, + order_id=todo.order.checked().id, + delivery=CommitDelivery(commit=commit, delivered_commit=commit), + claims=(MergedClaim(commit),), + ) + return request.cli_command(ThreadCaller(run.commander.commander)).argv(run.commander.root) + + +class DeletionTargetParityRow(ParityRow): + number = 31 + capability = "An order states its intent as a deletion target" + current = "Current order command with a declared FamilyDeletion" + native = "CommsTodoCreateTool and CommsOrderDoneTool check the delivered family count" + phases = (RecordPlanePhase,) + comparison = DeletionComparison.read diff --git a/src/agent_comms/commander_parity/delivery.py b/src/agent_comms/commander_parity/delivery.py new file mode 100644 index 0000000000..0026115cd7 --- /dev/null +++ b/src/agent_comms/commander_parity/delivery.py @@ -0,0 +1,211 @@ +"""Compare commit delivery through the current command and the native tool.""" + +from __future__ import annotations + +import contextlib +import io +import json +import subprocess +from dataclasses import dataclass, replace +from pathlib import Path + +from ..child_process import ChildOutcome, ChildResult, ExitedOutcome +from ..commander.order import OrderFile +from ..field_codec import FieldCodec +from ..progress_claims import MergedClaim +from ..todos import CommitDelivery, FileDeliveryEvidence, Todo +from ..tools import CommsOrderDoneTool +from . import ParityComparison +from .lifecycle import LifecycleRun + + +@dataclass(frozen=True) +class DeliveryOutcome: + outcome: ChildOutcome + pending: bool + order_line: str + refusal: str + + @classmethod + def saved(cls, todo: Todo, process: ChildResult) -> DeliveryOutcome: + return cls( + process.outcome, todo.pending, todo.order.line, process.stderr.decode(errors="replace") + ) + + @property + def delivered(self) -> bool: + return self.outcome.successful and not self.pending + + +@dataclass(frozen=True) +class DeliveryComparison(ParityComparison): + checked_commit: str + cases: tuple[tuple[str, str, DeliveryOutcome, DeliveryOutcome], ...] + + @property + def passed(self) -> bool: + return ( + len(self.cases) == 3 + and self.cases[0][2].delivered + and all( + (current.outcome, current.pending) == (native.outcome, native.pending) + and current.order_line == native.order_line + and ( + all( + outcome.outcome == ExitedOutcome(1) + and commit in outcome.refusal + and "not in" in outcome.refusal + for outcome in (current, native) + ) + if current.pending + else commit in current.order_line + ) + for _, commit, current, native in self.cases + ) + ) + + def describe(self) -> str: + return f"checked main={self.checked_commit}; outcomes={self.cases!r}" + + +@dataclass(frozen=True) +class DeliveryRepository: + path: Path + + def git(self, *arguments: str) -> str: + return subprocess.check_output( + ( + "git", + "-C", + str(self.path), + "-c", + "user.name=parity", + "-c", + "user.email=parity@invalid", + *arguments, + ), + text=True, + stderr=subprocess.PIPE, + ).strip() + + @classmethod + def create(cls, root: Path) -> DeliveryRepository: + path = root / "delivery-repository" + path.mkdir() + repo = cls(path) + remote = root / "delivery-origin.git" + subprocess.run(("git", "init", "--bare", "-q", str(remote)), check=True) + repo.git("init", "-q", "-b", "main") + url = "https://github.com/parity-local/delivery.git" + repo.git("config", f"url.{remote}.insteadOf", url) + repo.git("remote", "add", "origin", url) + repo.git("commit", "-q", "--allow-empty", "-m", "copied delivery base") + repo.git("push", "-q", "origin", "main") + repo.git("symbolic-ref", "refs/remotes/origin/HEAD", "refs/remotes/origin/main") + return repo + + def cases(self) -> tuple[tuple[str, str], ...]: + base = self.git("rev-parse", "HEAD") + self.git("switch", "-q", "-c", "main-answer") + self.git("commit", "-q", "--allow-empty", "-m", "answer integrated on main") + main = self.git("rev-parse", "HEAD") + self.git("switch", "-q", "main") + self.git("merge", "-q", "--no-ff", "main-answer", "-m", "integrate main answer") + self.git("push", "-q", "origin", "main") + self.git("switch", "-q", "-c", "branch-answer", base) + self.git("commit", "-q", "--allow-empty", "-m", "answer integrated elsewhere") + branch = self.git("rev-parse", "HEAD") + self.git("switch", "-q", "-c", "integration", base) + self.git("merge", "-q", "--no-ff", "branch-answer", "-m", "integrate branch answer") + self.git("push", "-q", "origin", "integration") + self.git("switch", "-q", "-c", "unmerged-answer", base) + self.git("commit", "-q", "--allow-empty", "-m", "published but not integrated") + unmerged = self.git("rev-parse", "HEAD") + self.git("push", "-q", "origin", "unmerged-answer") + self.git("switch", "-q", "main") + return ( + ("merged to main", main), + ("merged to another branch", branch), + ("not merged", unmerged), + ) + + +class DeliveredCommitCase: + @classmethod + def create_order(cls, run: LifecycleRun, repository: DeliveryRepository) -> Todo: + document = OrderFile.read(run.document.read_text()) + document = replace( + document, + order=replace( + document.order, check=replace(document.order.check, directory=str(repository.path)) + ), + ) + run.document.write_text(json.dumps(FieldCodec.encode(document))) + run.decisions.write_text(json.dumps([run.worker])) + before = { + todo.id + for todo in run.commander.comms.goals.todos.for_commander(run.commander.incarnation) + } + run.commander.problem( + str(run.document), run.worker, run.due, f"{Path(__file__).resolve()}:1" + ) + (created,) = ( + todo + for todo in run.commander.comms.goals.todos.for_commander(run.commander.incarnation) + if todo.id not in before + ) + run.decisions.write_text("[]") + return created + + @classmethod + def read(cls) -> ParityComparison: + cases = [] + with LifecycleRun.open() as run, contextlib.redirect_stdout(io.StringIO()): + repository = DeliveryRepository.create(run.commander.root) + run.decisions.write_text("[]") + run.artifact.write_text( + "Copied-bus delivered-commit comparison; no source input executed." + ) + for name, commit in repository.cases(): + current_order = cls.create_order(run, repository) + process = run.invoke( + "from agent_comms.commander.order import Commander; import sys; " + "Commander.model_validate_json(sys.argv[1]).done(" + "sys.argv[2],sys.argv[3],seen=sys.argv[4])", + run.commander.model_dump_json(), + current_order.id, + commit, + str(run.artifact), + ) + current = DeliveryOutcome.saved( + run.commander.comms.goals.todos.get(current_order.id), process + ) + native_order = cls.create_order(run, repository) + tool = CommsOrderDoneTool( + problem=native_order.id, + order_id=native_order.order.checked().id, + delivery=CommitDelivery( + commit=repository.git("rev-parse", "HEAD"), + delivered_commit=commit, + evidence=FileDeliveryEvidence(str(run.artifact)), + ), + claims=(MergedClaim(commit),), + ) + process = run.invoke( + "from pathlib import Path; import sys; " + "from agent_comms.comms import Comms; " + "from agent_comms.field_codec import FieldCodec; " + "from agent_comms.tool_caller import ThreadCaller; " + "from agent_comms.tools import CommsOrderDoneTool; " + "ThreadCaller(sys.argv[2]).invoke(FieldCodec.decode(CommsOrderDoneTool, " + "FieldCodec.parse(sys.argv[3])),Comms(Path(sys.argv[1])))", + str(run.commander.root), + run.commander.commander, + json.dumps(FieldCodec.encode(tool)), + ) + native = DeliveryOutcome.saved( + run.commander.comms.goals.todos.get(native_order.id), process + ) + cases.append((name, commit, current, native)) + checked_commit = repository.git("rev-parse", "HEAD") + return DeliveryComparison(checked_commit, tuple(cases)) diff --git a/src/agent_comms/commander_parity/handoff.py b/src/agent_comms/commander_parity/handoff.py new file mode 100644 index 0000000000..4faa6803ca --- /dev/null +++ b/src/agent_comms/commander_parity/handoff.py @@ -0,0 +1,152 @@ +"""Actual current/native refusals to transfer delivered checked work.""" + +from __future__ import annotations + +import contextlib +import io +import json +import time +from dataclasses import dataclass, replace +from pathlib import Path +from uuid import uuid4 + +from ..change_report import FileLine +from ..child_process import ChildOutcome, ChildResult, ExitedOutcome +from ..commander.config import CommanderConfig +from ..commander.order import OrderFile +from ..recorded_time import RecordedEventTime +from ..field_codec import FieldCodec +from ..goal_actions import OrderGoalAction +from ..tools import CommsOrderTool +from . import ParityComparison, UnverifiedParity +from .lifecycle import LifecycleRun, MeasuredCommander + + +@dataclass(frozen=True) +class HandoffComparison(ParityComparison): + cases: tuple[ + tuple[str, ChildOutcome, ChildOutcome, tuple[str, ...], tuple[str, ...], bool], ... + ] + + @property + def passed(self) -> bool: + return len(self.cases) == 2 and all( + all( + ( + expected == ExitedOutcome(1), + actual == ExitedOutcome(1), + bool(reference), + reference == refusal, + unchanged, + ) + ) + for _, expected, actual, reference, refusal, unchanged in self.cases + ) + + def describe(self) -> str: + return f"actual finished-order transfer refusals (current/native): {self.cases!r}" + + +class FinishedHandoffCase: + @staticmethod + def state(run: LifecycleRun, pid: str, receiver: str) -> str: + return json.dumps( + FieldCodec.encode( + ( + run.commander.comms.goals.todos.get(pid), + run.commander.comms.registry.require(run.worker).goal, + run.commander.comms.registry.require(receiver).goal, + ) + ), + sort_keys=True, + ) + + @staticmethod + def current(run: LifecycleRun, receiver: str, document: OrderFile, pid: str) -> ChildResult: + return run.invoke( + "import sys; from agent_comms.commander.order import Commander, OrderFile; " + "Commander.model_validate_json(sys.argv[1]).send(sys.argv[2], sys.argv[3], " + "OrderFile.read(sys.argv[4]), reassign=sys.argv[5], traced=sys.argv[6])", + run.commander.model_dump_json(), + receiver, + run.due, + json.dumps(FieldCodec.encode(document)), + "Compare the actual refusal to hand completed checked work to another real unit.", + f"{Path(__file__).resolve()}:1", + ) + + @staticmethod + def native(run: LifecycleRun, receiver: str, document: OrderFile, pid: str) -> ChildResult: + context = run.commander + todo = context.comms.goals.todos.get(pid) + tool = CommsOrderTool( + owner=receiver, + order=OrderGoalAction( + problem=pid, + expected_revision=todo.revision, + assigner=context.incarnation, + order=document.order, + deadline=RecordedEventTime(time.time()).at_clock(run.due), + commander=context.commander, + request=document.order.request( + context.comms.registry.require(context.commander), (pid,) + ), + traced=FileLine.from_wire(f"{Path(__file__).resolve()}:1"), + ), + ) + return run.invoke( + "from pathlib import Path; import sys; from agent_comms.comms import Comms; " + "from agent_comms.field_codec import FieldCodec; " + "from agent_comms.tool_caller import ThreadCaller; " + "from agent_comms.tools import CommsOrderTool; " + "ThreadCaller(sys.argv[2]).invoke(FieldCodec.decode(CommsOrderTool, " + "FieldCodec.parse(sys.argv[3])), Comms(Path(sys.argv[1])))", + str(context.root), + context.commander, + json.dumps(FieldCodec.encode(tool)), + ) + + @classmethod + def read(cls) -> ParityComparison: + authority = MeasuredCommander.read(CommanderConfig.path()) + owners = tuple( + dict.fromkeys( + todo.held().owner + for todo in authority.comms.goals.todos.held_for_commander(authority.incarnation) + ) + )[-2:] + if len(owners) < 2: + return UnverifiedParity("Finished-work transfer needs two real registered unit holders") + cases = [] + for label, execute in (("current", cls.current), ("native", cls.native)): + with LifecycleRun.open(extra_threads=owners) as run: + with contextlib.redirect_stdout(io.StringIO()): + receiver = next(owner for owner in owners if owner != run.worker) + delivered = run.complete(run.create()) + before = cls.state(run, delivered.id, receiver) + document = OrderFile.read(run.document.read_text()) + document = replace( + document, + problems=(delivered.id,), + order=replace(document.order, id=uuid4().hex), + ) + run.decisions.write_text(json.dumps([receiver])) + reference = run.invoke( + "from pathlib import Path; import sys; from agent_comms.comms import Comms; " + "todo=Comms(Path(sys.argv[1])).goals.todos.get(sys.argv[2]); " + "todo.order.require_transfer(todo)", + str(run.commander.root), + delivered.id, + ) + attempt = execute(run, receiver, document, delivered.id) + cases.append( + ( + label, + reference.outcome, + attempt.outcome, + tuple(reference.stderr.decode(errors="replace").splitlines()[-1:]), + tuple(attempt.stderr.decode(errors="replace").splitlines()[-1:]), + before == cls.state(run, delivered.id, receiver), + ) + ) + return HandoffComparison(tuple(cases)) diff --git a/src/agent_comms/commander_parity/lifecycle.py b/src/agent_comms/commander_parity/lifecycle.py index 2c0e441cb0..271de82cf3 100644 --- a/src/agent_comms/commander_parity/lifecycle.py +++ b/src/agent_comms/commander_parity/lifecycle.py @@ -2,11 +2,13 @@ from __future__ import annotations +import asyncio import contextlib import io import json import os import shlex +import subprocess import sys import time from abc import ABC, abstractmethod @@ -20,16 +22,22 @@ from ..change_report import FileLine from ..check_root import CheckRoot +from ..child_process import BoundedRun, ChildResult from ..commander.config import CommanderConfig -from ..commander.order import Commander, OrderFile, hhmm +from ..commander.order import Commander, OrderFile from ..commander.request import RequestAddVerb from ..field_codec import FieldCodec from ..goals import CountCheckResult, OrderCheck, ProblemGoalOrder +from ..launch_runtime import ProofBuildMismatchError, RecordedBuildIdentity +from ..progress_claims import WorkingClaim from ..todos import ( ApplicationOrderRun, + DeliveredProblemOrder, + FileDeliveryEvidence, MergedProgress, Moving, NoCommitProgress, + OperationDelivery, PendingProblemOrder, Todo, ) @@ -38,6 +46,7 @@ from ..tools import ( CommsChangeProblemTool, CommsFireTool, + CommsOrderDoneTool, CommsOrderTool, CommsProblemCloseTool, CommsTodoCreateTool, @@ -123,7 +132,6 @@ def open(cls, *, extra_threads: tuple[str, ...] = ()) -> Iterator[LifecycleRun]: commander = MeasuredCommander( commander=config.commander, root=root, - cli=config.cli, state=root / "commander-state", board=root / "board.md", narrative=root / "narrative.md", @@ -180,6 +188,21 @@ def open(cls, *, extra_threads: tuple[str, ...] = ()) -> Iterator[LifecycleRun]: ) yield cls(commander, worker.name, decisions, document, root / "application.json") + @staticmethod + def python_environment() -> dict[str, str]: + return {**os.environ, "PYTHONPATH": str(Path(__file__).resolve().parents[2])} + + @classmethod + def invoke(cls, program: str, *arguments: str) -> ChildResult: + return cls.invoke_command( + (sys.executable, "-c", program, *arguments), env=cls.python_environment() + ) + + @staticmethod + def invoke_command(command: tuple[str, ...], **options) -> ChildResult: + """BoundedRun.run owns the process options and their absent defaults.""" + return asyncio.run(BoundedRun.run(command, timeout=120, **options)) + @property def due(self) -> str: return time.strftime("%H:%M", time.localtime(time.time() + 3600)) @@ -325,12 +348,77 @@ def checks( class BuildBoundProofCase(InstalledEvidenceCase): @classmethod - def read(cls) -> ParityComparison: - closing = super().read() - return UnverifiedParity( - "Missing native owner: a build-bound application-run record linking the " - "runtime and Core/Toad revisions to closing evidence. The actual close " - "owner currently accepts a bare artifact path. " + closing.describe() + def exercise(cls, run: LifecycleRun) -> tuple[Todo, ...]: + delivered, closed = super().exercise(run) + stale_run = replace(run, artifact=run.artifact.with_name("older-build-proof.json")) + stale_run.decisions.write_text(json.dumps([stale_run.worker])) + created = stale_run.create() + installed = RecordedBuildIdentity.installed() + older = subprocess.check_output( + ("git", "rev-parse", f"{installed.core}^"), + cwd=created.order.checked().check.directory, + text=True, + ).strip() + stale_run.artifact.write_text( + json.dumps( + FieldCodec.encode(stale_run.commander.comms.registry.require(stale_run.worker).goal) + ) + ) + stale_run.decisions.write_text("[]") + stale_run.commander.call( + CommsOrderDoneTool( + problem=created.id, + order_id=created.order.checked().id, + delivery=OperationDelivery( + evidence=FileDeliveryEvidence(str(stale_run.artifact)), + proof_build=replace(installed, core=older), + ), + claims=(WorkingClaim(str(stale_run.artifact)),), + ) + ) + stale = stale_run.commander.comms.goals.todos.get(created.id) + try: + stale_run.commander.close( + stale.id, + "Read the copied native goal and confirm its recorded source location " + "and assigned worker.", + str(stale_run.artifact), + ) + except ProofBuildMismatchError as error: + print(error) + return delivered, closed, stale, stale_run.commander.comms.goals.todos.get(stale.id) + + @classmethod + def checks( + cls, run: LifecycleRun, rows: tuple[Todo, ...], native: tuple[ToolResult, ...], output: str + ) -> tuple[tuple[str, bool], ...]: + delivered, closed, stale, after = rows + assert isinstance(delivered.order, DeliveredProblemOrder) + assert isinstance(stale.order, DeliveredProblemOrder) + current_build = delivered.order.delivery.proof_build + older_build = stale.order.delivery.proof_build + assert isinstance(current_build, RecordedBuildIdentity) + assert isinstance(older_build, RecordedBuildIdentity) + return ( + ("installed proof closes", closed.order.accepted and not closed.state.open), + ( + "build recorded durably", + FieldCodec.decode(Todo, FieldCodec.encode(stale)) == stale, + ), + ("older proof refused without mutation", stale == after and after.state.open), + ( + "both revisions named", + all( + revision in output + for revision in ( + current_build.core, + current_build.toad, + older_build.core, + older_build.toad, + ) + ), + ), + ("native close returned", bool(native)), ) @@ -414,16 +502,16 @@ def checks( ( "unit messages hide clocks", all( - hhmm(clock) not in message.body + clock.clock() not in message.body for clock in (before.deadline, after.deadline) for message in messages ), ), ( "unit goal hides deadline", - "Deadline:" not in instructions and hhmm(after.deadline) not in instructions, + "Deadline:" not in instructions and after.deadline.clock() not in instructions, ), - ("commander still sees deadline", f"due {hhmm(after.deadline)}" in output), + ("commander still sees deadline", f"due {after.deadline.clock()}" in output), ) diff --git a/src/agent_comms/commander_parity/load.py b/src/agent_comms/commander_parity/load.py index 66ec536cef..7f131371ae 100644 --- a/src/agent_comms/commander_parity/load.py +++ b/src/agent_comms/commander_parity/load.py @@ -2,42 +2,18 @@ from __future__ import annotations -import subprocess +import asyncio from dataclasses import dataclass from pathlib import Path +from typing import ClassVar +from ..child_process import BoundedRun from ..commander.config import CommanderConfig from ..field_codec import FieldCodec +from ..launch_runtime import LaunchRuntime from ..private_path import FileRevision from . import ParityComparison, UnverifiedParity - -# Profile the real return, never replace the measurement, command, bus or reply. -# Comparing independent reads of a running fleet confuses new edits with drift. -OBSERVE_CAPACITY = """ -import contextlib, io, json, sys -import agent_comms.commander.capacity as capacity -from agent_comms.commander.cli import main -measured = [] -contexts = [] -def observe(frame, event, value): - if event == 'return' and frame.f_code is capacity.Load.of.__func__.__code__: - measured.append(value) - if event == 'call' and frame.f_code is capacity.main.__code__: - contexts.append(frame.f_locals['ctx']) -output = io.StringIO() -sys.setprofile(observe) -try: - with contextlib.redirect_stdout(output): - main() -finally: - sys.setprofile(None) -if len(measured) != 1 or len(contexts) != 1: - raise ValueError('Capacity must use exactly one native measurement and selected context') -config = contexts[0] -print(json.dumps(dict(current=output.getvalue(), native=str(measured[0]), - source=capacity.__file__, root=str(config.root), - commander=config.commander))) -""" +from .observations import CapacityObservation, MeasuredOutput def state_files(directory: Path) -> tuple[tuple[str, FileRevision], ...]: @@ -49,16 +25,9 @@ def state_files(directory: Path) -> tuple[tuple[str, FileRevision], ...]: @dataclass(frozen=True) -class MeasuredOutput: - current: str - native: str - source: str - root: str - commander: str - - -@dataclass(frozen=True) -class LoadComparison(ParityComparison): +class CommandComparison(ParityComparison): + arguments: ClassVar[tuple[str, ...]] + timeout: ClassVar[int] = 30 observation: MeasuredOutput state_before: tuple[tuple[str, FileRevision], ...] state_after: tuple[tuple[str, FileRevision], ...] @@ -80,18 +49,27 @@ def describe(self) -> str: @classmethod def read(cls) -> ParityComparison: - config = CommanderConfig.read(CommanderConfig.path()) + return cls.invoke(CommanderConfig.read(CommanderConfig.path()), cls.arguments) + + @classmethod + def invoke(cls, config: CommanderConfig, arguments: tuple[str, ...]) -> ParityComparison: + """Bind one native invocation and all its observations to the selected configuration.""" before = state_files(config.state) # -I uses installed imports even when this candidate runs with PYTHONPATH=src. # HOME and the environment stay unchanged; the actual CLI main runs. - current = subprocess.run( - (str(config.cli.parent / "python"), "-I", "-c", OBSERVE_CAPACITY, "capacity"), - capture_output=True, - text=True, - timeout=30, + runtime = LaunchRuntime.running() + current = asyncio.run( + BoundedRun.run( + (str(runtime.interpreter), "-I", *arguments), + timeout=cls.timeout, + runtime=runtime, + ) ) - if current.returncode: - return UnverifiedParity(current.stderr) + if not current.outcome.successful: + return UnverifiedParity( + f"{cls.__name__}: configured native owner {current.outcome.summary}; " + f"bound={cls.timeout}s; {current.stderr.decode(errors='replace')}" + ) observed = FieldCodec.decode(MeasuredOutput, FieldCodec.parse(current.stdout)) source = Path(observed.source) if ( @@ -103,3 +81,7 @@ def read(cls) -> ParityComparison: "The installed command used a different configuration or source" ) return cls(observed, before, state_files(config.state)) + + +class LoadComparison(CommandComparison): + arguments = CapacityObservation.arguments() diff --git a/src/agent_comms/commander_parity/observations.py b/src/agent_comms/commander_parity/observations.py new file mode 100644 index 0000000000..11067be95c --- /dev/null +++ b/src/agent_comms/commander_parity/observations.py @@ -0,0 +1,123 @@ +"""Observe real native command returns and their actual output.""" + +from __future__ import annotations + +import argparse +import io +import json +import sys +from abc import abstractmethod +from collections.abc import Callable, Iterator +from contextlib import contextmanager, redirect_stdout +from dataclasses import dataclass +from pathlib import Path +from types import CodeType, MethodType + +from agent_comms.commander import capacity +from agent_comms.commander.cli import main as commander_main +from agent_comms.commander.cli import tool_name +from agent_comms.commander.config import CommanderConfig +from agent_comms.commander.order import Commander +from agent_comms.declared_family import DeclaredFamily +from agent_comms.field_codec import FieldCodec + + +@dataclass(frozen=True) +class MeasuredOutput: + current: str + native: str + source: str + root: str + commander: str + + +class NativeObservation(DeclaredFamily, affix="Observation"): + @classmethod + def arguments(cls) -> tuple[str, ...]: + return (str(Path(__file__).resolve()), cls.declared_name) + + @abstractmethod + def read(self, root: Path) -> MeasuredOutput: ... + + +def method_code(method: object) -> CodeType: + if not isinstance(method, MethodType): + raise TypeError("An observation requires a bound Python method") + return method.__func__.__code__ + + +@contextmanager +def returned_values(callbacks: dict[CodeType, Callable[[object], None]]) -> Iterator[None]: + """Observe only declared native returns, not every Python call in the command.""" + monitor = sys.monitoring + tool = monitor.PROFILER_ID + + def returned(code: CodeType, offset: int, value: object) -> None: + callbacks[code](value) + + monitor.use_tool_id(tool, "commander-parity-observation") + try: + monitor.register_callback(tool, monitor.events.PY_RETURN, returned) + for code in callbacks: + monitor.set_local_events(tool, code, monitor.events.PY_RETURN) + yield + finally: + for code in callbacks: + monitor.set_local_events(tool, code, 0) + monitor.register_callback(tool, monitor.events.PY_RETURN, None) + monitor.free_tool_id(tool) + + +class CapacityObservation(NativeObservation): + def read(self, root: Path) -> MeasuredOutput: + measured: list[capacity.Load] = [] + contexts: list[Commander] = [] + + def measurement(value: object) -> None: + if not isinstance(value, capacity.Load): + raise TypeError("Load.of returned outside its declared contract") + measured.append(value) + + def context(value: object) -> None: + if not isinstance(value, Commander): + raise TypeError("The commander CLI returned an undeclared context") + contexts.append(value) + + output = io.StringIO() + # This process owns its command line. The actual CLI selects its root and command. + sys.argv[1:] = ["--root", str(root), tool_name(capacity.__name__)] + with ( + returned_values( + { + method_code(capacity.Load.of): measurement, + method_code(Commander.read_root): context, + } + ), + redirect_stdout(output), + ): + commander_main() + if len(measured) != 1 or len(contexts) != 1: + raise ValueError("Capacity must use one native measurement and selected context") + selected = contexts[0] + return MeasuredOutput( + output.getvalue(), + str(measured[0]), + str(capacity.__file__), + str(selected.comms.root), + selected.commander, + ) + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("observation", type=NativeObservation.member_named) + parser.add_argument( + "--root", type=Path, default=CommanderConfig.read(CommanderConfig.path()).root + ) + arguments = parser.parse_args() + observed = arguments.observation().read(arguments.root) + print(json.dumps(FieldCodec.encode(observed))) + + +if __name__ == "__main__": + main() diff --git a/src/agent_comms/commander_parity/rows.py b/src/agent_comms/commander_parity/rows.py index 3772660101..7847f5531c 100644 --- a/src/agent_comms/commander_parity/rows.py +++ b/src/agent_comms/commander_parity/rows.py @@ -11,6 +11,8 @@ StoryPhase, ) from . import ParityComparison, ParityRow, UnverifiedParity +from .delivery import DeliveredCommitCase +from .handoff import FinishedHandoffCase from .lifecycle import ( AssignOrderCase, BuildBoundProofCase, @@ -50,6 +52,7 @@ class DeliveredCommitParityRow(ParityRow): current = "`order.py done`" native = "`CommsOrderDoneTool`: merged to main, another branch, or not merged" phases = (OneOrderOwnerPhase, RecordPlanePhase) + comparison = DeliveredCommitCase.read class InstalledEvidenceParityRow(ParityRow): @@ -288,6 +291,8 @@ class FinishedWorkTransferParityRow(ParityRow): native = "Transfer refuses delivered checked work; that owner merges; check is reused" phases = (RecordPlanePhase,) + comparison = FinishedHandoffCase.read + class BuildBoundProofParityRow(ParityRow): number = 30 @@ -299,14 +304,6 @@ class BuildBoundProofParityRow(ParityRow): comparison = BuildBoundProofCase.read -class DeletionTargetParityRow(ParityRow): - number = 31 - capability = "An order states its intent as a deletion target" - current = "none; P520 renamed a decision rather than deleting it" - native = "Order owns deletion family/site/count; delivery refuses nonzero or renamed decisions" - phases = (RecordPlanePhase,) - - class UnitTimeParityRow(ParityRow): number = 32 capability = "Where units' time goes, classified" diff --git a/src/agent_comms/goals.py b/src/agent_comms/goals.py index 5cd5f236cf..be2daa6035 100644 --- a/src/agent_comms/goals.py +++ b/src/agent_comms/goals.py @@ -331,12 +331,13 @@ def recorded(self, sha: str, repository: str) -> tuple[ChildResult, ...]: return tuple(checked.result for checked in CheckedCommit.stored(repository, sha) if checked.command == self.command)[:1] - def at(self, revision: str, repository: str) -> ChildResult: + def at(self, revision: str, repository: str, checkout_base: Path) -> ChildResult: """Read the result on ``revision``, or explicitly run the check in a detached checkout.""" sha = CheckedCommit.git(repository, "rev-parse", f"{revision}^{{commit}}").strip() for recorded in self.recorded(sha, repository): return recorded - with tempfile.TemporaryDirectory(prefix="order-check-") as scratch: + checkout_base.mkdir(parents=True, exist_ok=True) + with tempfile.TemporaryDirectory(prefix="order-check-", dir=checkout_base) as scratch: tree = str(Path(scratch) / "tree") CheckedCommit.git(repository, "worktree", "add", "--detach", tree, sha) try: diff --git a/src/agent_comms/harness_install.py b/src/agent_comms/harness_install.py index 98bd666c24..04d49d69ac 100644 --- a/src/agent_comms/harness_install.py +++ b/src/agent_comms/harness_install.py @@ -101,9 +101,11 @@ def install(self, comms: Comms, commander: str, project: Path) -> HarnessInstall @classmethod def published_links(cls, home: Path) -> dict[Path, Path]: - """Installed links of every harness, which a runtime publication moves to the new runtime.""" + """Refresh present harnesses and retain links of previously installed harnesses.""" + available = {type(harness) for harness in cls.present(home)} return {link: target for member in cls.members_with(cls) - for link, target in member(home).links().items() if link.is_symlink()} + for link, target in member(home).links().items() + if member in available or link.is_symlink()} @dataclass(frozen=True) diff --git a/src/agent_comms/launch_runtime.py b/src/agent_comms/launch_runtime.py index d69b9676c7..58927cfcfa 100644 --- a/src/agent_comms/launch_runtime.py +++ b/src/agent_comms/launch_runtime.py @@ -2,8 +2,11 @@ from __future__ import annotations +import json import os +import subprocess import sys +from abc import abstractmethod from collections.abc import Callable, Mapping, MutableMapping from dataclasses import dataclass, field, fields, replace from pathlib import Path @@ -11,7 +14,14 @@ from .declared_family import DeclaredFamily from .errors import RelationViolationError -from .field_codec import FieldCodec, JsonShapeFamily, JsonShapeMember, PathText, TextRepresentation +from .field_codec import ( + FieldCodec, + GitCommitText, + JsonShapeFamily, + JsonShapeMember, + PathText, + TextRepresentation, +) class CallingThreadUnavailable(RelationViolationError): @@ -29,6 +39,14 @@ def running(cls) -> LaunchRuntime: """Keep ordinary launches on the calling process's installation.""" return cls(Path(sys.executable)) + @staticmethod + def current_pointer() -> Path: + return Path.home() / ".local/share/agent-comms/current" + + @classmethod + def installed(cls) -> LaunchRuntime: + return cls(cls.current_pointer().resolve(strict=True) / "bin/python") + @property def prefix(self) -> Path: return self.interpreter.parent.parent.resolve() @@ -39,7 +57,9 @@ def native_entrypoint(self) -> str: def environment(self, source: Mapping[str, str]) -> dict[str, str]: configuration = replace( - RestartEnvironment.inherit(source).for_interpreter(self.interpreter).without_pythonpath(), + RestartEnvironment.inherit(source) + .for_interpreter(self.interpreter) + .without_pythonpath(), virtual_env=str(self.prefix), ) return configuration.apply_runtime(source) @@ -49,6 +69,84 @@ def require_available(self) -> None: raise ValueError("Target interpreter is unavailable") +class ProofBuildMismatchError(ValueError): + """A retained proof does not identify the currently installed pair.""" + + +class BuildIdentity(DeclaredFamily, affix="BuildIdentity"): + @abstractmethod + def recorded(self) -> RecordedBuildIdentity: ... + + @abstractmethod + def require_installed(self) -> None: ... + + +@dataclass(frozen=True) +class UnrecordedBuildIdentity(BuildIdentity): + def recorded(self) -> RecordedBuildIdentity: + return RecordedBuildIdentity.installed() + + def require_installed(self) -> None: + installed = RecordedBuildIdentity.installed() + raise ProofBuildMismatchError( + f"Proof build is unrecorded; installed Core {installed.core}, " + f"Toad {installed.toad}; run and record a new proof." + ) + + +@dataclass(frozen=True) +class RecordedBuildIdentity(BuildIdentity): + """The installed pair, observed from its interpreter rather than this caller's imports.""" + + core: Annotated[str, GitCommitText] + toad: Annotated[str, GitCommitText] + + def recorded(self) -> RecordedBuildIdentity: + return self + + @classmethod + def installed(cls) -> RecordedBuildIdentity: + runtime = LaunchRuntime.installed() + runtime.require_available() + output = subprocess.check_output( + ( + str(runtime.interpreter), + "-c", + "import json; from importlib.metadata import distribution; " + "from agent_comms.package_stamp import STAMP; " + "print(json.dumps([STAMP.commit(), " + "distribution('batrachian-toad').read_text('direct_url.json')]))", + ), + env=runtime.environment(os.environ), + text=True, + timeout=30, + ) + core, metadata = FieldCodec.decode(tuple[str, str], json.loads(output)) + source = FieldCodec.decode(DistributionSource, json.loads(metadata)) + return cls(core, source.vcs_info.commit_id) + + def require_installed(self) -> None: + installed = self.installed() + if self != installed: + raise ProofBuildMismatchError( + f"Proof build Core {self.core}, Toad {self.toad} differs from installed " + f"Core {installed.core}, Toad {installed.toad}; run the proof again." + ) + + +@dataclass(frozen=True) +class VcsSource: + commit_id: Annotated[str, GitCommitText] + vcs: str + requested_revision: str + + +@dataclass(frozen=True) +class DistributionSource: + url: str + vcs_info: VcsSource + + class RestoredRuntime(LaunchRuntime): """Original-process recovery preserves its original import environment too.""" @@ -136,6 +234,7 @@ class RestartEnvironment(EnvironmentRecord): default=default_agent_directory, metadata={"wire_name": "PI_CODING_AGENT_DIR", "native": True}, ) + def __post_init__(self) -> None: object.__setattr__(self, "native_config", self.absolute(self.native_config)) object.__setattr__(self, "agent_directory", self.absolute(self.agent_directory)) @@ -149,7 +248,9 @@ def absolute(self, directory: Path) -> Path: def inherit(cls, environment: Mapping[str, str]): values = cls.inherited(environment) native_name, agent_name = cls.variable("native_config"), cls.variable("agent_directory") - values[native_name] = values.get(native_name) or values.get(agent_name) or str(cls.default_agent_directory) + values[native_name] = ( + values.get(native_name) or values.get(agent_name) or str(cls.default_agent_directory) + ) return FieldCodec.decode(cls, values) def for_agent(self, directory: Path) -> RestartEnvironment: @@ -163,14 +264,18 @@ def for_interpreter(self, interpreter: Path) -> RestartEnvironment: """Put this process's runtime first, exactly once; preserve other search paths.""" directory = str(interpreter.parent) inherited = os.defpath if self.path is None else self.path - path = os.pathsep.join((directory, *(entry for entry in inherited.split(os.pathsep) - if entry != directory))) + path = os.pathsep.join( + (directory, *(entry for entry in inherited.split(os.pathsep) if entry != directory)) + ) return replace(self, path=path) def encode_native(self) -> dict[str, str]: encoded = self.encode() - native = (FieldCodec.wire_key(declared) for declared in fields(self) - if declared.metadata.get("native")) + native = ( + FieldCodec.wire_key(declared) + for declared in fields(self) + if declared.metadata.get("native") + ) return {key: encoded[key] for key in native} def auth_revision(self) -> tuple[int, int]: @@ -252,8 +357,13 @@ def current(cls, environment: Mapping[str, str] = os.environ) -> ThreadLaunch: return FieldCodec.decode(cls, values) @classmethod - def called_by(cls, name: str, canonicalize: Callable[[str], str], *, - environment: Mapping[str, str] = os.environ) -> bool: + def called_by( + cls, + name: str, + canonicalize: Callable[[str], str], + *, + environment: Mapping[str, str] = os.environ, + ) -> bool: try: caller = cls.current(environment).name except CallingThreadUnavailable: diff --git a/src/agent_comms/maintenance_barrier.py b/src/agent_comms/maintenance_barrier.py index 735965d87d..0c70b2a26c 100644 --- a/src/agent_comms/maintenance_barrier.py +++ b/src/agent_comms/maintenance_barrier.py @@ -15,13 +15,14 @@ from abc import ABC, abstractmethod from collections.abc import AsyncIterator, Callable, Iterator from contextlib import AbstractAsyncContextManager, asynccontextmanager, contextmanager -from dataclasses import dataclass +from dataclasses import dataclass, replace from pathlib import Path -from typing import Annotated, Literal +from typing import Literal, Self +from .child_process import ProcessIdentity from .declared_family import DeclaredFamily from .errors import RelationViolationError, ToolRefusedError -from .field_codec import FieldCodec, Id128Text, MalformedRecord, TextRepresentation +from .field_codec import FieldCodec, Id128Text, MalformedRecord from .lifecycle import LifecycleState from .send_outcome import GrantedSend, HeldSend from .store_files import ( @@ -72,6 +73,10 @@ def require_open(): class MaintenancePhase(DeclaredFamily, LifecycleState, affix="Phase"): new_sends = running_sends = HeldSends + @classmethod + def holder_dead(cls, state: MaintenanceState) -> bool: + return not state.operator.alive() + @classmethod def hold_for(cls, barrier: MaintenanceBarrier, state: MaintenanceStatus, operator: MaintenanceOperator) -> MaintenanceState: @@ -123,6 +128,10 @@ def successors(cls): class ReadyPhase(MaintenancePhase): new_sends = running_sends = GrantedSends + @classmethod + def holder_dead(cls, state: MaintenanceState) -> bool: + return False + @classmethod def hold_for(cls, barrier: MaintenanceBarrier, state: MaintenanceStatus, operator: MaintenanceOperator) -> MaintenanceState: @@ -141,6 +150,10 @@ def hold_restore(cls, barrier: MaintenanceBarrier, operator: MaintenanceOperator drained() +class DeadHolderReleasedPhase(ReadyPhase): + """The stored operator is the dead incarnation whose hold this record released.""" + + # The holding operation hands its nonce to its own child processes through this variable. HOLDER_NONCE = "AGENT_COMMS_MAINTENANCE_HOLDER" @@ -155,6 +168,12 @@ class MaintenanceStatus: def turn_pending(self) -> bool: return self.phase.new_sends.pending() + def observed(self, barrier: MaintenanceBarrier) -> MaintenanceStatus: + return self + + async def observed_async(self, barrier: MaintenanceBarrier) -> MaintenanceStatus: + return self + def require_open(self) -> None: self.phase.new_sends.require_open() @@ -163,31 +182,19 @@ def require_open_to(self, holder: str) -> None: self.require_open() -@dataclass(frozen=True) -class MaintenanceOperator: - value: str +@dataclass(frozen=True, eq=False) +class MaintenanceOperator(ProcessIdentity, declared_name=None): + """A hold belongs to this exact process, using the existing identity codec.""" - def __post_init__(self) -> None: - if not self.value or len(self.value) > 128: - raise ValueError('Maintenance operator must have bounded identity') - - -class MaintenanceOperatorText(TextRepresentation[MaintenanceOperator]): @classmethod - def from_text(cls, value: str) -> MaintenanceOperator: - return MaintenanceOperator(value) - - @classmethod - def encode(cls, value: MaintenanceOperator) -> str: - return value.value - - -MaintenanceOperatorField = Annotated[MaintenanceOperator, MaintenanceOperatorText] + def current(cls) -> Self: + identity = ProcessIdentity.capture(os.getpid()) + return cls(identity.pid, identity.start_time) @dataclass(frozen=True, kw_only=True) class MaintenanceMarker: - version: Literal[1] + version: Literal[2] root: str generation: int @@ -198,7 +205,7 @@ def __post_init__(self): @dataclass(frozen=True, kw_only=True) class MaintenanceState(MaintenanceMarker, MaintenanceStatus): - operator: MaintenanceOperatorField + operator: MaintenanceOperator nonce: str phase: type[MaintenancePhase] @@ -206,6 +213,16 @@ def __post_init__(self): super().__post_init__() Id128Text.decode(self.nonce) + def observed(self, barrier: MaintenanceBarrier) -> MaintenanceStatus: + if self.phase.holder_dead(self): + return barrier.release_dead(self) + return self + + async def observed_async(self, barrier: MaintenanceBarrier) -> MaintenanceStatus: + if self.phase.holder_dead(self): + return await barrier.release_dead_async(self) + return self + def require_marker(self, marker: MaintenanceMarker, root: Path) -> MaintenanceState: if ( marker @@ -293,7 +310,7 @@ def _read(path: Path, declaration): os.close(descriptor) return value - def current_unlocked(self) -> MaintenanceStatus: + def _current_unlocked(self) -> MaintenanceStatus: """Check the phase under the registry lock OR the wire lock. Clients may read it before taking locks only to refuse work, never to @@ -310,6 +327,32 @@ def current_unlocked(self) -> MaintenanceStatus: raise RelationViolationError("Maintenance witness incomplete; new turns held") return state.require_marker(marker, self.registry_path.parent) + def current_unlocked(self) -> MaintenanceStatus: + return self._current_unlocked().observed(self) + + def _release_dead_unlocked(self, expected: MaintenanceState) -> MaintenanceStatus: + current = self._current_unlocked() + if current != expected: + return current + self._write_unlocked(replace( + expected, generation=expected.generation + 1, nonce=Id128Text.mint(), + phase=DeadHolderReleasedPhase, + )) + raise MaintenanceHeldError( + f"Maintenance released dead holder pid={expected.operator.pid} " + f"start_time={expected.operator.start_time}; retry the input" + ) + + def release_dead(self, expected: MaintenanceState) -> MaintenanceStatus: + # Registry acquisition already serializes every maintenance write. It is + # reentrant for a registry borrower and never upgrades a shared wire OFD. + with _store_lock(self.registry_path): + return self._release_dead_unlocked(expected) + + async def release_dead_async(self, expected: MaintenanceState) -> MaintenanceStatus: + async with _async_store_lock(self.registry_path): + return self._release_dead_unlocked(expected) + def assert_open_unlocked(self) -> MaintenanceStatus: hold = self.current_unlocked() hold.require_open_to(os.environ.get(HOLDER_NONCE, "")) @@ -333,9 +376,8 @@ def begin(self, operator: MaintenanceOperator) -> MaintenanceState: """Hold new turns before fleet drain; an uncertain write stays closed.""" with _store_lock(self.wire_path), _store_lock(self.registry_path): current = self.current_unlocked() - if current.phase is not ReadyPhase: - raise RelationViolationError('Maintenance already active or uncertain') - hold = MaintenanceState(version=1, root=str(self.registry_path.parent.resolve()), + current.require_open() + hold = MaintenanceState(version=2, root=str(self.registry_path.parent.resolve()), generation=current.generation + 1, operator=operator, nonce=Id128Text.mint(), phase=DrainingPhase) self._write_unlocked(hold) @@ -405,4 +447,4 @@ async def ingress_async(self) -> AsyncIterator[MaintenanceStatus]: pipe drain or a native response. """ async with _async_store_lock(self.wire_path, mode=SharedLock()): - yield self.current_unlocked() + yield await self._current_unlocked().observed_async(self) diff --git a/src/agent_comms/owner_cutover.py b/src/agent_comms/owner_cutover.py index 86c883dbf1..e80bc56ad7 100644 --- a/src/agent_comms/owner_cutover.py +++ b/src/agent_comms/owner_cutover.py @@ -121,13 +121,9 @@ def complete(self, stopped: StoppedOwnerBatch) -> tuple[OwnerRestartResult, ...] self.after_stopped(stopped.lifecycle) return stopped.launch(self.launch_runtime(stopped)) - @staticmethod - def current_pointer() -> Path: - return Path.home() / ".local/share/agent-comms/current" - def launch_runtime(self, stopped: StoppedOwnerBatch) -> LaunchRuntime: """Only installation selects the runtime after publishing its pointer.""" - selected = LaunchRuntime(self.current_pointer().resolve(strict=True) / "bin/python") + selected = LaunchRuntime.installed() selected.require_available() return selected diff --git a/src/agent_comms/recorded_time.py b/src/agent_comms/recorded_time.py index 75a073e15e..1708046988 100644 --- a/src/agent_comms/recorded_time.py +++ b/src/agent_comms/recorded_time.py @@ -7,12 +7,11 @@ from dataclasses import dataclass from math import isfinite - from .declared_family import DeclaredFamily from .field_codec import FieldCodec, FieldRepresentation -class MissingTranscriptTimestampError(ValueError): - """An external source cannot publish an event without its recorded time.""" +class UnrecordedTimeError(ValueError): + """An operation requires a source-recorded time, but none was supplied.""" class GeneratedTimestamp(FieldRepresentation): @@ -71,11 +70,11 @@ def clock(self) -> str: return "-" def require_recorded(self, label: str) -> None: - raise ValueError(f"{label} requires a recorded time.") + self.recorded_at(label) def recorded_at(self, label: str) -> float: - """The recorded instant; an unrecorded time refuses like ``require_recorded``.""" - raise ValueError(f"{label} requires a recorded time.") + """The recorded instant; callers requiring a time cannot invent one.""" + raise UnrecordedTimeError(f"{label} requires a recorded time.") def lateness(self, deadline: EventTime) -> Lateness: """How late this delivery time is against ``deadline``; either unrecorded refuses.""" @@ -87,9 +86,6 @@ def require_deadline(self) -> None: def require_delivery(self) -> None: self.require_recorded("A delivery") - def transcript_timestamp(self) -> float: - raise MissingTranscriptTimestampError("A transcript event requires its source's recorded time") - @abstractmethod def continued_from(self, original_at: float): ... @@ -108,9 +104,6 @@ def __post_init__(self): if type(self.at) not in (int, float) or not isfinite(self.at): raise ValueError("A recorded event time must be a finite Unix timestamp.") - def require_recorded(self, label: str) -> None: - pass - def chronology(self) -> tuple[int, float]: return (0, self.at) @@ -133,8 +126,5 @@ def at_clock(self, clock: str) -> RecordedEventTime: def recorded_at(self, label: str) -> float: return self.at - def transcript_timestamp(self) -> float: - return self.at - def continued_from(self, original_at): return self diff --git a/src/agent_comms/runtime_updates.py b/src/agent_comms/runtime_updates.py index 5f4583edc1..6ec4da313c 100644 --- a/src/agent_comms/runtime_updates.py +++ b/src/agent_comms/runtime_updates.py @@ -96,7 +96,7 @@ def events(self) -> tuple[TranscriptEvent, ...]: ) return ( (self.event_type(text, position=self.position, - timestamp=self.started_at.transcript_timestamp()),) if text else () + timestamp=self.started_at.recorded_at("A transcript event")),) if text else () ) def snapshot(self, turn_id: str, item_id: str) -> tuple[DraftItem, ...]: diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 6d9105eeed..deecba8f88 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -37,6 +37,7 @@ from .command import Command from .declared_family import DeclaredFamily from .errors import ToolRefusedError +from .external_transcript import ExternalTranscript from .field_codec import ( FieldCodec, GitCommitText, @@ -47,6 +48,7 @@ WireValue, ) from .goals import CheckedCommit, OrderCheck, ProblemGoalOrder +from .launch_runtime import BuildIdentity, UnrecordedBuildIdentity from .order_story import ProblemStory, StoryEvent, StoryLine, StorySource from .plan_phases import OrderPhase, UnplannedPhase from .recorded_time import EventTime, RecordedEventTime, UnrecordedEventTime @@ -91,6 +93,8 @@ class OrderDelivery(DeclaredFamily, StoryEvent, affix="Delivery"): at: EventTime = field(default_factory=UnrecordedEventTime) evidence: DeliveryEvidence = field(default_factory=NoDeliveryEvidence, metadata={"wire_omit_default": True}) + proof_build: BuildIdentity = field(default_factory=UnrecordedBuildIdentity, + metadata={"wire_omit_default": True}) @property def work(self) -> str: @@ -115,14 +119,15 @@ async def verified(self, todo: Todo, registry: Registration) -> OrderDelivery: check = order.check for each in (check, *order.deletes.checks): await self.require_source(each) - each.result.require(await self.checked(each)) - return replace(self, at=await self.delivery_time(check)) + each.result.require(await self.checked(each, registry.storage_rule.checkout_base)) + return replace(self, at=await self.delivery_time(check), + proof_build=await asyncio.to_thread(self.proof_build.recorded)) @abstractmethod async def require_source(self, check): ... @abstractmethod - async def checked(self, check: OrderCheck) -> ChildResult: ... + async def checked(self, check: OrderCheck, checkout_base: Path) -> ChildResult: ... @abstractmethod async def delivery_time(self, check): ... @@ -169,8 +174,8 @@ def check_failed(self, goals: Goals, todo: Todo, error: OrderCheckFailedError) - async def require_source(self, check: OrderCheck) -> None: raise ValueError("A commit delivery names the unit's delivered commit and the checked one.") - async def checked(self, check: OrderCheck) -> ChildResult: - return await asyncio.to_thread(check.at, self.commit, check.directory) + async def checked(self, check: OrderCheck, checkout_base: Path) -> ChildResult: + return await asyncio.to_thread(check.at, self.commit, check.directory, checkout_base) async def delivery_time(self, check: OrderCheck) -> EventTime: return EventTime.capture(float( @@ -207,7 +212,8 @@ async def verified(self, todo: Todo, registry: Registration) -> OrderDelivery: verified = await super().verified(todo, registry) merge = await self.merge_state(todo, registry) return CheckedCommitDelivery(commit=self.commit, delivered_commit=self.delivered, - at=verified.at, evidence=self.evidence, merge=merge) + at=verified.at, evidence=self.evidence, merge=merge, + proof_build=verified.proof_build) async def merge_state(self, todo: Todo, registry: Registration) -> DeliveryMerge: """Observe integration, not merely publication, at the time of completion. @@ -301,7 +307,7 @@ def line(self) -> str: async def require_source(self, check): pass - async def checked(self, check: OrderCheck) -> ChildResult: + async def checked(self, check: OrderCheck, checkout_base: Path) -> ChildResult: return await check.run() async def delivery_time(self, check): @@ -1484,6 +1490,9 @@ def require_transfer(self, todo: Todo) -> None: def require_close(self): raise ValueError("Closing requires recorded installed application evidence.") + def require_installation(self) -> None: + raise ValueError("Check the delivered order before recording installation.") + def require_withdrawal(self) -> None: raise ValueError("Only an undelivered order can be withdrawn.") @@ -1801,6 +1810,9 @@ def closing_artifact(self) -> str: def corrected(self, correction: CorrectCheckTodo) -> ProblemOrder: raise ValueError("A delivered order keeps the check its delivery passed.") + def require_installation(self) -> None: + self.delivery.proof_build.require_installed() + def installed(self, works, artifact): self.run.require_closing(works) return self.copy_as(InstalledProblemOrder, works=works, artifact=artifact) @@ -2619,6 +2631,12 @@ class InstallTodo(CloseOrderTodo): works: str artifact: str + def prepare(self, goals: Goals, current: Todo) -> InstallTodo: + current.require_revision(self.expected_revision) + current.require_commander(self.actor) + current.order.require_installation() + return self + def closed(self, current: Todo) -> ProblemOrder: if not Path(self.artifact).is_file(): raise ValueError("Retain the installed application artifact before closing.") @@ -2902,6 +2920,16 @@ def writing(self) -> Iterator[sqlite3.Connection]: f"{self.path}: {type(error).__name__}: {error}" ) from error + @contextmanager + def writer_connection( + self, *, check_same_thread: bool = True + ) -> Iterator[sqlite3.Connection]: + with ( + MaintenanceBarrier(self.root / "registry.json").allow_input(), + super().writer_connection(check_same_thread=check_same_thread) as connection, + ): + yield connection + def record_hunt(self, commander: ThreadIncarnation, hunt: Hunt) -> HuntRecord: hunt.require_commander(self, commander) recorded = HuntRecord(uuid4().hex, commander, hunt) diff --git a/src/agent_comms/tools.py b/src/agent_comms/tools.py index 80f1ee08b8..f391302df0 100644 --- a/src/agent_comms/tools.py +++ b/src/agent_comms/tools.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import sys import threading from abc import abstractmethod from collections.abc import Mapping @@ -12,6 +13,7 @@ from pydantic import JsonValue +from . import cli as comms_cli from .commander.capacity import Load from .progress_claims import ProgressClaim from .pi_vocabulary import ServiceTier, ThinkingLevel, ThinkingLevelField @@ -68,7 +70,7 @@ TodoOperation, VerdictAudience) from .change_report import ChangeReport from .messages import InfoKind, MessageKind, NoticeRole, SendableMessageKind -from .native_tools import NativeToolDeclaration, NativeRequestTools +from .native_tools import ArgumentTool, NativeToolDeclaration, NativeRequestTools from .tool_caller import ToolCaller, ThreadCaller, NoThreadCaller, ThreadListing from .thread_identity import ThreadIncarnation from .cli_result import ToolCatalogResult @@ -293,7 +295,7 @@ def tool_field( @dataclass(frozen=True, kw_only=True) class ToolRequest[Caller: ToolCaller = ToolCaller]( - ThreadCommandAvailability, NativeRequestTools, NativeToolDeclaration + ArgumentTool, ThreadCommandAvailability, NativeRequestTools, NativeToolDeclaration ): description: ClassVar[str] bus_effect: ClassVar[str] = ("Changes the bus. Tristan has authorized claude-ai " @@ -329,6 +331,11 @@ def schema(cls) -> dict[str, object]: ) return result + def cli_command(self, caller: ToolCaller) -> InvokeCliCommand: + """Project this declaration into Core's existing external tool command.""" + return InvokeCliCommand(caller=caller.acting_name(''), + tool=self.declared_name, arguments=self.arguments) + @classmethod def invoke( cls, comms: Comms, arguments: Mapping[str, object], *, caller: ToolCaller @@ -1388,6 +1395,13 @@ class InvokeCliCommand(CliCommand, declared_name="tool"): "--arguments", help="Tool arguments as a JSON object", default_factory=dict, parser_default="{}") + def argv(self, root: Path) -> tuple[str, ...]: + """The CLI boundary owns its module, grammar and argument serialization.""" + caller = ToolCaller.from_actor(self.caller) + return (sys.executable, '-m', comms_cli.__name__, '--root', str(root), + self.declared_name, '--from', caller.acting_name(''), '--tool', self.tool, + '--arguments', json.dumps(self.arguments)) + def apply(self, ctx: Comms) -> dict[str, JsonValue]: return invoke_tool(ctx, self.tool, self.arguments, caller=ToolCaller.from_actor(self.caller)) diff --git a/src/agent_comms/turn_digests.py b/src/agent_comms/turn_digests.py index 335f184d2a..414657faf8 100644 --- a/src/agent_comms/turn_digests.py +++ b/src/agent_comms/turn_digests.py @@ -18,6 +18,7 @@ if TYPE_CHECKING: from .comms import Comms + from .registry_document import RegistrySnapshot from .threads import Thread from .todos import Todo @@ -122,13 +123,25 @@ class OrderRead[Result](TranscriptRead[Result]): @classmethod def for_commander(cls, comms: Comms, commander: str) -> Iterator[Self]: - actor = comms.registry.require(commander).incarnation - for todo in comms.goals.todos.held_for_commander(actor): + snapshot = comms.registry.snapshot() + actor = snapshot.require(commander).incarnation + yield from cls.for_orders( + comms.root, snapshot, comms.goals.todos.held_for_commander(actor) + ) + + @classmethod + def for_orders( + cls, root: Path, snapshot: RegistrySnapshot, todos: Iterable[Todo] + ) -> Iterator[Self]: + """Read these orders against their captured roster, without reopening the registry.""" + for todo in todos: assignment = todo.held() - holder = comms.registry.require(assignment.owner) + holder = snapshot.require(assignment.owner) if holder.incarnation != assignment.owner_identity: raise ToolRefusedError(f"Order {todo.id} holder incarnation no longer exists") - yield cls(comms.root, holder, TurnsSince(todo.order.checked().at.recorded_at("An order")), todo) + yield cls( + root, holder, TurnsSince(todo.order.checked().at.recorded_at("An order")), todo + ) def edits_from_turns(self, turns: tuple[TurnDigest, ...]) -> tuple[EditCall, ...]: judged = {verdict.tool_call_id for verdict in self.todo.verdicts} diff --git a/tests/conftest.py b/tests/conftest.py index 4e45d3708a..0e456bffaf 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -300,7 +300,11 @@ def commander_script(commander, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) from agent_comms.commander.config import CommanderConfig comms, _ = commander + installed = LaunchRuntime.current_pointer().resolve(strict=True) monkeypatch.setenv("HOME", str(tmp_path)) + pointer = LaunchRuntime.current_pointer() + pointer.parent.mkdir(parents=True) + pointer.symlink_to(installed) repo = tmp_path / "repo" repo.mkdir() for command in (("init", "-q"), ("remote", "add", "origin", "https://github.com/OpenHCSDev/agent-comms.git")): diff --git a/tests/runtime_publication_journey.py b/tests/runtime_publication_journey.py index 972607958b..cd2e3d151d 100644 --- a/tests/runtime_publication_journey.py +++ b/tests/runtime_publication_journey.py @@ -17,11 +17,11 @@ source = Path(__file__).resolve().parents[1] from agent_comms.active_route import ActiveRoute, publish_active_route from agent_comms.comms import Comms -from agent_comms.owner_cutover import StoppedOwnerInstallation +from agent_comms.launch_runtime import LaunchRuntime from agent_comms.maintenance_barrier import ReadyPhase from agent_comms.runtime import socket_path -from tools.publish_runtime import RuntimePublication -from tools.cutover.carry import CarryChain, CarryLedger +from publish_runtime import RuntimePublication +from cutover.carry import CarryChain, CarryLedger def main(): @@ -41,7 +41,7 @@ def main(): for key in tuple(os.environ): if key.startswith(('AGENT_COMMS_', 'PI_')) and key not in environment: del os.environ[key] - current = StoppedOwnerInstallation.current_pointer() + current = LaunchRuntime.current_pointer() current.parent.mkdir(mode=0o700, parents=True) current.symlink_to(old) starter = ''' @@ -75,14 +75,14 @@ def main(): try: # Kill only the isolated publisher while it is waiting, then re-enter # against the real maintenance witness it left behind. - waiting = subprocess.Popen([target / 'bin/python', '-I', '-c', ''' + waiting = subprocess.Popen([target / 'bin/python', '-E', '-s', '-c', ''' import sys from pathlib import Path -from tools.publish_runtime import RuntimePublication +from publish_runtime import RuntimePublication from agent_comms.comms import Comms root=Path(sys.argv[2]);c=Comms(root) RuntimePublication(Path(sys.argv[3]),{}).publish(root,checkout=Path(sys.argv[1])) -''', str(source), str(root), str(target)], env=dict(os.environ)) +''', str(source), str(root), str(target)], env=dict(os.environ), cwd=source / 'tools') deadline = time.monotonic() + 10 while not comms.owners.maintenance.read().turn_pending: assert waiting.poll() is None and time.monotonic() < deadline diff --git a/tests/test_carry.py b/tests/test_carry.py index 92bd2ee332..f07284ca7e 100644 --- a/tests/test_carry.py +++ b/tests/test_carry.py @@ -41,7 +41,7 @@ from agent_comms.recorded_time import RecordedEventTime from test_todos import order, thread from cutover.backup import Backup, UpgradeRefused, bus_files, transcripts_held -from tools.upgrade_bus import ( +from upgrade_bus import ( Revisions, Stack, before_stop, @@ -503,7 +503,7 @@ def test_turn_records_drop_an_idle_stored_turn_and_refuse_a_held_one(tmp_path, m def test_upgrade_stops_and_starts_exactly_the_captured_set_concurrently(comms, tmp_path, monkeypatch): from threading import Barrier - import tools.upgrade_bus as upgrade_bus + import upgrade_bus from agent_comms.maintenance_barrier import ReadyPhase names = [f"worker-{number}" for number in range(19)] @@ -582,7 +582,7 @@ def cli(self, root, *arguments): def test_selected_revision_prepares_pins_and_cannot_return_a_failed_cached_tree(tmp_path, monkeypatch): - import tools.upgrade_bus as upgrade_bus + import upgrade_bus commands = [] @@ -609,7 +609,7 @@ def missing_dependency(*command): def test_failed_carry_preserves_changed_bus_and_main_under_maintenance(comms, tmp_path, monkeypatch): - import tools.upgrade_bus as upgrade_bus + import upgrade_bus from agent_comms.maintenance_barrier import ReadyPhase datum = comms.root / "kept.txt" @@ -686,12 +686,12 @@ def bridge_publishes(bus: Path) -> str: def rehearsed(stack: Stack, root: Path) -> dict: - from tools.upgrade_bus import rehearse + from upgrade_bus import rehearse return rehearse(root, (), stack, None) def installed_live(stack: Stack, root: Path) -> dict: - from tools.upgrade_bus import live + from upgrade_bus import live return live(stack.checkout, root, "old", "new", (), stack, stack, None, "worker") @@ -699,7 +699,7 @@ def installed_live(stack: Stack, root: Path) -> dict: def test_a_step_writing_the_transcript_store_runs_after_the_hold(comms, tmp_path, monkeypatch, install): """The d88a12185 rehearsal: ToolCallIndex's BEGIN IMMEDIATE timed out against the installer's own transcript hold, which then stayed up across every carry.""" - import tools.upgrade_bus as upgrade_bus + import upgrade_bus from cutover.carry_tool_call_index import ToolCallIndex unit = comms.registry.declare(comms.registry.new_thread( @@ -752,7 +752,7 @@ def test_a_writer_during_the_drain_is_refused_or_its_file_is_declared(comms, mon output.enter(ToolRunningPhase("before the drain")) with transcript.writing(): pass - MaintenanceBarrier(comms.root / "registry.json").begin(MaintenanceOperator("drain-test")) + MaintenanceBarrier(comms.root / "registry.json").begin(MaintenanceOperator.current()) before = bus_files(comms.root) monkeypatch.setattr(CoordinationStore, "writer_timeout", 0.1) with transcripts_held(comms.root): diff --git a/tests/test_claude_tool_preview.py b/tests/test_claude_tool_preview.py index 118983aab1..0bf1ffc0be 100644 --- a/tests/test_claude_tool_preview.py +++ b/tests/test_claude_tool_preview.py @@ -24,13 +24,15 @@ def test_claude_tool_names_arguments_and_diff_reach_shared_events(name, argument record=TimedRecord(position=0,uuid='record',timestamp='2026-10-17T00:00:00+00:00') tool=ToolUse(id='call',name=name,input=arguments) active,=tool.updates(record,session) - call=session.tools['call'] + call,source=session.tools['call'] + assert source is record assert call.name==native and call.arguments==expected and call.title==title assert active.detail == title result=ToolResult(tool_use_id='call',content=[{'type':'text','text':'done'}]) completed,_=result.updates(record,session) started,ended=completed.events assert started.raw_input==expected and started.title==title + assert started.timestamp == ended.timestamp == record.timestamp if patch is None: assert ended.diff is None else: diff --git a/tests/test_cli_watch_exit.py b/tests/test_cli_watch_exit.py index 41ce78fb5d..3adb77f19f 100644 --- a/tests/test_cli_watch_exit.py +++ b/tests/test_cli_watch_exit.py @@ -164,7 +164,7 @@ async def test_a_watch_pauses_for_an_upgrade_and_resumes_after_its_carry(tmp_pat from cutover.carry import STEPS, CarryChain, CarryLedger from test_bus_route_counts_locking import _case from test_carry import REPO, CheckoutStack, Repository - from tools.upgrade_bus import Revisions, carry_all + from upgrade_bus import Revisions, carry_all from agent_comms.maintenance_barrier import MaintenanceBarrier, MaintenanceOperator comms = _case(tmp_path / 'wire', monkeypatch) @@ -192,7 +192,7 @@ async def deliver(self, message): try: assert await asyncio.wait_for(arrived.get(), 20) == 'first' maintenance = MaintenanceBarrier(comms.registry.store.path) - held = maintenance.begin(MaintenanceOperator('upgrade-bus')) + held = maintenance.begin(MaintenanceOperator.current()) assert await asyncio.wait_for(children[0].wait(), 20) is not None (tmp_path / 'checkout/src').mkdir(parents=True) repo = Repository(tmp_path / 'checkout') @@ -230,7 +230,7 @@ async def test_a_bridge_started_during_a_drain_waits_and_then_receives(tmp_path, comms.registry.declare(replace(recipient, execution=ExternalThreadExecution)) monkeypatch.setenv('HOME', str(tmp_path / 'home')) maintenance = MaintenanceBarrier(comms.registry.store.path) - held = maintenance.begin(MaintenanceOperator('upgrade-bus')) + held = maintenance.begin(MaintenanceOperator.current()) arrived: asyncio.Queue[str] = asyncio.Queue() children = [] start = AttachedChild.start diff --git a/tests/test_codex_earlier_pages.py b/tests/test_codex_earlier_pages.py index 35e0f47538..fb8a573a58 100644 --- a/tests/test_codex_earlier_pages.py +++ b/tests/test_codex_earlier_pages.py @@ -165,12 +165,12 @@ def test_stopped_carry_preserves_external_items(tmp_path): from agent_comms.external_transcript import ExternalTranscript, NoEarlierPage from agent_comms.field_codec import FieldCodec from agent_comms.transcript_events import AssistantTranscript - from tools.cutover.carry_external_pages import ExternalPages + from cutover.carry_external_pages import ExternalPages root = tmp_path / 'root' root.mkdir() path = ExternalTranscript.path_in(root) - saved = json.dumps(FieldCodec.encode((AssistantTranscript('preserved', position='i'),))) + saved = json.dumps(FieldCodec.encode((AssistantTranscript('preserved', position='i', timestamp=1.0),))) # This historical conversion predates the separate event table. with closing(CoordinationStore.writer(path)) as connection: connection.execute('CREATE TABLE external_item (thread TEXT NOT NULL, turn_id TEXT NOT NULL, ' diff --git a/tests/test_commander_parity_delivery.py b/tests/test_commander_parity_delivery.py new file mode 100644 index 0000000000..023bf6faa1 --- /dev/null +++ b/tests/test_commander_parity_delivery.py @@ -0,0 +1,34 @@ +from agent_comms.commander_parity.delivery import DeliveryRepository, DeliveredCommitCase +from agent_comms.commander_parity.rows import DeliveredCommitParityRow +from agent_comms.goals import CountCheckResult, OrderCheck + + +def test_delivery_graph_is_pushed_and_distinguishes_integration(tmp_path): + repository = DeliveryRepository.create(tmp_path) + main, branch, unmerged = repository.cases() + for _, commit in (main, branch, unmerged): + assert repository.git("branch", "--remotes", "--contains", commit, "--list", "origin/*") + assert repository.git("merge-base", main[1], "origin/main") == main[1] + assert repository.git("merge-base", branch[1], "origin/integration") == branch[1] + assert repository.git("merge-base", branch[1], "origin/main") != branch[1] + assert repository.git("merge-base", unmerged[1], "origin/main") != unmerged[1] + assert repository.git("merge-base", unmerged[1], "origin/integration") != unmerged[1] + + +def test_check_uses_configured_persistent_checkout_and_retains_real_result(tmp_path): + repository = DeliveryRepository.create(tmp_path) + checkout_base = tmp_path / "persistent-checkouts" + marker = tmp_path / "observed-checkouts" + check = OrderCheck(str(repository.path), f'pwd >> "{marker}"; echo 0', CountCheckResult(0)) + commit = repository.git("rev-parse", "HEAD") + first = check.at(commit, str(repository.path), checkout_base) + assert check.at(commit, str(repository.path), checkout_base) == first + (actual,) = marker.read_text().splitlines() + assert actual.startswith(str(checkout_base) + "/order-check-") + assert not tuple(checkout_base.iterdir()) + assert len(check.recorded(commit, str(repository.path))) == 1 + + +def test_delivery_row_declares_comparison_once(): + assert "verify" not in vars(DeliveredCommitParityRow) + assert DeliveredCommitParityRow.comparison == DeliveredCommitCase.read diff --git a/tests/test_commander_parity_load.py b/tests/test_commander_parity_load.py index 6e26d798fa..129c5fd33c 100644 --- a/tests/test_commander_parity_load.py +++ b/tests/test_commander_parity_load.py @@ -1,7 +1,13 @@ +import shutil +import sys from dataclasses import replace +import pytest + from agent_comms.commander.capacity import Load -from agent_comms.commander_parity.load import LoadComparison, MeasuredOutput, state_files +from agent_comms.commander.config import CommanderConfig +from agent_comms.commander_parity.load import CommandComparison, LoadComparison, state_files +from agent_comms.commander_parity.observations import CapacityObservation, MeasuredOutput def test_load_comparison_requires_every_reported_field_and_no_private_write(tmp_path): @@ -22,3 +28,50 @@ def test_load_comparison_requires_every_reported_field_and_no_private_write(tmp_ path.write_text("after") after = state_files(tmp_path) assert not LoadComparison(observed, before, after).passed + + +def test_capacity_observes_actual_cli_selected_root(commander_script, tmp_path, monkeypatch): + original, _ = commander_script + selected = tmp_path / "selected" + shutil.copytree(original.root, selected) + before = state_files(original.root) + monkeypatch.setattr(sys, "argv", ["capacity-observation"]) + + observed = CapacityObservation().read(selected) + + assert observed.root == str(selected.resolve()) + assert observed.commander == original.commander + assert observed.current.strip() == observed.native + assert state_files(original.root) == before + assert sys.monitoring.get_tool(sys.monitoring.PROFILER_ID) is None + + +@pytest.mark.parametrize( + ("program", "expected"), + ( + ("raise SystemExit(7)", "exited with code 7"), + ("import time; time.sleep(10)", "timed out"), + ), +) +def test_failed_real_children_remain_named_unverified(tmp_path, program, expected): + """Negative process checks produce no fabricated native measurement or parity pass.""" + + class BriefComparison(CommandComparison): + timeout = 3 + + config = CommanderConfig( + commander="reader", + root=tmp_path / "bus", + state=tmp_path / "state", + board=tmp_path / "board.md", + narrative=tmp_path / "narrative.md", + repos=(), + worktrees=tmp_path / "projects", + scratch=tmp_path / "scratch", + disk=tmp_path, + ) + comparison = BriefComparison.invoke(config, ("-c", program)) + assert not comparison.passed + assert expected in comparison.describe() + assert "bound=3s" in comparison.describe() + assert LoadComparison.timeout == 30 diff --git a/tests/test_connector.py b/tests/test_connector.py index 4b6be05fbd..f519a77151 100644 --- a/tests/test_connector.py +++ b/tests/test_connector.py @@ -306,7 +306,7 @@ async def test_maintenance_refuses_before_locked_stores_on_a_copied_bus( async with TestClient(TestServer(connector.application())) as reader: headers = {"Authorization": f"Bearer {await approved_token(reader, store)}"} barrier = MaintenanceBarrier(copied_root / "registry.json") - held = barrier.begin(MaintenanceOperator("connector-refusal-proof")) + held = barrier.begin(MaintenanceOperator.current()) for phase in phases: held = barrier.advance(held, phase) with _store_lock(barrier.wire_path), _store_lock(barrier.registry_path), ExitStack() as locks: diff --git a/tests/test_maintenance_barrier.py b/tests/test_maintenance_barrier.py index 84446c15f2..368d6f449d 100644 --- a/tests/test_maintenance_barrier.py +++ b/tests/test_maintenance_barrier.py @@ -20,7 +20,7 @@ DrainingPhase, InstallingPhase, MaintenanceBarrier, - MaintenanceOperatorField, + MaintenanceOperator, MaintenancePhase, PausedPhase, ReadyPhase, @@ -44,7 +44,7 @@ async def test_channel_post_queues_without_start_during_cutover(tmp_path: Path, # A post pending when the drain begins; the drain refuses new publication. # Use the actual scratch bus, not a constructed message pretending to be a post. post = comms.messaging.send('sender', '#team', 'mid-drain channel post').message - hold = control.begin(FieldCodec.decode(MaintenanceOperatorField, 'scratch-cutover')) + hold = control.begin(MaintenanceOperator.current()) queued = ScheduledTurn.incoming(post) starts = [] @@ -83,7 +83,7 @@ async def start(*args, **kwargs): assert next_sealed_assignment(store, RecordedThreadIncarnation("fixture", beta.created_at).lookup, beta.name).wire_seq == post.seq # Selection scheduled just before another drain must still refuse at the # effect boundary without consuming that durable assignment. - another_drain = control.begin(FieldCodec.decode(MaintenanceOperatorField, 'second-cutover')) + another_drain = control.begin(MaintenanceOperator.current()) assert await selected[0].run() == another_drain with Coordination(comms.root / 'coordination.sqlite3') as store: assert next_sealed_assignment(store, RecordedThreadIncarnation("fixture", beta.created_at).lookup, beta.name).wire_seq == post.seq @@ -108,7 +108,7 @@ async def test_cutover_clear_delivers_pending_channel_post(tmp_path: Path, monke comms, agent, _root_id = _session(tmp_path) gate = comms.owners.maintenance post = comms.messaging.send('sender', '#team', 'while draining').message - hold = gate.begin(FieldCodec.decode(MaintenanceOperatorField, 'scratch-cutover')) + hold = gate.begin(MaintenanceOperator.current()) assert await agent.inputs.drain_inbox('beta') == 0 assert await agent.inputs.drain_inbox('beta') == 0 beta = comms.registry.require('beta') @@ -133,12 +133,12 @@ async def test_cutover_clear_delivers_pending_channel_post(tmp_path: Path, monke @pytest.mark.parametrize('text', [None, '', 'x' * 129, 42]) def test_maintenance_operator_rejects_invalid_external_text(text: object) -> None: with pytest.raises((ValueError, TypeError)): - FieldCodec.decode(MaintenanceOperatorField, text) + FieldCodec.decode(MaintenanceOperator, text) def test_draining_lends_existing_turn_but_holds_new_prompt(tmp_path: Path) -> None: gate = MaintenanceBarrier(tmp_path / 'wire' / 'registry.json') - hold = gate.begin(FieldCodec.decode(MaintenanceOperatorField, 'operator')) + hold = gate.begin(MaintenanceOperator.current()) assert hold.phase is DrainingPhase with gate.read().phase.send_new(None) as new: assert isinstance(new, HeldSend) @@ -153,14 +153,14 @@ def test_draining_lends_existing_turn_but_holds_new_prompt(tmp_path: Path) -> No def test_phase_owner_is_default_off_and_clear_is_exact(tmp_path: Path) -> None: gate = Comms(tmp_path / 'fresh').owners.maintenance assert gate.read().phase is ReadyPhase - hold = gate.begin(FieldCodec.decode(MaintenanceOperatorField, 'operator')) + hold = gate.begin(MaintenanceOperator.current()) assert gate.read() == hold ready = gate.clear(hold) assert ready.phase is ReadyPhase and gate.read().phase is ReadyPhase with pytest.raises(RelationViolationError, match='generation/operator'): gate.clear(hold) assert gate.marker_path.is_file() and gate.state_path.is_file() - assert gate.read().operator == FieldCodec.decode(MaintenanceOperatorField, 'operator') + assert gate.read().operator == MaintenanceOperator.current() def test_only_hold_bound_cutover_may_retire_during_draining(tmp_path: Path) -> None: @@ -175,7 +175,7 @@ def complete(self, stopped): raise AssertionError('No process should be stopped by a phase check') comms = Comms(tmp_path / 'wire') - hold = comms.owners.maintenance.begin(FieldCodec.decode(MaintenanceOperatorField, 'scratch-cutover')) + hold = comms.owners.maintenance.begin(MaintenanceOperator.current()) cutover = ScratchCutover() cutover.maintenance_hold = hold with _store_lock(comms.root / 'wire'): @@ -193,13 +193,13 @@ def complete(self, stopped): def test_clear_stopped_uses_acquired_wire_descriptor(tmp_path: Path) -> None: gate = MaintenanceBarrier(tmp_path / 'wire' / 'registry.json') - hold = gate.begin(FieldCodec.decode(MaintenanceOperatorField, 'stopped-cutover')) + hold = gate.begin(MaintenanceOperator.current()) with _store_lock(gate.wire_path) as lock: assert gate.clear_stopped(hold, lock.descriptor).phase is ReadyPhase assert gate.read().phase is ReadyPhase with _store_lock(tmp_path / 'unrelated') as other: with pytest.raises(RelationViolationError, match='wire lock changed'): - next_hold = gate.begin(FieldCodec.decode(MaintenanceOperatorField, 'new-cutover')) + next_hold = gate.begin(MaintenanceOperator.current()) gate.clear_stopped(next_hold, other.descriptor) @@ -246,7 +246,7 @@ def test_default_off_then_close_reopen_and_no_stale_transition(tmp_path: Path) - assert comms.agents.begin_turn("owner", "before") comms.agents.finish_turn(comms.registry.require("owner").turn_lease) control = gate - first = control.begin(FieldCodec.decode(MaintenanceOperatorField, "operator-one")) + first = control.begin(MaintenanceOperator.current()) assert first.phase == MaintenancePhase.member_named("draining") and first.generation == 1 assert MaintenanceBarrier(comms.registry.store.path).read() == first with pytest.raises(RelationViolationError, match="Maintenance"): @@ -278,7 +278,7 @@ def test_state_generation_requires_exact_matching_positive_int( ) -> None: gate = MaintenanceBarrier(tmp_path / "wire" / "registry.json") assert gate.read().phase is ReadyPhase # Fresh roots remain default OFF. - gate.begin(FieldCodec.decode(MaintenanceOperatorField, "disposable-fixture")) + gate.begin(MaintenanceOperator.current()) state = json.loads(gate.state_path.read_text()) state["phase"] = "ready" # Disposable READY detects a false-open parser result. state["generation"] = generation @@ -303,7 +303,7 @@ def test_state_generation_requires_exact_matching_positive_int( ) def test_enabled_witness_damage_never_restores_default_off(tmp_path: Path, fault: str) -> None: gate = MaintenanceBarrier(tmp_path / "wire" / "registry.json") - gate.begin(FieldCodec.decode(MaintenanceOperatorField, "operator")) + gate.begin(MaintenanceOperator.current()) if fault == "missing-state": gate.state_path.unlink() elif fault == "missing-marker": @@ -332,7 +332,7 @@ async def test_direct_claim_and_bind_denied_after_phase_ack_in_other_process(tmp comms.registry.declare( comms.registry.new_thread(name='owner', tags=frozenset(), worktree=str(tmp_path), process_identity=ProcessIdentity.capture(proc.pid)) ) - hold = comms.owners.maintenance.begin(FieldCodec.decode(MaintenanceOperatorField, "operator")) + hold = comms.owners.maintenance.begin(MaintenanceOperator.current()) ready.set() proc.join(10) assert proc.exitcode == 0 @@ -362,7 +362,7 @@ def uncertain_write(path: Path, text: str, *, fsync_parent: bool = False) -> Non monkeypatch.setattr(barrier_module, "_atomic_write_text", uncertain_write) with pytest.raises(OSError, match="injected"): - gate.begin(FieldCodec.decode(MaintenanceOperatorField, "operator")) + gate.begin(MaintenanceOperator.current()) with pytest.raises(RelationViolationError, match="Maintenance"): gate.assert_open_unlocked() if fault_at == "before-state": @@ -380,7 +380,7 @@ def test_rename_and_stopped_same_pid_cannot_reactivate_under_gate(tmp_path: Path comms.registry.declare(owner) comms.registry.rename("owner", "renamed") comms.registry.unregister("renamed") - comms.owners.maintenance.begin(FieldCodec.decode(MaintenanceOperatorField, "operator")) + comms.owners.maintenance.begin(MaintenanceOperator.current()) with pytest.raises(RelationViolationError, match="Maintenance"): comms.owners.start("owner") with pytest.raises(RelationViolationError, match="Maintenance"): @@ -400,7 +400,7 @@ def test_cross_process_claim_races_pause_at_registry_lock(tmp_path: Path) -> Non comms.registry.new_thread(name='owner', tags=frozenset(), worktree=str(tmp_path), process_identity=ProcessIdentity.capture(child.pid)) ) ready.set() - hold = comms.owners.maintenance.begin(FieldCodec.decode(MaintenanceOperatorField, "operator")) + hold = comms.owners.maintenance.begin(MaintenanceOperator.current()) child.join(10) assert child.exitcode == 0 result, _ = q.get(timeout=2) @@ -423,7 +423,7 @@ async def test_actual_native_held_input_resumes_from_existing_observer(native_ba async with owner.open_owner(runtime_enabled=True) as (agent, session_id): async with owner.original_input(agent, session_id, "Held original input") as turn: gate = agent._comms.owners.maintenance - hold = gate.begin(FieldCodec.decode(MaintenanceOperatorField, "operator")) + hold = gate.begin(MaintenanceOperator.current()) sending = asyncio.create_task(turn.run_native()) try: await wait_for(lambda: sending in TurnSession.active and @@ -484,7 +484,7 @@ async def fake_send() -> None: assert release.wait(5) def pause() -> None: - control.begin(FieldCodec.decode(MaintenanceOperatorField, "operator")) + control.begin(MaintenanceOperator.current()) changed.set() sender = threading.Thread(target=lambda: asyncio.run(fake_send())) @@ -517,7 +517,7 @@ def written() -> tuple[bytes, ...]: before = written() gate = MaintenanceBarrier(tmp_path / 'registry.json') - hold = gate.begin(FieldCodec.decode(MaintenanceOperatorField, 'upgrade-bus')) + hold = gate.begin(MaintenanceOperator.current()) for argv in (('send', '--from', 'a', '--to', 'b', '--body', 'hi'), ('activity', '--name', 'a', '--state', 'working')): code, out = cli(*argv) @@ -531,7 +531,6 @@ async def test_bus_waits_before_mutating_a_transcript_batch(tmp_path: Path) -> N from functools import partial from agent_comms.activity import ActivityState from agent_comms.runtime_updates import AppendDraftItem, StartedItem, TextDraft, ThreadOutput - from agent_comms.maintenance_barrier import MaintenanceOperator from agent_comms.recorded_time import RecordedEventTime from agent_comms.transcript_events import AssistantTranscript @@ -542,7 +541,7 @@ async def test_bus_waits_before_mutating_a_transcript_batch(tmp_path: Path) -> N TextDraft.from_parts(AssistantTranscript, ('prefix',), RecordedEventTime(1.), 'item')), AppendDraftItem('turn', 'item', ' once')] barrier = MaintenanceBarrier(comms.registry.store.path) - hold = barrier.begin(MaintenanceOperator('batch-check')) + hold = barrier.begin(MaintenanceOperator.current()) pending = asyncio.create_task(comms.bus.when_open(partial(output.publish_batch, updates))) try: with pytest.raises(TimeoutError): diff --git a/tests/test_maintenance_holder.py b/tests/test_maintenance_holder.py new file mode 100644 index 0000000000..2d2d06d416 --- /dev/null +++ b/tests/test_maintenance_holder.py @@ -0,0 +1,171 @@ +"""Exact, real process death releases maintenance without a timeout or PID table fake.""" + +from __future__ import annotations + +import asyncio +import json +import multiprocessing as mp +import os +import signal +from contextlib import contextmanager +from pathlib import Path + +import pytest + +from agent_comms.field_codec import FieldCodec, MalformedRecord +from agent_comms.maintenance_barrier import ( + DeadHolderReleasedPhase, + DrainingPhase, + InstallingPhase, + MaintenanceBarrier, + MaintenanceHeldError, + MaintenanceOperator, + MaintenancePhase, + MaintenanceState, + PausedPhase, + ReadyPhase, +) +from agent_comms.store_files import ExclusiveLock, SharedLock, _async_store_lock, _store_lock +from agent_comms.todos import GoalRef, ObservedFirstCallEvent, RecordFirstCallTodo, TodoStore +from agent_comms.recorded_time import RecordedEventTime +from cutover.carry_maintenance_identity import MaintenanceIdentity +from test_todos import order, thread + + +def _hold(root: Path, phase: type[MaintenancePhase], ready) -> None: + barrier = MaintenanceBarrier(root / "registry.json") + held = barrier.begin(MaintenanceOperator.current()) + for successor in (PausedPhase, InstallingPhase): + if held.phase is phase: + break + held = barrier.advance(held, successor) + ready.send(json.dumps(FieldCodec.encode(held))) + ready.close() + signal.pause() + + +@contextmanager +def real_holder(root: Path, phase: type[MaintenancePhase] = DrainingPhase): + receive, send = mp.Pipe(duplex=False) + child = mp.Process(target=_hold, args=(root, phase, send), daemon=True) + child.start() + send.close() + try: + assert receive.poll(10), "Real holder did not publish its hold" + held = FieldCodec.decode(MaintenanceState, FieldCodec.parse(receive.recv())) + assert held.operator.pid == child.pid and held.operator.alive() + yield child, held + finally: + if child.is_alive(): + child.kill() + child.join(5) + receive.close() + assert not child.is_alive() + + +def test_operator_uses_process_identity_codec() -> None: + operator = MaintenanceOperator.current() + assert operator.pid == os.getpid() and operator.start_time > 0 + assert FieldCodec.encode(operator) == {"pid": operator.pid, "start_time": operator.start_time} + assert FieldCodec.decode(MaintenanceOperator, FieldCodec.encode(operator)) == operator + with pytest.raises(MalformedRecord): + FieldCodec.decode(MaintenanceOperator, "upgrade-bus") + + +@pytest.mark.parametrize("phase", [DrainingPhase, PausedPhase, InstallingPhase]) +@pytest.mark.parametrize("file,mode", [("wire", SharedLock()), ("registry.json", ExclusiveLock())]) +def test_dead_holder_refuses_detecting_input_then_opens(tmp_path, phase, file, mode): + root = tmp_path / "bus" + barrier = MaintenanceBarrier(root / "registry.json") + with real_holder(root, phase) as (child, held): + child.kill() + child.join(5) + with _store_lock(root / file, mode=mode): + with pytest.raises( + MaintenanceHeldError, + match=f"pid={held.operator.pid} start_time={held.operator.start_time}", + ): + barrier.assert_open_unlocked() + released = barrier.read() + assert released.phase is DeadHolderReleasedPhase + assert released.operator == held.operator + assert released.generation == held.generation + 1 + with barrier.allow_input(): + pass + assert barrier.read() == released + # A released record is READY behavior, not a special-case phase table. + replacement = barrier.begin(MaintenanceOperator.current()) + assert replacement.phase is DrainingPhase + barrier.clear(replacement) + + +async def test_async_recovery_borrows_registry_without_blocking_event_loop(tmp_path: Path): + root = tmp_path / "bus" + barrier = MaintenanceBarrier(root / "registry.json") + with real_holder(root) as (child, _held): + child.kill() + child.join(5) + + async def incoming(): + async with barrier.ingress_async(): + pytest.fail("The detecting input must be refused") + + async with _async_store_lock(root / "registry.json"): + attempt = asyncio.create_task(incoming()) + await asyncio.sleep(0.025) + assert not attempt.done() + with pytest.raises(MaintenanceHeldError, match="released dead holder"): + await asyncio.wait_for(attempt, 5) + async with barrier.ingress_async() as status: + status.require_open() + + +def test_commander_first_call_write_is_admitted_before_sqlite_mutation(tmp_path: Path): + store = TodoStore(tmp_path) + lead = thread("lead", tmp_path, 1.0) + original = store.create( + "P1", + "org/repo", + "Fix maintenance", + creator=lead, + owner=lead, + generation="one", + goal=GoalRef("lead", 1.0, "request"), + deadline=RecordedEventTime(1000), + order=order(tmp_path), + ) + barrier = MaintenanceBarrier(tmp_path / "registry.json") + held = barrier.begin(MaintenanceOperator.current()) + before = store.path.read_bytes() + notification = RecordFirstCallTodo(original.revision, lead.incarnation, ObservedFirstCallEvent) + with pytest.raises(MaintenanceHeldError): + store.apply("P1", notification) + assert store.path.read_bytes() == before and store.get("P1") == original + barrier.clear(held) + assert store.apply("P1", notification).revision == original.revision + 1 + + +def test_legacy_ready_has_one_registered_conversion(tmp_path: Path): + barrier = MaintenanceBarrier(tmp_path / "registry.json") + tmp_path.mkdir(exist_ok=True) + marker = {"version": 1, "root": str(tmp_path.resolve()), "generation": 4} + state = {**marker, "operator": "upgrade-bus", "nonce": "a" * 32, "phase": "ready"} + barrier.marker_path.write_text(json.dumps(marker)) + barrier.state_path.write_text(json.dumps(state)) + assert MaintenanceIdentity.carry(tmp_path, apply=False).changed == 1 + assert MaintenanceIdentity.carry(tmp_path, apply=True).changed == 1 + assert barrier.read().phase is ReadyPhase + assert barrier.read().operator == MaintenanceOperator.current() + assert MaintenanceIdentity.carry(tmp_path, apply=True).changed == 0 + + +def test_legacy_active_name_is_not_fabricated_into_a_process(tmp_path: Path): + barrier = MaintenanceBarrier(tmp_path / "registry.json") + marker = {"version": 1, "root": str(tmp_path.resolve()), "generation": 4} + state = {**marker, "operator": "upgrade-bus", "nonce": "a" * 32, "phase": "draining"} + barrier.marker_path.write_text(json.dumps(marker)) + barrier.state_path.write_text(json.dumps(state)) + before = barrier.state_path.read_bytes() + with pytest.raises(MalformedRecord): + MaintenanceIdentity.carry(tmp_path, apply=True) + assert barrier.state_path.read_bytes() == before diff --git a/tests/test_maintenance_native_boundary.py b/tests/test_maintenance_native_boundary.py index b12b4e3502..388f8f9389 100644 --- a/tests/test_maintenance_native_boundary.py +++ b/tests/test_maintenance_native_boundary.py @@ -13,7 +13,7 @@ import agent_comms.coordinated_runtime as runtime import agent_comms.native_pi as native_pi -from agent_comms.maintenance_barrier import MaintenancePhase +from agent_comms.maintenance_barrier import DeadHolderReleasedPhase, MaintenanceHeldError from agent_comms.child_process import ProcessIdentity from agent_comms.coordinator import Coordination from agent_comms.errors import RelationViolationError @@ -32,12 +32,12 @@ async def test_private_native_raw_prompt_refused_after_pause_ack(tmp_path: Path, "from pathlib import Path\n" "from agent_comms.field_codec import FieldCodec\n" "from agent_comms.maintenance_barrier import (\n" - " MaintenanceBarrier, MaintenanceOperatorField, PausedPhase)\n" + " MaintenanceBarrier, MaintenanceOperator, PausedPhase)\n" f"Path({str(child_pid)!r}).write_text(str(os.getpid()))\n" "request = json.loads(sys.stdin.readline())\n" f"with open({str(received)!r}, 'a') as log: log.write(json.dumps(request) + '\\n')\n" f"control = MaintenanceBarrier(Path({str(root / 'registry.json')!r}))\n" - "first = control.begin(FieldCodec.decode(MaintenanceOperatorField, 'disposable-operator'))\n" + "first = control.begin(MaintenanceOperator.current())\n" "assert control.advance(first, PausedPhase).phase is PausedPhase\n" "file = sys.argv[1]\n" f"reply = {{'type':'response','id':request['id'],'command':'get_state','success':True," @@ -77,11 +77,20 @@ def launch(package, *, worktree, session, **_kwargs): opt_in=True, cursor_delivery=cursor_delivery, ).run() - assert comms.owners.maintenance.read().phase == MaintenancePhase.member_named("paused") writes = [json.loads(line) for line in received.read_text().splitlines()] assert [item["type"] for item in writes] == ["get_state"] + holder_pid = int(child_pid.read_text()) with pytest.raises(ProcessLookupError): - ProcessIdentity.capture(int(child_pid.read_text())) + ProcessIdentity.capture(holder_pid) + # The raw prompt was refused while its RPC holder was alive. Retirement + # killed that holder, so the detecting read refuses once and releases it. + with pytest.raises(MaintenanceHeldError, match=f"released dead holder pid={holder_pid} "): + comms.owners.maintenance.read() + released = comms.owners.maintenance.read() + assert released.phase is DeadHolderReleasedPhase + assert released.operator.pid == holder_pid and released.operator.start_time > 0 + with comms.owners.maintenance.allow_input(): + pass with Coordination(str(root / "coordination.sqlite3")) as store: assert ( store.session._connection.execute( diff --git a/tests/test_order_progress.py b/tests/test_order_progress.py index deefcfb470..4cddca923c 100644 --- a/tests/test_order_progress.py +++ b/tests/test_order_progress.py @@ -139,7 +139,7 @@ def test_uncounted_check_keeps_other_overdue_orders_visible(commander, commander replace(todo, order=broken).upsert(db) create(commander, 'P2', deadline=1) commit(unit) - broken.check.at(git(unit, 'rev-parse', 'HEAD'), str(unit)) + broken.check.at(git(unit, 'rev-parse', 'HEAD'), str(unit), unit.parent) progress = read((comms.goals.todos.get('P1'), unit)) assert progress.line.endswith("; check prints no integer: printf 'no count here'") @@ -275,7 +275,7 @@ def test_progress_reads_only_explicit_check_receipts(checkout, tmp_path): source = MeasuredProgressSource.of(todo, StorageRule(checkout_base=unit.parent), Listed()) assert not source.readings().readings assert not marker.exists() - check.at(git(unit, 'rev-parse', 'HEAD').strip(), str(unit)) + check.at(git(unit, 'rev-parse', 'HEAD').strip(), str(unit), unit.parent) assert marker.read_text() == 'x' assert len(source.readings().readings) == 1 assert source.readings().readings[0].distance == CountPrinted(todo.order.checked().untouched(())) @@ -313,8 +313,8 @@ def test_order_due_moves_the_deadline_in_core_and_on_the_record(commander, comma script.due('P1', '23:59', 'too short') script.due('P1', '23:59', 'the owner widened the order to the rotation line too') moved = comms.goals.todos.get('P1') - assert commander_order.hhmm(moved.deadline) == '23:59' and moved.revision == before.revision + 1 - assert 'P1 order order-P1 due ' in (tmp_path / 'narrative.md').read_text() + assert moved.deadline.clock() == '23:59:00' and moved.revision == before.revision + 1 + assert 'P1 order order-P1 deadline changed): ' in (tmp_path / 'narrative.md').read_text() class HeldSource: diff --git a/tests/test_proof_build.py b/tests/test_proof_build.py new file mode 100644 index 0000000000..5ad7b1af3b --- /dev/null +++ b/tests/test_proof_build.py @@ -0,0 +1,109 @@ +"""Build-bound close uses real native tools and a retained-history bus copy.""" + +import subprocess +from dataclasses import replace +from pathlib import Path + +import pytest + +from agent_comms.change_report import FileLine +from agent_comms.check_root import CheckRoot +from agent_comms.comms import Comms +from agent_comms.field_codec import FieldCodec +from agent_comms.goal_actions import CreateProblemGoalAction, SetGoalAction +from agent_comms.goals import CountCheckResult, OrderCheck, RequestGoalOrder +from agent_comms.launch_runtime import RecordedBuildIdentity +from agent_comms.progress_claims import WorkingClaim +from agent_comms.recorded_time import RecordedEventTime +from agent_comms.tool_caller import ThreadCaller +from agent_comms.tools import CommsOrderDoneTool, CommsProblemCloseTool, CommsTodoCreateTool +from agent_comms.todos import ( + ApplicationOrderRun, + FileDeliveryEvidence, + OperationDelivery, + PendingProblemOrder, + Todo, +) + + +def test_installed_proof_closes_and_older_build_is_refused_on_copied_bus(tmp_path): + source = Comms(tmp_path / "source") + source.messaging.initialize_private_initial_protocol() + for number, name in enumerate(("lead", "worker"), 1): + source.registry.declare( + source.registry.new_thread( + name, frozenset({"comms"}), str(tmp_path), created_at=float(number) + ) + ) + installed = RecordedBuildIdentity.installed() + older_core = subprocess.check_output( + ("git", "rev-parse", f"{installed.core}^"), + cwd=Path(__file__).resolve().parents[1], + text=True, + ).strip() + older = replace(installed, core=older_core) + caller = ThreadCaller("lead") + works = "Open the installed application and confirm the copied goal keeps its source location." + artifact = tmp_path / "proof.txt" + artifact.write_text(works) + with CheckRoot(source, ("lead", "worker"), tmp_path / "copied").open() as copied: + copied.goals.update_goal( + "lead", + SetGoalAction( + text="Record the copied application proof", + goal_id="request", + order=RequestGoalOrder(), + ), + ) + for problem, build in (("P1", installed), ("P2", older)): + order = PendingProblemOrder( + owner=FileLine(__file__, 1), + check=OrderCheck(str(tmp_path), "printf 0", CountCheckResult(0)), + edits=("Record the copied application proof on its installed build.",), + banned=("Do not change the source bus.",), + report="Report the actual close result.", + run=ApplicationOrderRun(works), + ) + caller.invoke( + CommsTodoCreateTool( + owner="worker", + order=CreateProblemGoalAction( + problem=problem, + order=order, + request="request", + deadline=RecordedEventTime(4102444800), + commander="lead", + repo="OpenHCSDev/agent-comms", + text="Bind proof to its build", + ), + ), + copied, + ) + caller.invoke( + CommsOrderDoneTool( + problem=problem, + order_id=order.id, + delivery=OperationDelivery( + evidence=FileDeliveryEvidence(str(artifact)), proof_build=build + ), + claims=(WorkingClaim(str(artifact)),), + ), + copied, + ) + delivered = copied.goals.todos.get(problem) + assert delivered.order.delivery.proof_build == build + assert FieldCodec.decode(Todo, FieldCodec.encode(delivered)) == delivered + close = CommsProblemCloseTool(problem=problem, works=works, artifact=str(artifact)) + if build == installed: + caller.invoke(close, copied) + saved = copied.goals.todos.get(problem) + assert saved.order.accepted and not saved.state.open + else: + with pytest.raises(ValueError) as refusal: + caller.invoke(close, copied) + assert all( + revision in str(refusal.value) + for revision in (installed.core, installed.toad, older.core, older.toad) + ) + assert copied.goals.todos.get(problem) == delivered + assert not source.goals.todos.for_commander(source.registry.require("lead").incarnation) diff --git a/tests/test_registration_process_identity.py b/tests/test_registration_process_identity.py index eec476e3be..6ebbc32728 100644 --- a/tests/test_registration_process_identity.py +++ b/tests/test_registration_process_identity.py @@ -28,7 +28,7 @@ def test_bridge_releases_during_maintenance_without_allowing_a_new_owner(tmp_pat gate = comms.owners.maintenance process = ProcessIdentity.capture(os.getpid()) with registry.external_process('bridge', process): - hold = gate.begin(MaintenanceOperator('bridge-release')) + hold = gate.begin(MaintenanceOperator.current()) for phase in phases: hold = gate.advance(hold, phase) with pytest.raises(RelationViolationError, match='Maintenance'): diff --git a/tests/test_runtime_publication.py b/tests/test_runtime_publication.py index 0d95d50910..8ac4d827b5 100644 --- a/tests/test_runtime_publication.py +++ b/tests/test_runtime_publication.py @@ -8,7 +8,7 @@ import pytest from agent_comms.active_route import LocalRoute -from agent_comms.owner_cutover import StoppedOwnerInstallation +from agent_comms.launch_runtime import LaunchRuntime from tools.publish_runtime import RuntimePublication @@ -23,7 +23,7 @@ def test_cli_failure_leaves_links_untouched(tmp_path, monkeypatch, command): link = tmp_path / "command" link.symlink_to("previous-command") publication = RuntimePublication(runtime, {link: binary}) - current = StoppedOwnerInstallation.current_pointer() + current = LaunchRuntime.current_pointer() current.parent.mkdir(parents=True) current.symlink_to("previous-runtime") before = (link.lstat(), current.lstat()) @@ -39,7 +39,7 @@ def test_cli_failure_leaves_links_untouched(tmp_path, monkeypatch, command): def test_pending_carries_refuse_before_cli_or_maintenance(tmp_path, monkeypatch): monkeypatch.setenv('HOME', str(tmp_path)) - current = StoppedOwnerInstallation.current_pointer() + current = LaunchRuntime.current_pointer() current.parent.mkdir(parents=True) current.symlink_to(Path(sys.executable).parent.resolve().parent) root = tmp_path / "private-bus" @@ -50,24 +50,28 @@ def test_pending_carries_refuse_before_cli_or_maintenance(tmp_path, monkeypatch) assert not (root / "runtime-publication.json").exists() -def test_pointer_publication_does_not_kill_unselected_runtime_processes( - tmp_path, monkeypatch): +def test_pointer_publication_does_not_kill_unselected_runtime_processes(tmp_path, monkeypatch): monkeypatch.setenv("HOME", str(tmp_path)) + # These tiny process-lifetime runtimes contain no packaged harness commands. + monkeypatch.setenv("PATH", str(tmp_path / "no-harnesses")) runtimes = {name: tmp_path / name for name in ("previous-runtime", "runtime")} for runtime in runtimes.values(): (runtime / "bin").mkdir(parents=True) (runtime / "bin/python").symlink_to(sys.executable) - current = StoppedOwnerInstallation.current_pointer() + current = LaunchRuntime.current_pointer() current.parent.mkdir(parents=True) current.symlink_to(runtimes["previous-runtime"]) - retired, kept = (subprocess.Popen([runtime / "bin/python", "-c", "import time; time.sleep(60)"]) - for runtime in runtimes.values()) + retired, kept = ( + subprocess.Popen([runtime / "bin/python", "-c", "import time; time.sleep(60)"]) + for runtime in runtimes.values() + ) try: RuntimePublication(runtimes["runtime"].resolve(), {}).publish_links() assert retired.poll() is None assert kept.poll() is None and current.readlink() == runtimes["runtime"].resolve() - with pytest.raises(ValueError, match="would retire the runtime it publishes"): - RuntimePublication(runtimes["runtime"].resolve(), {}).publish_links() + RuntimePublication(runtimes["runtime"].resolve(), {}).publish_links() + assert retired.poll() is None and kept.poll() is None + assert current.resolve() == runtimes["runtime"].resolve() finally: for child in (retired, kept): child.kill() diff --git a/tests/test_send_refusals.py b/tests/test_send_refusals.py index cc21afebad..4dbc5ae12e 100644 --- a/tests/test_send_refusals.py +++ b/tests/test_send_refusals.py @@ -8,7 +8,7 @@ from agent_comms.coordination_tables.refusals import SendRefusal from agent_comms.errors import RelationViolationError, ToolRefusedError from agent_comms.notification_assignment import NotificationAssignment -from tools.cutover.carry_send_refusals import SendRefusalOutcomes +from cutover.carry_send_refusals import SendRefusalOutcomes def test_install_preserves_existing_coordination_rows(comms): diff --git a/tests/test_stall_watcher_strike.py b/tests/test_stall_watcher_strike.py index 9b4d09ef8a..2bf82b015b 100644 --- a/tests/test_stall_watcher_strike.py +++ b/tests/test_stall_watcher_strike.py @@ -205,7 +205,7 @@ async def test_three_connector_passes_allow_real_store_maintenance_between_passe from agent_comms.maintenance_barrier import MaintenanceBarrier, MaintenanceOperator from agent_comms.store_files import _store_lock barrier = MaintenanceBarrier(Path(sys.argv[1]) / 'registry.json') -held = barrier.begin(MaintenanceOperator('P717 copied-root proof')) +held = barrier.begin(MaintenanceOperator.current()) try: with _store_lock(barrier.wire_path), _store_lock(barrier.registry_path): directory = os.open(active_route_path().parent, os.O_RDONLY | os.O_DIRECTORY) diff --git a/tests/test_standing_rules.py b/tests/test_standing_rules.py index 26f099f238..570dac6828 100644 --- a/tests/test_standing_rules.py +++ b/tests/test_standing_rules.py @@ -207,17 +207,17 @@ def commit(left): check = f"echo run >> {runs}; cat left" todo = create(commander, "P1", command=check, expected=0, deadline=time.time() + 3600) ordered_check = todo.order.checked().check - ordered_check.at(git("rev-parse", "HEAD").strip(), str(unit)) + ordered_check.at(git("rev-parse", "HEAD").strip(), str(unit), unit.parent) def progress(): return OrderProgress.read(comms.goals.todos.get("P1"), storage, PullRequests()) commit(2) - ordered_check.at(git("rev-parse", "HEAD").strip(), str(unit)) + ordered_check.at(git("rev-parse", "HEAD").strip(), str(unit), unit.parent) assert "; distance 2, projected done " in progress().line assert isinstance(progress().distance_readings().trend, MovingTrend) commit(2) - ordered_check.at(git("rev-parse", "HEAD").strip(), str(unit)) + ordered_check.at(git("rev-parse", "HEAD").strip(), str(unit), unit.parent) assert progress().line.endswith("; distance 2, not falling") assert runs.read_text().count("run") == 3 # once per commit: base, then each branch commit diff --git a/tests/test_thread_identity.py b/tests/test_thread_identity.py index dda305b003..4426095827 100644 --- a/tests/test_thread_identity.py +++ b/tests/test_thread_identity.py @@ -13,7 +13,7 @@ from agent_comms.coordinator import Coordination from agent_comms.errors import RelationViolationError from agent_comms.field_codec import FieldCodec -from agent_comms.maintenance_barrier import MaintenanceBarrier, MaintenanceOperatorField +from agent_comms.maintenance_barrier import MaintenanceBarrier, MaintenanceOperator from agent_comms.messages import InfoKind, Message from agent_comms.read_basis import Conversation from agent_comms.read_ledger import ReadDocument @@ -304,7 +304,7 @@ def test_each_thread_role_member_owns_its_role_behaviour( barrier = MaintenanceBarrier(tmp_path / "wire" / "registry.json") role.require_maintenance_open(barrier, True) - barrier.begin(FieldCodec.decode(MaintenanceOperatorField, "operator")) + barrier.begin(MaintenanceOperator.current()) role.require_maintenance_open(barrier, False) if executes: with pytest.raises(RelationViolationError, match="under maintenance"): diff --git a/tests/test_transcript_event_time.py b/tests/test_transcript_event_time.py index 49a042ffb1..aa7356e456 100644 --- a/tests/test_transcript_event_time.py +++ b/tests/test_transcript_event_time.py @@ -56,10 +56,10 @@ def test_original_record_clock_survives_split_and_routed_projection(tmp_path, st def test_new_native_entry_inherits_original_clock_projection(): class ExternalNoteEntry(NativeEntry): - def _events(self, context): - return [AssistantTranscript("a new external note", position=context.position)] + def _events(self, context, timestamp): + return [AssistantTranscript("a new external note", position=context.position, timestamp=timestamp)] - # Adding a content declaration does not add another clock propagation site. + # The inherited entry owner decodes the source clock before constructing any content. stamp = "2026-09-28T23:12:33.123Z" expected = datetime.fromisoformat(stamp).timestamp() note = ExternalNoteEntry(timestamp=RecordedTimestamp(stamp)) @@ -153,3 +153,41 @@ async def serve(reader, writer): print( f"actual native clocks={expected}; ACP snapshot + rebuild exact; native history/proof unchanged" ) + + +@pytest.mark.parametrize('source', ('claude', 'codex')) +@pytest.mark.parametrize('stamp', (None, 'missing')) +def test_external_source_without_a_time_refuses_before_event_construction(source, stamp): + from pydantic import ValidationError + from agent_comms.cli_bridges.claude_protocol import TimedRecord + from agent_comms.cli_bridges.codex_rollout import ROLLOUT_RECORD + from agent_comms.recorded_time import UnrecordedTimeError + + record = ({'uuid': 'record', 'position': 0} if source == 'claude' else + {'type': 'response_item', 'payload': {'type': 'message', 'role': 'assistant', 'content': []}}) + if stamp is None: + record['timestamp'] = None + decode = TimedRecord.model_validate if source == 'claude' else ROLLOUT_RECORD.validate_python + with pytest.raises(ValidationError) as refused: + decode(record) + assert any(isinstance(error.get('ctx', {}).get('error'), UnrecordedTimeError) + for error in refused.value.errors()) + + +def test_codex_rollout_projects_the_enclosing_record_clock(): + from agent_comms.cli_bridges.codex_rollout import ROLLOUT_RECORD, RolloutState + + record = ROLLOUT_RECORD.validate_python({ + 'type': 'response_item', 'timestamp': '2026-10-19T18:00:00Z', + 'payload': {'type': 'message', 'role': 'assistant', 'id': 'item', + 'content': [{'text': 'saved source text'}]}, + }) + update, = record.updates(RolloutState(), 17) + assert update.events[0].timestamp == datetime.fromisoformat('2026-10-19T18:00:00Z').timestamp() + + +def test_malformed_shell_words_refuse_instead_of_reaching_any_rule(tmp_path): + from agent_comms.storage_rules import ShellInvocation + + with pytest.raises(ValueError): + tuple(ShellInvocation.parse('uv pip install package "unterminated', tmp_path)) diff --git a/tests/verify_connector_maintenance.py b/tests/verify_connector_maintenance.py index 71a36975a0..a3b853124a 100644 --- a/tests/verify_connector_maintenance.py +++ b/tests/verify_connector_maintenance.py @@ -66,7 +66,7 @@ def prove(scratch: Path) -> None: try: wait_started(service, server_log) barrier = MaintenanceBarrier(copied / "registry.json") - held = barrier.begin(MaintenanceOperator("connector-maintenance-proof")) + held = barrier.begin(MaintenanceOperator.current()) held = barrier.advance(held, PausedPhase) held = barrier.advance(held, InstallingPhase) with store.granted("private maintenance proof") as token: diff --git a/tests/verify_external_event_times.py b/tests/verify_external_event_times.py new file mode 100644 index 0000000000..e1d80b326a --- /dev/null +++ b/tests/verify_external_event_times.py @@ -0,0 +1,39 @@ +"""Run the source-backed carry on a SQLite snapshot, preserving the live store.""" +from __future__ import annotations + +import json +import sys +from contextlib import closing +from pathlib import Path + +from agent_comms.commander.order import COMMS +from agent_comms.coordination_database import CoordinationStore +from agent_comms.external_transcript import ExternalEventRow, ExternalTranscript +from agent_comms.field_codec import FieldCodec +from agent_comms.transcript_events import TranscriptEvent +from cutover.carry_external_event_times import ExternalEventTimes + + +root = Path(sys.argv[1]) +root.mkdir(parents=True, exist_ok=False) +source = ExternalTranscript.path_in(COMMS.root) +copy = root / source.name +with closing(CoordinationStore.read_snapshot(source)) as reader: + with closing(CoordinationStore.writer(copy)) as writer: + reader.backup(writer) +with CoordinationStore.reading(copy) as connection: + before, = connection.execute( + f'SELECT count(*) FROM "{ExternalEventRow.declared_name}" WHERE timestamp IS NULL' + ).fetchone() +print('copied undated rows:', before, flush=True) +result = ExternalEventTimes.carry(root, apply=True) +print(json.dumps(result.report('carry_external_event_times', True)), flush=True) +assert ExternalEventTimes.carry(root, apply=True).changed == 0 +count = 0 +with CoordinationStore.reading(copy) as connection: + for encoded, in connection.execute(f'SELECT event FROM "{ExternalEventRow.declared_name}"'): + event = FieldCodec.decode(TranscriptEvent, FieldCodec.parse(encoded)) + assert isinstance(event.timestamp, float) + count += 1 +print('all copied events decoded with recorded float timestamps:', count) +print('source-backed carry idempotent; live store opened read-only') diff --git a/tests/verify_install_publication.py b/tests/verify_install_publication.py new file mode 100644 index 0000000000..a722b85da0 --- /dev/null +++ b/tests/verify_install_publication.py @@ -0,0 +1,122 @@ +"""Exercise candidate Stack publication with installed packages and a private HOME. + +No runtime installation, environment creation, real harness configuration, or +service restart occurs. The actual installed SessionStart executable is invoked +through the privately published PATH and its written hook declaration. +""" + +from __future__ import annotations + +import json +import os +import shlex +import shutil +import subprocess +import sys +from dataclasses import dataclass +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +MAIN = Path("/home/ts/code/projects/agent-comms") +sys.path[:0] = [str(MAIN / "src"), str(ROOT / "tools")] + +from agent_comms.commander.config import CommanderConfig +from agent_comms.field_codec import FieldCodec +from install_gate import ExternalRecord +from agent_comms_hooks import ClaudeHookProgram +from agent_comms_hooks.skill import SkillText +from upgrade_bus import Stack + + +@dataclass(frozen=True) +class InstalledCoreStack(Stack): + @property + def python(self) -> Path: + return MAIN / "stack/.venv/bin/python" + + def environment(self, *paths: Path) -> dict[str, str]: + return super().environment(MAIN / "src", *paths) + + +@dataclass(frozen=True) +class WrittenCommand(ExternalRecord): + command: str + + +@dataclass(frozen=True) +class WrittenRule(ExternalRecord): + hooks: tuple[WrittenCommand, ...] + + +@dataclass(frozen=True) +class WrittenSettings(ExternalRecord): + hooks: dict[str, tuple[WrittenRule, ...]] + retained_user_setting: str + + +@dataclass(frozen=True) +class HookContext: + hookEventName: str + additionalContext: str + + +@dataclass(frozen=True) +class HookResult: + hookSpecificOutput: HookContext + + +def main() -> None: + home = Path("/home/ts/.cache/agent-scratch/pi-worker-cpu/p281/publication-proof") + home.mkdir(mode=0o700) + settings = home / ".claude/settings.json" + settings.parent.mkdir() + settings.write_text(json.dumps({"retained_user_setting": "keep", "hooks": {}})) + stack = InstalledCoreStack(ROOT, home) + stack.publish() + first = settings.read_bytes() + stack.publish() + assert settings.read_bytes() == first, "Repeated refresh duplicated or changed configuration" + written = FieldCodec.decode(WrittenSettings, FieldCodec.parse(first)) + assert written.retained_user_setting == "keep" + environment = stack.environment(ROOT) + environment["PATH"] = str(home / ".local/bin") + os.pathsep + os.environ["PATH"] + for program in ClaudeHookProgram.members(): + name = shlex.split(program.command())[0] + selected = shutil.which(name, path=environment["PATH"]) + assert selected == str(home / ".local/bin" / name), name + for event in program.events: + commands = tuple( + hook.command for entry in written.hooks.get(event, ()) for hook in entry.hooks + ) + assert commands.count(program.command()) == 1, (event, program.command(), commands) + print(f"published/configured: {program.command()} {program.events}") + invoked = subprocess.run( + shlex.split(SkillText.command()), + input=json.dumps( + {"session_id": "private-publication-proof", "hook_event_name": SkillText.events[0]} + ), + env=environment, + capture_output=True, + text=True, + timeout=30, + ) + assert invoked.returncode == 0, invoked.stderr + result = FieldCodec.decode(HookResult, FieldCodec.parse(invoked.stdout)) + assert result.hookSpecificOutput.hookEventName == SkillText.events[0] + sources = (CommanderConfig.packaged_skill(), *CommanderConfig.packaged_rules_files()) + for source in sources: + assert source.name in result.hookSpecificOutput.additionalContext + assert ( + source.read_text(encoding="utf-8", newline="") + in result.hookSpecificOutput.additionalContext + ) + print( + f"SessionStart rc=0; all {len(sources)} packaged files retained; refresh idempotent; user setting preserved" + ) + print( + "Only the private HOME was changed; production publication and service restart were not run" + ) + + +if __name__ == "__main__": + main() diff --git a/tests/verify_maintenance_holder.py b/tests/verify_maintenance_holder.py new file mode 100644 index 0000000000..38007f2340 --- /dev/null +++ b/tests/verify_maintenance_holder.py @@ -0,0 +1,206 @@ +"""Real detached live-phase/SIGKILL proof; reuse installed dependencies, never stage/build. + +This exercises upgrade_bus.live, its detached launcher and the real Core CLI on +an isolated copied bus. Full upgrade/rehearsal preparation belongs to the merger. +""" + +from __future__ import annotations + +import hashlib +import os +import signal +import subprocess +import sys +import time +from collections.abc import Callable +from dataclasses import dataclass +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path[:0] = [str(ROOT / "src"), str(ROOT / "tools")] + +from agent_comms.child_process import Platform, ProcessIdentity +from agent_comms.comms import Comms +from agent_comms.field_codec import FieldCodec +from agent_comms.maintenance_barrier import DeadHolderReleasedPhase, MaintenanceState +from upgrade_bus import Revisions, Stack, copy_bus, detached_live, live + +SCRATCH = Path("/home/ts/.cache/agent-scratch/pi-worker-cpu/p281/holder-death") +BUS = SCRATCH / "copy" + + +@dataclass(frozen=True) +class ExistingStack(Stack): + @property + def python(self) -> Path: + return Path(sys.executable) + + +STACK = ExistingStack(ROOT, SCRATCH / "home") + + +def command(operation: Callable[[], None]) -> tuple[str, ...]: + return ( + sys.executable, + "-c", + f"from verify_maintenance_holder import {operation.__name__}; {operation.__name__}()", + ) + + +def environment() -> dict[str, str]: + return STACK.environment(ROOT / "tests", ROOT / "tools") + + +def busy_owner() -> None: + comms = Comms(BUS) + owner = comms.registry.new_thread( + "busy", frozenset(), str(ROOT), process_identity=ProcessIdentity.capture(os.getpid()) + ) + comms.registry.declare(owner) + assert comms.agents.begin_turn("busy", "p281-real-busy-turn") + signal.signal(signal.SIGTERM, lambda *_: None) + print("busy", flush=True) + sys.stdin.read() + + +def live_worker() -> None: + # Exactly the live installation phase, not a rehearse-only run. Empty pending + # steps need no new revision worktree. The real busy owner holds its drain. + head = subprocess.check_output(('git', 'rev-parse', 'HEAD'), cwd=ROOT, text=True).strip() + live(ROOT, BUS, head, head, (), STACK, STACK, + Revisions(ROOT, SCRATCH / 'unused-revisions'), 'pi-worker-cpu') + + +def caller() -> None: + raise SystemExit(detached_live(command(live_worker), BUS)) + + +def cli(*arguments: str) -> subprocess.CompletedProcess[str]: + return subprocess.run( + ( + sys.executable, + "-c", + "from agent_comms.cli import main; raise SystemExit(main())", + "--root", + str(BUS), + *arguments, + ), + cwd=ROOT, + env=environment(), + capture_output=True, + text=True, + timeout=30, + ) + + +def digest(root: Path) -> str: + value = hashlib.sha256() + for path in sorted(root.rglob("*")): + if path.is_file(): + value.update(str(path.relative_to(root)).encode()) + value.update(path.read_bytes()) + return value.hexdigest() + + +def probe() -> None: + SCRATCH.mkdir(mode=0o700) + STACK.home.mkdir(mode=0o700) + original = SCRATCH / "original" + source = Comms(original) + for name in ("sender", "receiver"): + source.registry.declare(source.registry.new_thread(name, frozenset(), str(ROOT))) + before = digest(original) + copy_bus(original, BUS) + busy = subprocess.Popen( + command(busy_owner), + env=environment(), + cwd=ROOT, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + start_new_session=True, + ) + observer = None + members: tuple[ProcessIdentity, ...] = () + try: + assert busy.stdout is not None and busy.stdout.readline().strip() == "busy" + observer = subprocess.Popen( + command(caller), + env=environment(), + cwd=ROOT, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) + # Use the barrier's declared path, not a second hold file. + from agent_comms.maintenance_barrier import MaintenanceBarrier + + state_path = MaintenanceBarrier(BUS / "registry.json").state_path + deadline = time.monotonic() + 30 + while not state_path.exists(): + assert time.monotonic() < deadline, "Live installer did not enter drain" + assert observer.poll() is None, "Caller exited before entering drain" + time.sleep(0.01) + held = FieldCodec.decode(MaintenanceState, FieldCodec.parse(state_path.read_text())) + assert held.phase.declared_name == "draining" + worker = ProcessIdentity.capture(held.operator.pid) + assert worker == held.operator and worker.pid != observer.pid + assert os.getsid(worker.pid) == worker.pid != os.getsid(observer.pid) + # Kill the calling session first: its separately recorded worker survives. + os.killpg(observer.pid, signal.SIGKILL) + observer.wait(timeout=5) + assert worker.alive(), "Calling-session death killed the detached installer" + print( + f"caller killed; detached holder alive pid={worker.pid} start_time={worker.start_time}" + ) + # Freeze the actual installer job at its recorded drain, then SIGKILL the + # job. No stopped CLI descendant can retain a lock or consume recovery. + os.killpg(worker.pid, signal.SIGSTOP) + members = Platform.current().group_members(worker) + stopped = FieldCodec.decode(MaintenanceState, FieldCodec.parse(state_path.read_text())) + assert stopped.phase.declared_name == "draining" + os.killpg(worker.pid, signal.SIGKILL) + while worker.alive(): + assert time.monotonic() < deadline, "SIGKILL did not retire holder" + time.sleep(0.01) + first = cli("send", "--from", "sender", "--to", "receiver", "--body", "p281-refused-input") + print(f"first send rc={first.returncode}: {first.stdout.strip()}") + assert first.returncode != 0 and "released dead holder" in first.stdout + release_bytes = state_path.read_bytes() + released = FieldCodec.decode(MaintenanceState, FieldCodec.parse(release_bytes.decode())) + assert released.phase is DeadHolderReleasedPhase and released.operator == held.operator + second = cli( + "send", "--from", "sender", "--to", "receiver", "--body", "p281-accepted-input" + ) + print(f"second send rc={second.returncode}: {second.stdout.strip()}") + assert second.returncode == 0, second.stderr + history = cli("history", "--limit", "10") + assert history.returncode == 0 and "p281-accepted-input" in history.stdout + assert "p281-refused-input" not in history.stdout + assert state_path.read_bytes() == release_bytes + assert digest(original) == before + print("release record: " + release_bytes.decode().strip()) + print( + "original unchanged; accepted message retained; rehearsal/environment preparation not run" + ) + finally: + for identity in members: + if identity.alive(): + Platform.current().send(identity, signal.SIGKILL) + if observer is not None and observer.poll() is None: + observer.kill() + observer.wait(timeout=5) + if busy.poll() is None: + busy.kill() + busy.wait(timeout=5) + if busy.stdin is not None: + busy.stdin.close() + if busy.stdout is not None: + busy.stdout.close() + if busy.stderr is not None: + busy.stderr.close() + + +if __name__ == "__main__": + probe() diff --git a/tools/cutover/carry.py b/tools/cutover/carry.py index 3becac6e14..01df3554a9 100644 --- a/tools/cutover/carry.py +++ b/tools/cutover/carry.py @@ -559,8 +559,7 @@ def publish(self, publication, comms, intent, hold, resources, timeout: float) - # No data or route changes: existing readers can keep their shared lock. with comms.route.hold_client_route(): print('Acquired shared route lock for unchanged-format switch', flush=True) - if not publication.published(): - publication.publish_links() + publication.publish_links() publication.release_if_safe(self, comms.owners.maintenance, hold) # Existing exact-incarnation queue owns idle admission and settlement. for original in OwnerRestartSelection.live(intent.owners, comms.registry.snapshot()): diff --git a/tools/cutover/carry_maintenance_identity.py b/tools/cutover/carry_maintenance_identity.py new file mode 100644 index 0000000000..4fb2d60792 --- /dev/null +++ b/tools/cutover/carry_maintenance_identity.py @@ -0,0 +1,97 @@ +"""Carry an old READY maintenance witness from a name to an exact process record. + +A legacy active hold contains no PID or birth time. It cannot honestly be called +abandoned; this one-time converter refuses it rather than fabricating an owner. +There is no legacy reader in the runtime barrier. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Literal, cast + +from agent_comms.errors import RelationViolationError +from agent_comms.field_codec import FieldCodec, Id128Text, compile_decoder +from agent_comms.maintenance_barrier import ( + MaintenanceBarrier, + MaintenanceOperator, + MaintenanceState, + ReadyPhase, +) + +from .carry import Carried, Carry + + +@dataclass(frozen=True, kw_only=True) +class OldMarker: + version: Literal[1] + root: str + generation: int + + +@dataclass(frozen=True, kw_only=True) +class OldReadyState(OldMarker): + operator: str + nonce: str + phase: type[ReadyPhase] + + def carried(self, barrier: MaintenanceBarrier, *, apply: bool) -> Carried: + marker = FieldCodec.decode(OldMarker, FieldCodec.parse(barrier.marker_path.read_text())) + expected = OldMarker(version=1, root=self.root, generation=self.generation) + if marker != expected or self.root != str(barrier.registry_path.parent.resolve()): + raise RelationViolationError("Legacy maintenance witnesses do not identify this root") + converted = MaintenanceState( + version=2, + root=self.root, + generation=self.generation + 1, + operator=MaintenanceOperator.current(), + nonce=Id128Text.mint(), + phase=ReadyPhase, + ) + if apply: + barrier._write_unlocked(converted) + return Carried(1, rewritten=tuple(path.name for path in barrier.witness_paths)) + + +@dataclass(frozen=True, kw_only=True) +class IdentifiedState(MaintenanceState): + def carried(self, barrier: MaintenanceBarrier, *, apply: bool) -> Carried: + barrier._current_unlocked() + return Carried(0) + + +StoredMaintenanceState = IdentifiedState | OldReadyState + + +class MaintenanceIdentity(Carry): + @classmethod + def precedes(cls, later: type[Carry]) -> bool: + return later is not cls + + @classmethod + def backup_files(cls, root: Path) -> tuple[Path, ...]: + return MaintenanceBarrier(root / "registry.json").witness_paths + + @classmethod + def before_stop(cls, root: Path, *, apply: bool) -> Carried: + return cls.carry(root, apply=apply) + + @classmethod + def carry(cls, root: Path, *, apply: bool) -> Carried: + barrier = MaintenanceBarrier(root / "registry.json") + if not barrier.state_path.exists(): + if barrier.marker_path.exists(): + raise RelationViolationError("Maintenance has a marker but no stored holder") + return Carried(0) + state = cast( + StoredMaintenanceState, + compile_decoder(StoredMaintenanceState).decode( + FieldCodec.parse(barrier.state_path.read_text()) + ), + ) + return state.carried(barrier, apply=apply) + + +if __name__ == "__main__": + MaintenanceIdentity.main() diff --git a/tools/proofs/cn_phase3_record.py b/tools/proofs/cn_phase3_record.py index d925660ffd..bbadb40bd5 100644 --- a/tools/proofs/cn_phase3_record.py +++ b/tools/proofs/cn_phase3_record.py @@ -7,19 +7,15 @@ from __future__ import annotations import argparse -import json import os import shutil import subprocess -import sys import time from dataclasses import replace from pathlib import Path -from agent_comms import cli as comms_cli from agent_comms.change_report import FileLine from agent_comms.comms import Comms -from agent_comms.field_codec import FieldCodec from agent_comms.goal_actions import CreateProblemGoalAction, SetGoalAction from agent_comms.goals import CountCheckResult, OrderCheck, RequestGoalOrder from agent_comms.progress_claims import MergedClaim @@ -50,22 +46,7 @@ def cli[Caller: ToolCaller](caller: str, tool: ToolRequest[Caller]) -> str: - arguments = FieldCodec.encode(tool) - del arguments[tool.family_discriminator] - command = [ - sys.executable, - "-m", - comms_cli.__name__, - "--root", - str(ROOT), - "tool", - "--from", - caller, - "--tool", - tool.declared_name, - "--arguments", - json.dumps(arguments), - ] + command = tool.cli_command(ToolCaller.from_actor(caller)).argv(ROOT) result = subprocess.run( command, capture_output=True, diff --git a/tools/proofs/cn_progress_read_only.py b/tools/proofs/cn_progress_read_only.py index 5065895634..05d2364053 100644 --- a/tools/proofs/cn_progress_read_only.py +++ b/tools/proofs/cn_progress_read_only.py @@ -63,7 +63,7 @@ def verify() -> None: assert not MARKER.exists() sha = CheckedCommit.git(str(checkout), "rev-parse", "HEAD").strip() assert not check.recorded(sha, str(checkout)) - check.at(sha, str(checkout)) # Deliberately explicit, never a board read. + check.at(sha, str(checkout), comms.registry.storage_rule.checkout_base) assert MARKER.read_bytes() == MARKER_CONTENT assert len(board(todo).readings) == 1 assert MARKER.read_bytes() == MARKER_CONTENT diff --git a/tools/publish_runtime.py b/tools/publish_runtime.py index a105df825d..bee1dc983d 100644 --- a/tools/publish_runtime.py +++ b/tools/publish_runtime.py @@ -6,35 +6,46 @@ from __future__ import annotations import argparse -from contextlib import AbstractContextManager, ExitStack, contextmanager, nullcontext, redirect_stdout -from dataclasses import dataclass, field -from abc import abstractmethod -import hashlib import fcntl -from io import StringIO -from importlib.metadata import distribution +import hashlib import json import os -from pathlib import Path import subprocess import sys +from abc import abstractmethod +from contextlib import ( + AbstractContextManager, + ExitStack, + contextmanager, + nullcontext, + redirect_stdout, +) +from dataclasses import dataclass, field +from importlib.metadata import distribution +from io import StringIO +from pathlib import Path from typing import Annotated -from agent_comms.active_route import read_active_route, active_route_path, acquire_exclusive_route_lock -from agent_comms.comms import wire +from agent_comms.active_route import ( + acquire_exclusive_route_lock, + active_route_path, + read_active_route, +) from agent_comms.cli_commands import StatusCliCommand -from agent_comms.field_codec import FieldCodec, PathText, SubclassUnion, NullMember, RecordMember +from agent_comms.comms import wire +from agent_comms.field_codec import FieldCodec, NullMember, PathText, RecordMember, SubclassUnion from agent_comms.harness_install import Harness, replace_link from agent_comms.launch_runtime import LaunchRuntime from agent_comms.maintenance_barrier import MaintenanceOperator, MaintenanceState -from agent_comms.owner_cutover import StoppedOwnerInstallation, MaintainedOwnerCutover +from agent_comms.owner_cutover import MaintainedOwnerCutover from agent_comms.owner_lifecycle import OwnerRestartSelection, OwnerRestartResult from agent_comms.owner_restart import OwnerRestartHandoff, OwnerRestartRequest, StoppedOwnerBatch from agent_comms.registry_document import RegistrySnapshot from agent_comms.threads import Thread from agent_comms.store_files import _atomic_write_text, _store_lock -from .cutover.carry import PendingConversion, PreparedCarries -from .cutover.backup import StoreBackup + +from cutover.backup import StoreBackup +from cutover.carry import PendingConversion, PreparedCarries class PublicationIntent(SubclassUnion, affix=''): @@ -124,7 +135,7 @@ def publish(self, root: Path, *, checkout: Path, timeout: float = 1200, input is resubmitted. Unrelated maintenance and changed incarnations are refused rather than guessed. """ - installed = StoppedOwnerInstallation.current_pointer().resolve(strict=True) + installed = LaunchRuntime.current_pointer().resolve(strict=True) if Path(sys.executable).parent.resolve().parent != installed: raise RuntimeError('Drain and stop must run with the installed source interpreter') carries = PreparedCarries.at(checkout, root, self.backup(root), self.published()) @@ -135,11 +146,10 @@ def publish(self, root: Path, *, checkout: Path, timeout: float = 1200, raise ValueError('Publication requires the selected active root') comms = wire(route) barrier = comms.owners.maintenance - operator = MaintenanceOperator('runtime-publication:' + hashlib.sha256( - os.fsencode(self.runtime.resolve())).hexdigest()) + operator = MaintenanceOperator.current() intent_path = root / 'runtime-publication.json' with _store_lock(root / 'runtime-publication'): - previous = StoppedOwnerInstallation.current_pointer().resolve(strict=True) + previous = LaunchRuntime.current_pointer().resolve(strict=True) intent = PublicationIntent.read(intent_path) intent.require_target(self.runtime, previous) current = barrier.read() @@ -176,7 +186,7 @@ def backup(self, root: Path) -> StoreBackup: return StoreBackup(root, active_route_path(), path) def published(self) -> bool: - return StoppedOwnerInstallation.current_pointer().resolve(strict=True) == self.runtime.resolve() + return LaunchRuntime.current_pointer().resolve(strict=True) == self.runtime.resolve() def release_if_safe(self, carries: PreparedCarries, barrier, hold: MaintenanceState) -> None: if barrier.read() == hold and carries.cleanup(self.published()): @@ -197,13 +207,14 @@ def selected_route(self, root: Path): os.close(descriptor) def publish_links(self) -> None: - """Move declarations and the pointer, without killing unrelated processes.""" - current = StoppedOwnerInstallation.current_pointer() - if self.published(): - raise ValueError(f'Publication would retire the runtime it publishes: {self.runtime}') + """Refresh the declared installation, including an already-current runtime.""" + current = LaunchRuntime.current_pointer() + selected = self.runtime.resolve(strict=True) for link, target in self.links.items(): replace_link(link, current / target.relative_to(self.runtime)) - replace_link(current, self.runtime) + replace_link(current, selected) + for harness in Harness.present(Path.home()): + harness.configure() @dataclass(frozen=True) @@ -237,10 +248,9 @@ def complete(self, stopped: StoppedOwnerBatch) -> tuple[OwnerRestartResult, ...] stopped.handoff, self.maintenance_hold, self.carries.checkout) result = subprocess.run([ - self.runtime / 'bin/python', '-I', '-B', '-c', - 'import sys; sys.path.insert(0, sys.argv.pop(1)); ' - 'from tools.publish_runtime import main; main()', - str(request.checkout), '--complete-stopped', + self.runtime / 'bin/python', '-E', '-s', '-B', + str(request.checkout / Path(__file__).parent.name / Path(__file__).name), + '--complete-stopped', str(stopped.wire_descriptor), str(self.route_descriptor), ], input=json.dumps(FieldCodec.encode(request)), text=True, stdout=subprocess.PIPE, pass_fds=(stopped.wire_descriptor, self.route_descriptor), check=True) @@ -269,8 +279,7 @@ def complete(self, wire_descriptor: int, route_descriptor: int) -> tuple[OwnerRe if StatusCliCommand().run(comms, output): raise RuntimeError('Converted bus status failed: ' + output.getvalue()) print('Converted bus status: ' + output.getvalue().strip(), flush=True) - if not publication.published(): - publication.publish_links() + publication.publish_links() stopped = StoppedOwnerBatch.accept(comms.owners, wire_descriptor, self.handoff) comms.owners.maintenance.clear_stopped(self.maintenance_hold, wire_descriptor) fcntl.flock(route_descriptor, fcntl.LOCK_UN) @@ -281,6 +290,8 @@ def main() -> None: parser = argparse.ArgumentParser(description=__doc__) actions = parser.add_mutually_exclusive_group(required=True) actions.add_argument('--check', action='store_true') + actions.add_argument('--refresh', action='store_true', + help='Refresh this installed runtime\'s declared launchers and harness config') actions.add_argument('--complete-stopped', type=int, nargs=2, metavar=('WIRE_FD', 'ROUTE_FD')) options = parser.parse_args() if options.complete_stopped is not None: @@ -290,6 +301,9 @@ def main() -> None: print(json.dumps(FieldCodec.encode(result))) return publication = RuntimePublication.installed() + if options.refresh: + publication.publish_links() + return route = read_active_route() if route is None: raise ValueError('Runtime publication requires the installed route') diff --git a/tools/upgrade_bus.py b/tools/upgrade_bus.py index 8e7c9a6eb7..92eac349b1 100644 --- a/tools/upgrade_bus.py +++ b/tools/upgrade_bus.py @@ -44,18 +44,18 @@ # tree's src, never the src the checkout being upgraded has installed. sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) -from cutover.carry import Carried, CarryChain, CarryLedger, Step, TodoColumnCarry -from cutover.backup import Backup, StoreBackup, UpgradeRefused, bus_files, transcripts_held - import agent_comms.activity_checkpoint # noqa: F401 (declares the activity runtime files) from agent_comms.external_transcript import ExternalTranscript -from agent_comms.field_codec import FieldCodec +from agent_comms.field_codec import FieldCodec, Id128Text from agent_comms.history_result import EverythingHistoryPageResult from agent_comms.host_services import HostService from agent_comms.maintenance_barrier import HOLDER_NONCE, MaintenanceBarrier, MaintenanceOperator from agent_comms.registry_store import RegistryStore from agent_comms.store_files import _store_lock from agent_comms.todos import PlannedInstallDeadline +import publish_runtime +from cutover.backup import Backup, StoreBackup, UpgradeRefused, bus_files, transcripts_held +from cutover.carry import Carried, CarryChain, CarryLedger, Step, TodoColumnCarry REPO = Path(__file__).resolve().parents[1] CLIENT_IMPORTS = "import toad.app, agent_comms.acp" @@ -122,6 +122,13 @@ def imports(self) -> None: def prepare_native(self) -> None: run(self.checkout / "stack/bin/prepare-pi-native", cwd=self.checkout) + def publish(self) -> None: + """The selected interpreter owns console metadata and harness declarations.""" + run( + self.python, publish_runtime.__file__, "--refresh", + cwd=self.checkout, env=self.environment(), + ) + def environment(self, *paths: Path) -> dict: """``paths`` first; then this checkout's ``src``. ``home`` holds the route its clients read.""" return {**os.environ, "HOME": str(self.home), @@ -261,7 +268,7 @@ def rehearse(root: Path, pending: tuple[Step, ...], stack: Stack, revisions: Rev stack.prepare_native() # Hold the copy before reading it, exactly as the live run holds the bus. os.environ[HOLDER_NONCE] = MaintenanceBarrier(copy / "registry.json").begin( - MaintenanceOperator("upgrade-bus")).nonce + MaintenanceOperator.current()).nonce with transcripts_held(copy): before = bus_files(copy) stopping = before_stop(copy, pending, stack, revisions) @@ -358,6 +365,7 @@ def rehearsal(self) -> dict: return {"unchanged": self.target} def live(self) -> dict: + self.installed.publish() PlannedInstallDeadline.after(self.installer, time.time()).plan(self.root) return {"unchanged": self.target} @@ -366,15 +374,15 @@ def live(checkout: Path, root: Path, current: str, target: str, pending: tuple[S installed: Stack, staged: Stack, revisions: Revisions, installer: str) -> dict: """The rehearsal's passes under the rehearsal's stack; ``installed`` only stops, starts and reads back its own owners.""" - running = [thread["name"] for thread in installed.cli(root, "threads")["threads"] - if thread["role"] == "agent" and thread["status"] == "running"] before_stop(root, pending, staged, revisions) + running = [thread["name"] for thread in staged.cli(root, "threads")["threads"] + if thread["role"] == "agent" and thread["status"] == "running"] maintenance = MaintenanceBarrier(RegistryStore.at(root).path) - held = maintenance.begin(MaintenanceOperator("upgrade-bus")) + held = maintenance.begin(MaintenanceOperator.current()) os.environ[HOLDER_NONCE] = held.nonce # this run's own CLI calls read under its hold lock = installed.lock try: - installed.owners(root, "stop", running) + staged.owners(root, "stop", running) with transcripts_held(root): backup = Backup.take(root, ROUTE, (root,), root.parent / f"{root.name}.upgrade-{time.strftime('%Y%m%dT%H%M%S')}") @@ -391,12 +399,13 @@ def live(checkout: Path, root: Path, current: str, target: str, pending: tuple[S installed.relock() installed.prepare_native() passed = carry_and_read(root, backup.files(), pending, staged, installed, revisions, []) + installed.publish() maintenance.clear(held) services = HostService.restart_running(installed.python) PlannedInstallDeadline.after(installer, time.time()).plan(root) except BaseException as error: state = maintenance.read() - state.phase.hold_for(maintenance, state, MaintenanceOperator("upgrade-bus")) + state.phase.hold_for(maintenance, state, MaintenanceOperator.current()) print(f"upgrade-bus: stopped after the backup ({type(error).__name__}: {error}); " f"maintenance held, checkout and bus preserved; backup: {backup.path}", file=sys.stderr) @@ -406,6 +415,31 @@ def live(checkout: Path, root: Path, current: str, target: str, pending: tuple[S "backup": str(backup.path)} +def detached_live(command: tuple[str, ...], root: Path) -> int: + """The live installer is a new session, not the monitor's owned child group. + + Its output descriptors belong to files, not a caller that may disappear. + Deliberately do not borrow a turn-owned child lifetime or use Popen's cleanup + context: retiring the observer must not terminate this independent install. + """ + logs = root.parent / f".{root.name}.installer-{Id128Text.mint()}" + logs.mkdir(mode=0o700) + output, errors = logs / "result.json", logs / "stderr.log" + with output.open("xb") as stdout, errors.open("xb") as stderr: + process = subprocess.Popen( + command, stdin=subprocess.DEVNULL, stdout=stdout, stderr=stderr, + start_new_session=True, + ) + print(f"upgrade-bus: detached installer pid={process.pid}; diagnostics {errors}", + file=sys.stderr, flush=True) + code = process.wait() + if code: + print(f"upgrade-bus: detached installer failed ({code}); see {errors}", file=sys.stderr) + else: + print(output.read_text(), end="") + return code + + def stopped(signum: int, frame: object) -> None: """A stop signal runs the same abort as a failure; a second one cannot cut that abort short.""" for each in (signal.SIGTERM, signal.SIGINT): @@ -423,6 +457,10 @@ def main() -> None: help="The unit running this install; the plan names it") parser.add_argument("--rehearse", action="store_true", help="Carry and read a copy of the bus; change nothing") args = parser.parse_args() + if not args.rehearse and os.getsid(0) != os.getpid(): + raise SystemExit(detached_live( + (sys.executable, str(Path(__file__).resolve()), *sys.argv[1:]), args.root, + )) for signum in (signal.SIGTERM, signal.SIGINT): signal.signal(signum, stopped) try: