From 561b51c780a042461fd88c9e2b75cbe51b2fb6df Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 19:07:59 -0400 Subject: [PATCH 1/5] Merged orders serve the phase their order served Rotate crashed with AttributeError: 'MergedProblemOrder' object has no attribute 'serves', since PlanPhase.serving reads every order's phase and #2004's merged order kept none. Co-Authored-By: Claude Opus 5.5 --- src/agent_comms/todos.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/agent_comms/todos.py b/src/agent_comms/todos.py index 19c45babcb..678b22a01c 100644 --- a/src/agent_comms/todos.py +++ b/src/agent_comms/todos.py @@ -1325,6 +1325,11 @@ class MergedProblemOrder(ProblemOrder): def id(self) -> str: return self.original.id + @property + def serves(self) -> type[OrderPhase]: + """A merge finishes the phase its order served.""" + return self.original.serves + def require_close(self) -> None: pass From f156121fba8198046b5401c7df388516a765f66b Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 19:11:32 -0400 Subject: [PATCH 2/5] Copy the bus's SQLite stores with SQLite's backup so a practice install never races a checkpoint Co-Authored-By: Claude Opus 5.5 --- tools/upgrade_bus.py | 275 +++++++++++++++++++++++++++++++++---------- 1 file changed, 212 insertions(+), 63 deletions(-) diff --git a/tools/upgrade_bus.py b/tools/upgrade_bus.py index ebab0fbfbb..3132cc50ef 100644 --- a/tools/upgrade_bus.py +++ b/tools/upgrade_bus.py @@ -23,6 +23,7 @@ repair. It never resets live main or replaces a possibly changed bus. Nothing is sent and no input is replayed: owners start by the normal start path. """ + from __future__ import annotations import argparse @@ -30,6 +31,7 @@ import os import shutil import signal +import sqlite3 import stat import subprocess import sys @@ -37,7 +39,7 @@ import time from abc import ABC, abstractmethod from concurrent.futures import ThreadPoolExecutor -from contextlib import contextmanager, suppress +from contextlib import closing, contextmanager, suppress from dataclasses import dataclass, replace from pathlib import Path @@ -49,6 +51,7 @@ from cutover.backup import Backup, UpgradeRefused, bus_files, transcripts_held import agent_comms.activity_checkpoint # noqa: F401 (declares the activity runtime files) +from agent_comms.coordination_database import CoordinationStore from agent_comms.external_transcript import ExternalTranscript from agent_comms.field_codec import FieldCodec from agent_comms.history_result import EverythingHistoryPageResult @@ -65,10 +68,13 @@ def run(*command: str | Path, cwd: Path | None = None, env: dict | None = None) -> str: print("+", " ".join(map(str, command)), file=sys.stderr, flush=True) - completed = subprocess.run([str(part) for part in command], cwd=cwd, env=env, - capture_output=True, text=True) + completed = subprocess.run( + [str(part) for part in command], cwd=cwd, env=env, capture_output=True, text=True + ) if completed.returncode: - raise UpgradeRefused(f"{command[0]} failed: {completed.stderr.strip() or completed.stdout.strip()}") + raise UpgradeRefused( + f"{command[0]} failed: {completed.stderr.strip() or completed.stdout.strip()}" + ) return completed.stdout @@ -108,14 +114,23 @@ def require_uv(self) -> None: def relock(self) -> None: """OpenHCSDev packages track their mains; the lock only records this resolution.""" - run(self.uv, "lock", "--project", "stack", "--upgrade", "--python", "3.14", - cwd=self.checkout) + run( + self.uv, + "lock", + "--project", + "stack", + "--upgrade", + "--python", + "3.14", + cwd=self.checkout, + ) self.sync() self.imports() def sync(self) -> None: - run(self.uv, "sync", "--project", "stack", "--locked", "--python", "3.14", - cwd=self.checkout) + run( + self.uv, "sync", "--project", "stack", "--locked", "--python", "3.14", cwd=self.checkout + ) def imports(self) -> None: run(self.python, "-c", CLIENT_IMPORTS, cwd=self.checkout, env=self.environment()) @@ -125,24 +140,39 @@ def prepare_native(self) -> None: 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), - "PYTHONPATH": os.pathsep.join(map(str, (*paths, self.checkout / "src")))} + return { + **os.environ, + "HOME": str(self.home), + "PYTHONPATH": os.pathsep.join(map(str, (*paths, self.checkout / "src"))), + } def cli(self, root: Path, *arguments: str) -> dict: """One JSON CLI result for this bus alone, never the route's, waiting while the route index is still being built.""" for _ in range(240): completed = subprocess.run( - [str(self.python), "-c", "from agent_comms.cli import main; main()", "--root", str(root), - *arguments], cwd=self.checkout, env=self.environment(), capture_output=True, - text=True) + [ + str(self.python), + "-c", + "from agent_comms.cli import main; main()", + "--root", + str(root), + *arguments, + ], + cwd=self.checkout, + env=self.environment(), + capture_output=True, + text=True, + ) if "Route index covers" not in completed.stdout: break time.sleep(5) result = json.loads(completed.stdout or "{}") if completed.returncode or "error" in result: - raise UpgradeRefused(f"agent-comms {' '.join(arguments)}: " - f"{completed.stdout.strip() or completed.stderr.strip()}") + raise UpgradeRefused( + f"agent-comms {' '.join(arguments)}: " + f"{completed.stdout.strip() or completed.stderr.strip()}" + ) return result def owners(self, root: Path, command: str, names: list[str]) -> None: @@ -154,8 +184,11 @@ def verify(self, root: Path) -> dict[str, int]: threads = self.cli(root, "threads")["threads"] present = self.cli(root, "who") history = self.cli(root, "history", "--limit", "50") # #any: every channel - return {"threads": len(threads), "who": len(present["who"]), - "history": len(history[EverythingHistoryPageResult.messages_key])} + return { + "threads": len(threads), + "who": len(present["who"]), + "history": len(history[EverythingHistoryPageResult.messages_key]), + } @dataclass @@ -183,7 +216,9 @@ def tree(self, revision: str) -> Path: def close(self) -> None: for path in self.trees.values(): - subprocess.run(["git", "-C", str(self.repo), "worktree", "remove", "--force", str(path)]) + subprocess.run( + ["git", "-C", str(self.repo), "worktree", "remove", "--force", str(path)] + ) def carry(step: Step, root: Path, stack: Stack, revisions: Revisions, *flags: str) -> Carried: @@ -193,7 +228,8 @@ def carry(step: Step, root: Path, stack: Stack, revisions: Revisions, *flags: st if "--before-stop" in flags: return TodoColumnCarry.before_stop(root, apply=True) return TodoColumnCarry.carry_step( - root, step, CarryChain.at(revisions.repo, step.revision), apply=True) + root, step, CarryChain.at(revisions.repo, step.revision), apply=True + ) tree = revisions.tree(step.revision) environment = stack.environment(tree / "src", tree / "tools") worker = ( @@ -202,8 +238,17 @@ def carry(step: Step, root: Path, stack: Stack, revisions: Revisions, *flags: st "print(json.dumps(asdict(getattr(step,sys.argv[3])(Path(sys.argv[4]),apply=True))))" ) operation = "before_stop" if "--before-stop" in flags else "carry" - output = run(stack.python, "-c", worker, f"cutover.{step.name}", step.declaration, - operation, root, cwd=tree, env=environment) + output = run( + stack.python, + "-c", + worker, + f"cutover.{step.name}", + step.declaration, + operation, + root, + cwd=tree, + env=environment, + ) return FieldCodec.decode(Carried, FieldCodec.parse(output.strip().splitlines()[-1])) @@ -212,8 +257,10 @@ def before_stop( ) -> list[dict]: """What the running installation needs before its own stop command reads the bus.""" with _store_lock(root / "wire"): - return [carry(step, root, stack, revisions, "--before-stop", "--apply").report( - step.name, True) for step in pending] + return [ + carry(step, root, stack, revisions, "--before-stop", "--apply").report(step.name, True) + for step in pending + ] def carry_all( @@ -223,8 +270,10 @@ def carry_all( for step in pending: with _store_lock(root / "wire"): result = CarryLedger(root).apply( - CarryChain.at(revisions.repo, step.revision), step.name, - lambda: carry(step, root, stack, revisions, "--apply")) + CarryChain.at(revisions.repo, step.revision), + step.name, + lambda: carry(step, root, stack, revisions, "--apply"), + ) results.append(result.report(step.name, True)) return results @@ -234,15 +283,38 @@ def copy_bus(root: Path, copy: Path) -> None: # The maintenance files describe the live root; a copy is a new root and starts open. held = set(MaintenanceBarrier(root / "registry.json").witness_paths) + stores: list[Path] = [] + def skipped(directory: str, names: list[str]) -> set[str]: - return {name for name in names - if Path(directory, name) in held - or stat.S_ISSOCK(os.lstat(os.path.join(directory, name)).st_mode)} + # SQLite stores are copied whole by SQLite below; a file copy races their + # writers, who may checkpoint and remove a companion mid-copy. + found = [ + Path(directory, name) + for name in names + if Path(directory, name).is_file() + and not Path(directory, name).is_symlink() + and CoordinationStore.is_database(Path(directory, name)) + ] + stores.extend(found) + store_files = {file.name for store in found for file in CoordinationStore.files(store)} + return { + name + for name in names + if name in store_files + or Path(directory, name) in held + or stat.S_ISSOCK(os.lstat(os.path.join(directory, name)).st_mode) + } from agent_comms.store_files import _store_lock with _store_lock(root / "wire"): shutil.copytree(root, copy, symlinks=True, ignore=skipped) + for store in stores: + with ( + closing(CoordinationStore.read_snapshot(store)) as source, + closing(sqlite3.connect(copy / store.relative_to(root))) as target, + ): + source.backup(target) def rehearsal_home(scratch: Path, root: Path, copy: Path) -> Path: @@ -269,8 +341,11 @@ def rehearse(root: Path, pending: tuple[Step, ...], stack: Stack, revisions: Rev stack = replace(stack, home=rehearsal_home(Path(scratch), root, copy)) 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 + os.environ[HOLDER_NONCE] = ( + MaintenanceBarrier(copy / "registry.json") + .begin(MaintenanceOperator("upgrade-bus")) + .nonce + ) with transcripts_held(copy): before = bus_files(copy) stopping = before_stop(copy, pending, stack, revisions) @@ -279,8 +354,15 @@ def rehearse(root: Path, pending: tuple[Step, ...], stack: Stack, revisions: Rev return carry_and_read(copy, before, pending, stack, stack, revisions, stopping) -def carry_and_read(root: Path, before: dict[str, str], pending: tuple[Step, ...], staged: Stack, - reader: Stack, revisions: Revisions, stopping: list[dict]) -> dict: +def carry_and_read( + root: Path, + before: dict[str, str], + pending: tuple[Step, ...], + staged: Stack, + reader: Stack, + revisions: Revisions, + stopping: list[dict], +) -> dict: """The pass a rehearsal and a live install share once ``before`` was read under the transcript hold. The hold is released first: SQLite cannot lend its writer lock to a step's process, so a step that writes the transcript store could not run under it.""" @@ -292,7 +374,9 @@ def upgrade(checkout: Path, root: Path, target: str, installer: str, rehearse_on current = run("git", "-C", checkout, "rev-parse", "HEAD").strip() run("git", "-C", checkout, "fetch", "-q", "origin") target = run("git", "-C", checkout, "rev-parse", "--verify", f"{target}^{{commit}}").strip() - if subprocess.run(["git", "-C", str(checkout), "merge-base", "--is-ancestor", current, target]).returncode: + if subprocess.run( + ["git", "-C", str(checkout), "merge-base", "--is-ancestor", current, target] + ).returncode: raise UpgradeRefused(f"{target[:12]} does not follow the checkout's {current[:12]}") if run("git", "-C", checkout, "status", "--porcelain", "--untracked-files=no").strip(): raise UpgradeRefused("The checkout has local changes") @@ -310,10 +394,15 @@ def upgrade(checkout: Path, root: Path, target: str, installer: str, rehearse_on try: staged = Stack(revisions.tree(target)) staged.relock() - install = Install.of(checkout, root, current, target, pending, installed, staged, revisions, - installer) - report = {"from": current, "to": target, "pending": [step.name for step in pending], - "rehearsal": install.rehearsal()} + install = Install.of( + checkout, root, current, target, pending, installed, staged, revisions, installer + ) + report = { + "from": current, + "to": target, + "pending": [step.name for step in pending], + "rehearsal": install.rehearsal(), + } if rehearse_only: return report report["live"] = install.live() @@ -337,11 +426,23 @@ class Install(ABC): installer: str @classmethod - def of(cls, checkout: Path, root: Path, current: str, target: str, pending: tuple[Step, ...], - installed: Stack, staged: Stack, revisions: Revisions, installer: str) -> Install: + def of( + cls, + checkout: Path, + root: Path, + current: str, + target: str, + pending: tuple[Step, ...], + installed: Stack, + staged: Stack, + revisions: Revisions, + installer: str, + ) -> Install: unchanged = current == target and staged.lock.read_bytes() == installed.lock.read_bytes() member = (ChangedInstall, UnchangedInstall)[unchanged] - return member(checkout, root, current, target, pending, installed, staged, revisions, installer) + return member( + checkout, root, current, target, pending, installed, staged, revisions, installer + ) @abstractmethod def rehearsal(self) -> dict: ... @@ -355,8 +456,17 @@ def rehearsal(self) -> dict: return rehearse(self.root, self.pending, self.staged, self.revisions) def live(self) -> dict: - return live(self.checkout, self.root, self.current, self.target, self.pending, - self.installed, self.staged, self.revisions, self.installer) + return live( + self.checkout, + self.root, + self.current, + self.target, + self.pending, + self.installed, + self.staged, + self.revisions, + self.installer, + ) class UnchangedInstall(Install): @@ -371,12 +481,24 @@ def live(self) -> dict: return {"unchanged": self.target} -def live(checkout: Path, root: Path, current: str, target: str, pending: tuple[Step, ...], - installed: Stack, staged: Stack, revisions: Revisions, installer: str) -> dict: +def live( + checkout: Path, + root: Path, + current: str, + target: str, + pending: tuple[Step, ...], + 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"] + 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) maintenance = MaintenanceBarrier(RegistryStore.at(root).path) held = maintenance.begin(MaintenanceOperator("upgrade-bus")) @@ -385,13 +507,20 @@ def live(checkout: Path, root: Path, current: str, target: str, pending: tuple[S try: installed.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')}") + backup = Backup.take( + root, + ROUTE, + (root,), + root.parent / f"{root.name}.upgrade-{time.strftime('%Y%m%dT%H%M%S')}", + ) except BaseException as error: maintenance.clear(held) installed.owners(root, "start", running) - print(f"upgrade-bus: aborted before the backup ({type(error).__name__}: {error}); " - f"maintenance cleared and {running} restarted", file=sys.stderr) + print( + f"upgrade-bus: aborted before the backup ({type(error).__name__}: {error}); " + f"maintenance cleared and {running} restarted", + file=sys.stderr, + ) raise shutil.copy2(lock, backup.path / "uv.lock") if lock.exists() else None try: @@ -406,13 +535,19 @@ def live(checkout: Path, root: Path, current: str, target: str, pending: tuple[S except BaseException as error: state = maintenance.read() state.phase.hold_for(maintenance, state, MaintenanceOperator("upgrade-bus")) - 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) + 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, + ) raise installed.owners(root, "start", running) - return {"stopped_and_started": running, **passed, "services": services, - "backup": str(backup.path)} + return { + "stopped_and_started": running, + **passed, + "services": services, + "backup": str(backup.path), + } def stopped(signum: int, frame: object) -> None: @@ -423,19 +558,33 @@ def stopped(signum: int, frame: object) -> None: def main() -> None: - parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) - parser.add_argument("--checkout", type=lambda value: Path(value).expanduser().resolve(), default=REPO) - parser.add_argument("--root", type=lambda value: Path(value).expanduser().resolve(), - default=Path.home() / ".agent-comms") + parser = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + parser.add_argument( + "--checkout", type=lambda value: Path(value).expanduser().resolve(), default=REPO + ) + parser.add_argument( + "--root", + type=lambda value: Path(value).expanduser().resolve(), + default=Path.home() / ".agent-comms", + ) parser.add_argument("--to", default="origin/main") - parser.add_argument("--installer", required=True, - 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") + parser.add_argument( + "--installer", required=True, 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() for signum in (signal.SIGTERM, signal.SIGINT): signal.signal(signum, stopped) try: - print(json.dumps(upgrade(args.checkout, args.root, args.to, args.installer, args.rehearse), indent=2)) + print( + json.dumps( + upgrade(args.checkout, args.root, args.to, args.installer, args.rehearse), indent=2 + ) + ) except UpgradeRefused as error: raise SystemExit(f"upgrade-bus: {error}") From c241806ffe8701158eb434f8fcaaa2a7fb6ba01b Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 19:14:20 -0400 Subject: [PATCH 3/5] Keep each copied store's owner-only mode Co-Authored-By: Claude Opus 5.5 --- tools/upgrade_bus.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tools/upgrade_bus.py b/tools/upgrade_bus.py index 3132cc50ef..a81fbd5b6e 100644 --- a/tools/upgrade_bus.py +++ b/tools/upgrade_bus.py @@ -315,6 +315,7 @@ def skipped(directory: str, names: list[str]) -> set[str]: closing(sqlite3.connect(copy / store.relative_to(root))) as target, ): source.backup(target) + shutil.copymode(store, copy / store.relative_to(root)) def rehearsal_home(scratch: Path, root: Path, copy: Path) -> Path: From b5a4d4c96296f87b6da5ffcf285d698b37f1d055 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 19:17:49 -0400 Subject: [PATCH 4/5] Leave the bus copy to its own order; this PR fixes the merged order's phase only Co-Authored-By: Claude Opus 5.5 --- tools/upgrade_bus.py | 276 ++++++++++--------------------------------- 1 file changed, 63 insertions(+), 213 deletions(-) diff --git a/tools/upgrade_bus.py b/tools/upgrade_bus.py index a81fbd5b6e..ebab0fbfbb 100644 --- a/tools/upgrade_bus.py +++ b/tools/upgrade_bus.py @@ -23,7 +23,6 @@ repair. It never resets live main or replaces a possibly changed bus. Nothing is sent and no input is replayed: owners start by the normal start path. """ - from __future__ import annotations import argparse @@ -31,7 +30,6 @@ import os import shutil import signal -import sqlite3 import stat import subprocess import sys @@ -39,7 +37,7 @@ import time from abc import ABC, abstractmethod from concurrent.futures import ThreadPoolExecutor -from contextlib import closing, contextmanager, suppress +from contextlib import contextmanager, suppress from dataclasses import dataclass, replace from pathlib import Path @@ -51,7 +49,6 @@ from cutover.backup import Backup, UpgradeRefused, bus_files, transcripts_held import agent_comms.activity_checkpoint # noqa: F401 (declares the activity runtime files) -from agent_comms.coordination_database import CoordinationStore from agent_comms.external_transcript import ExternalTranscript from agent_comms.field_codec import FieldCodec from agent_comms.history_result import EverythingHistoryPageResult @@ -68,13 +65,10 @@ def run(*command: str | Path, cwd: Path | None = None, env: dict | None = None) -> str: print("+", " ".join(map(str, command)), file=sys.stderr, flush=True) - completed = subprocess.run( - [str(part) for part in command], cwd=cwd, env=env, capture_output=True, text=True - ) + completed = subprocess.run([str(part) for part in command], cwd=cwd, env=env, + capture_output=True, text=True) if completed.returncode: - raise UpgradeRefused( - f"{command[0]} failed: {completed.stderr.strip() or completed.stdout.strip()}" - ) + raise UpgradeRefused(f"{command[0]} failed: {completed.stderr.strip() or completed.stdout.strip()}") return completed.stdout @@ -114,23 +108,14 @@ def require_uv(self) -> None: def relock(self) -> None: """OpenHCSDev packages track their mains; the lock only records this resolution.""" - run( - self.uv, - "lock", - "--project", - "stack", - "--upgrade", - "--python", - "3.14", - cwd=self.checkout, - ) + run(self.uv, "lock", "--project", "stack", "--upgrade", "--python", "3.14", + cwd=self.checkout) self.sync() self.imports() def sync(self) -> None: - run( - self.uv, "sync", "--project", "stack", "--locked", "--python", "3.14", cwd=self.checkout - ) + run(self.uv, "sync", "--project", "stack", "--locked", "--python", "3.14", + cwd=self.checkout) def imports(self) -> None: run(self.python, "-c", CLIENT_IMPORTS, cwd=self.checkout, env=self.environment()) @@ -140,39 +125,24 @@ def prepare_native(self) -> None: 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), - "PYTHONPATH": os.pathsep.join(map(str, (*paths, self.checkout / "src"))), - } + return {**os.environ, "HOME": str(self.home), + "PYTHONPATH": os.pathsep.join(map(str, (*paths, self.checkout / "src")))} def cli(self, root: Path, *arguments: str) -> dict: """One JSON CLI result for this bus alone, never the route's, waiting while the route index is still being built.""" for _ in range(240): completed = subprocess.run( - [ - str(self.python), - "-c", - "from agent_comms.cli import main; main()", - "--root", - str(root), - *arguments, - ], - cwd=self.checkout, - env=self.environment(), - capture_output=True, - text=True, - ) + [str(self.python), "-c", "from agent_comms.cli import main; main()", "--root", str(root), + *arguments], cwd=self.checkout, env=self.environment(), capture_output=True, + text=True) if "Route index covers" not in completed.stdout: break time.sleep(5) result = json.loads(completed.stdout or "{}") if completed.returncode or "error" in result: - raise UpgradeRefused( - f"agent-comms {' '.join(arguments)}: " - f"{completed.stdout.strip() or completed.stderr.strip()}" - ) + raise UpgradeRefused(f"agent-comms {' '.join(arguments)}: " + f"{completed.stdout.strip() or completed.stderr.strip()}") return result def owners(self, root: Path, command: str, names: list[str]) -> None: @@ -184,11 +154,8 @@ def verify(self, root: Path) -> dict[str, int]: threads = self.cli(root, "threads")["threads"] present = self.cli(root, "who") history = self.cli(root, "history", "--limit", "50") # #any: every channel - return { - "threads": len(threads), - "who": len(present["who"]), - "history": len(history[EverythingHistoryPageResult.messages_key]), - } + return {"threads": len(threads), "who": len(present["who"]), + "history": len(history[EverythingHistoryPageResult.messages_key])} @dataclass @@ -216,9 +183,7 @@ def tree(self, revision: str) -> Path: def close(self) -> None: for path in self.trees.values(): - subprocess.run( - ["git", "-C", str(self.repo), "worktree", "remove", "--force", str(path)] - ) + subprocess.run(["git", "-C", str(self.repo), "worktree", "remove", "--force", str(path)]) def carry(step: Step, root: Path, stack: Stack, revisions: Revisions, *flags: str) -> Carried: @@ -228,8 +193,7 @@ def carry(step: Step, root: Path, stack: Stack, revisions: Revisions, *flags: st if "--before-stop" in flags: return TodoColumnCarry.before_stop(root, apply=True) return TodoColumnCarry.carry_step( - root, step, CarryChain.at(revisions.repo, step.revision), apply=True - ) + root, step, CarryChain.at(revisions.repo, step.revision), apply=True) tree = revisions.tree(step.revision) environment = stack.environment(tree / "src", tree / "tools") worker = ( @@ -238,17 +202,8 @@ def carry(step: Step, root: Path, stack: Stack, revisions: Revisions, *flags: st "print(json.dumps(asdict(getattr(step,sys.argv[3])(Path(sys.argv[4]),apply=True))))" ) operation = "before_stop" if "--before-stop" in flags else "carry" - output = run( - stack.python, - "-c", - worker, - f"cutover.{step.name}", - step.declaration, - operation, - root, - cwd=tree, - env=environment, - ) + output = run(stack.python, "-c", worker, f"cutover.{step.name}", step.declaration, + operation, root, cwd=tree, env=environment) return FieldCodec.decode(Carried, FieldCodec.parse(output.strip().splitlines()[-1])) @@ -257,10 +212,8 @@ def before_stop( ) -> list[dict]: """What the running installation needs before its own stop command reads the bus.""" with _store_lock(root / "wire"): - return [ - carry(step, root, stack, revisions, "--before-stop", "--apply").report(step.name, True) - for step in pending - ] + return [carry(step, root, stack, revisions, "--before-stop", "--apply").report( + step.name, True) for step in pending] def carry_all( @@ -270,10 +223,8 @@ def carry_all( for step in pending: with _store_lock(root / "wire"): result = CarryLedger(root).apply( - CarryChain.at(revisions.repo, step.revision), - step.name, - lambda: carry(step, root, stack, revisions, "--apply"), - ) + CarryChain.at(revisions.repo, step.revision), step.name, + lambda: carry(step, root, stack, revisions, "--apply")) results.append(result.report(step.name, True)) return results @@ -283,39 +234,15 @@ def copy_bus(root: Path, copy: Path) -> None: # The maintenance files describe the live root; a copy is a new root and starts open. held = set(MaintenanceBarrier(root / "registry.json").witness_paths) - stores: list[Path] = [] - def skipped(directory: str, names: list[str]) -> set[str]: - # SQLite stores are copied whole by SQLite below; a file copy races their - # writers, who may checkpoint and remove a companion mid-copy. - found = [ - Path(directory, name) - for name in names - if Path(directory, name).is_file() - and not Path(directory, name).is_symlink() - and CoordinationStore.is_database(Path(directory, name)) - ] - stores.extend(found) - store_files = {file.name for store in found for file in CoordinationStore.files(store)} - return { - name - for name in names - if name in store_files - or Path(directory, name) in held - or stat.S_ISSOCK(os.lstat(os.path.join(directory, name)).st_mode) - } + return {name for name in names + if Path(directory, name) in held + or stat.S_ISSOCK(os.lstat(os.path.join(directory, name)).st_mode)} from agent_comms.store_files import _store_lock with _store_lock(root / "wire"): shutil.copytree(root, copy, symlinks=True, ignore=skipped) - for store in stores: - with ( - closing(CoordinationStore.read_snapshot(store)) as source, - closing(sqlite3.connect(copy / store.relative_to(root))) as target, - ): - source.backup(target) - shutil.copymode(store, copy / store.relative_to(root)) def rehearsal_home(scratch: Path, root: Path, copy: Path) -> Path: @@ -342,11 +269,8 @@ def rehearse(root: Path, pending: tuple[Step, ...], stack: Stack, revisions: Rev stack = replace(stack, home=rehearsal_home(Path(scratch), root, copy)) 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 - ) + os.environ[HOLDER_NONCE] = MaintenanceBarrier(copy / "registry.json").begin( + MaintenanceOperator("upgrade-bus")).nonce with transcripts_held(copy): before = bus_files(copy) stopping = before_stop(copy, pending, stack, revisions) @@ -355,15 +279,8 @@ def rehearse(root: Path, pending: tuple[Step, ...], stack: Stack, revisions: Rev return carry_and_read(copy, before, pending, stack, stack, revisions, stopping) -def carry_and_read( - root: Path, - before: dict[str, str], - pending: tuple[Step, ...], - staged: Stack, - reader: Stack, - revisions: Revisions, - stopping: list[dict], -) -> dict: +def carry_and_read(root: Path, before: dict[str, str], pending: tuple[Step, ...], staged: Stack, + reader: Stack, revisions: Revisions, stopping: list[dict]) -> dict: """The pass a rehearsal and a live install share once ``before`` was read under the transcript hold. The hold is released first: SQLite cannot lend its writer lock to a step's process, so a step that writes the transcript store could not run under it.""" @@ -375,9 +292,7 @@ def upgrade(checkout: Path, root: Path, target: str, installer: str, rehearse_on current = run("git", "-C", checkout, "rev-parse", "HEAD").strip() run("git", "-C", checkout, "fetch", "-q", "origin") target = run("git", "-C", checkout, "rev-parse", "--verify", f"{target}^{{commit}}").strip() - if subprocess.run( - ["git", "-C", str(checkout), "merge-base", "--is-ancestor", current, target] - ).returncode: + if subprocess.run(["git", "-C", str(checkout), "merge-base", "--is-ancestor", current, target]).returncode: raise UpgradeRefused(f"{target[:12]} does not follow the checkout's {current[:12]}") if run("git", "-C", checkout, "status", "--porcelain", "--untracked-files=no").strip(): raise UpgradeRefused("The checkout has local changes") @@ -395,15 +310,10 @@ def upgrade(checkout: Path, root: Path, target: str, installer: str, rehearse_on try: staged = Stack(revisions.tree(target)) staged.relock() - install = Install.of( - checkout, root, current, target, pending, installed, staged, revisions, installer - ) - report = { - "from": current, - "to": target, - "pending": [step.name for step in pending], - "rehearsal": install.rehearsal(), - } + install = Install.of(checkout, root, current, target, pending, installed, staged, revisions, + installer) + report = {"from": current, "to": target, "pending": [step.name for step in pending], + "rehearsal": install.rehearsal()} if rehearse_only: return report report["live"] = install.live() @@ -427,23 +337,11 @@ class Install(ABC): installer: str @classmethod - def of( - cls, - checkout: Path, - root: Path, - current: str, - target: str, - pending: tuple[Step, ...], - installed: Stack, - staged: Stack, - revisions: Revisions, - installer: str, - ) -> Install: + def of(cls, checkout: Path, root: Path, current: str, target: str, pending: tuple[Step, ...], + installed: Stack, staged: Stack, revisions: Revisions, installer: str) -> Install: unchanged = current == target and staged.lock.read_bytes() == installed.lock.read_bytes() member = (ChangedInstall, UnchangedInstall)[unchanged] - return member( - checkout, root, current, target, pending, installed, staged, revisions, installer - ) + return member(checkout, root, current, target, pending, installed, staged, revisions, installer) @abstractmethod def rehearsal(self) -> dict: ... @@ -457,17 +355,8 @@ def rehearsal(self) -> dict: return rehearse(self.root, self.pending, self.staged, self.revisions) def live(self) -> dict: - return live( - self.checkout, - self.root, - self.current, - self.target, - self.pending, - self.installed, - self.staged, - self.revisions, - self.installer, - ) + return live(self.checkout, self.root, self.current, self.target, self.pending, + self.installed, self.staged, self.revisions, self.installer) class UnchangedInstall(Install): @@ -482,24 +371,12 @@ def live(self) -> dict: return {"unchanged": self.target} -def live( - checkout: Path, - root: Path, - current: str, - target: str, - pending: tuple[Step, ...], - installed: Stack, - staged: Stack, - revisions: Revisions, - installer: str, -) -> dict: +def live(checkout: Path, root: Path, current: str, target: str, pending: tuple[Step, ...], + 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" - ] + 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) maintenance = MaintenanceBarrier(RegistryStore.at(root).path) held = maintenance.begin(MaintenanceOperator("upgrade-bus")) @@ -508,20 +385,13 @@ def live( try: installed.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')}", - ) + backup = Backup.take(root, ROUTE, (root,), + root.parent / f"{root.name}.upgrade-{time.strftime('%Y%m%dT%H%M%S')}") except BaseException as error: maintenance.clear(held) installed.owners(root, "start", running) - print( - f"upgrade-bus: aborted before the backup ({type(error).__name__}: {error}); " - f"maintenance cleared and {running} restarted", - file=sys.stderr, - ) + print(f"upgrade-bus: aborted before the backup ({type(error).__name__}: {error}); " + f"maintenance cleared and {running} restarted", file=sys.stderr) raise shutil.copy2(lock, backup.path / "uv.lock") if lock.exists() else None try: @@ -536,19 +406,13 @@ def live( except BaseException as error: state = maintenance.read() state.phase.hold_for(maintenance, state, MaintenanceOperator("upgrade-bus")) - 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, - ) + 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) raise installed.owners(root, "start", running) - return { - "stopped_and_started": running, - **passed, - "services": services, - "backup": str(backup.path), - } + return {"stopped_and_started": running, **passed, "services": services, + "backup": str(backup.path)} def stopped(signum: int, frame: object) -> None: @@ -559,33 +423,19 @@ def stopped(signum: int, frame: object) -> None: def main() -> None: - parser = argparse.ArgumentParser( - description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter - ) - parser.add_argument( - "--checkout", type=lambda value: Path(value).expanduser().resolve(), default=REPO - ) - parser.add_argument( - "--root", - type=lambda value: Path(value).expanduser().resolve(), - default=Path.home() / ".agent-comms", - ) + parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument("--checkout", type=lambda value: Path(value).expanduser().resolve(), default=REPO) + parser.add_argument("--root", type=lambda value: Path(value).expanduser().resolve(), + default=Path.home() / ".agent-comms") parser.add_argument("--to", default="origin/main") - parser.add_argument( - "--installer", required=True, 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" - ) + parser.add_argument("--installer", required=True, + 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() for signum in (signal.SIGTERM, signal.SIGINT): signal.signal(signum, stopped) try: - print( - json.dumps( - upgrade(args.checkout, args.root, args.to, args.installer, args.rehearse), indent=2 - ) - ) + print(json.dumps(upgrade(args.checkout, args.root, args.to, args.installer, args.rehearse), indent=2)) except UpgradeRefused as error: raise SystemExit(f"upgrade-bus: {error}") From 829821a02d078d4e47a68eaef95832ca77bc2aeb Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 19:17:51 -0400 Subject: [PATCH 5/5] Budget the merged order's phase property Src-Budget: 5