diff --git a/benchmark/edgebench/cli_source_activation.py b/benchmark/edgebench/cli_source_activation.py new file mode 100644 index 0000000000..9c39d15aad --- /dev/null +++ b/benchmark/edgebench/cli_source_activation.py @@ -0,0 +1,306 @@ +"""Controller-only CLI activation at an independently observed idle boundary. + +The running Host retains its imported code. Only its owned CLI symlink changes; +work, native sessions, authority, model configuration and phase files stay put. +No controller mailbox or checkpoint capture identity is consumed by activation. +""" +from __future__ import annotations + +from collections.abc import Callable +import hashlib +import io +import json +from pathlib import Path, PurePosixPath +import re +import math +import os +import stat +import tarfile +import time + +from loopx.control_plane.effect_runtime import effect_runtime_result +from loopx.file_lock import exclusive_file_lock, write_json_atomic +from .checkpoint_worker import _worker_lock +from .cli_source_probe import _archive_response, source_goal_control + + +class SourceActivationHeld(RuntimeError): + """Reconcile this exact intent without replaying activation or dispatch.""" + + +def _path(value: str) -> str: + path = PurePosixPath(value) + if not path.is_absolute() or str(path) != value or ".." in path.parts: + raise ValueError("Activation paths must be normalized and absolute") + return value + + +def _link_record(container, path: str) -> dict: + """Read the symlink itself through the daemon, including while paused.""" + from docker.errors import NotFound + + blocks, length = [], 0 + try: + with _archive_response(container, path) as (chunks, _): + for block in chunks: + length += len(block) + if length > 128 * 1024: + raise ValueError("Owned CLI launcher archive exceeds the symlink bound") + blocks.append(block) + except NotFound: + return {} + with io.BytesIO(b"".join(blocks)) as data, tarfile.open(fileobj=data) as archive: + members = archive.getmembers() + if len(members) != 1 or not members[0].issym(): + return {} + member = members[0] + return {"launcher_target": _path(member.linkname), "launcher_uid": member.uid, + "launcher_gid": member.gid, "launcher_mode": member.mode & 0o777} + + +def _link(container, path: str) -> str | None: + return _link_record(container, path).get("launcher_target") + + +def _put_link(container, path: str, target: str, *, uid: int = 0, gid: int = 0) -> None: + """Dispatch-atomic only under our physical pause, not an atomic rename. + + A partial stream or lost acknowledgement is resolved by _link before release. + No code is executed inside the paused container. + """ + payload = io.BytesIO() + with tarfile.open(fileobj=payload, mode="w") as archive: + entry = tarfile.TarInfo(PurePosixPath(path).name) + entry.type, entry.linkname, entry.mode = tarfile.SYMTYPE, target, 0o777 + entry.uid, entry.gid = uid, gid + archive.addfile(entry) + payload.seek(0) + if not container.put_archive(str(PurePosixPath(path).parent), payload): + raise RuntimeError("CLI launcher transport did not confirm the write") + + +def _validate_plan(plan: dict) -> None: + fields = {"activation_id", "container_id", "started_at", "goal_id", "agent_id", "todo_id", "turn_instance_id", "session_id", + "registry_path", "launcher", "launcher_uid", "launcher_gid", "old_target", "new_target", "release_root", "target_revision", + "target_archive_sha256", "worker_lock"} + if not isinstance(plan, dict) or set(plan) != fields: + raise ValueError("CLI activation needs its exact source/release declaration") + for field, length in (("activation_id", 32), ("container_id", 64), + ("target_revision", 40), ("target_archive_sha256", 64)): + if not isinstance(plan[field], str) or not re.fullmatch(f"[0-9a-f]{{{length}}}", plan[field]): + raise ValueError("Invalid CLI activation identity or digest") + for field in ("launcher_uid", "launcher_gid"): + if type(plan[field]) is not int or plan[field] < 0: + raise ValueError("CLI activation requires the original launcher ownership") + for field in fields - {"activation_id", "container_id", "target_revision", "target_archive_sha256", "launcher_uid", "launcher_gid"}: + if not isinstance(plan[field], str) or not plan[field]: + raise ValueError("CLI activation source fields must be explicit") + for field in ("registry_path", "launcher", "old_target", "new_target", "release_root", "worker_lock"): + _path(plan[field]) + root = PurePosixPath(plan["release_root"]) + for field in ("old_target", "new_target"): + relative = PurePosixPath(plan[field]).relative_to(root) + if len(relative.parts) != 3 or relative.parts[1:] != ("scripts", "loopx"): + raise ValueError("CLI activation needs an owned versioned launcher") + if (plan["old_target"] == plan["new_target"] or + root in PurePosixPath(plan["launcher"]).parents): + raise ValueError("CLI activation must select a distinct owned versioned release") + + +def _state(container, plan): + container.reload() + state = container.attrs["State"] + if container.id != plan["container_id"] or state["StartedAt"] != plan["started_at"]: + raise SourceActivationHeld("Original source namespace identity changed") + return state + + +def _held_state(container, plan): + state = _state(container, plan) + if not state["Running"] or not state["Paused"]: + raise SourceActivationHeld("Original writer barrier changed; no launcher effect is authorized") + return state + + +def _goal_control(container, plan): + try: + return source_goal_control(container, plan, plan["registry_path"]) + except Exception: + return "unknown" # An unavailable owner cannot authorize a write/release. + + +def _historical(value): + return {**value, "historical_receipt": True, "replayed": True, + "current_source_verified": False, "model_receipt_adoption": "unobserved"} + + +def activate_cli_source(client, *, plan: dict, receipt: Path, + preflight: Callable, probe: Callable, witness: Callable, verify_release: Callable, + control: Callable[[], str], execute: bool = False, + reconcile: bool = False, repair: str | None = None, + timeout: float = 120) -> dict: + """Change one launcher; callbacks are source/native owners, never model calls. + + ``probe`` observes a paused source using an isolated reader. ``witness`` + observes exactly that probe's declared source participants. ``verify_release`` + verifies the staged immutable bytes against the admitted archive. A controller + supplies its current ordinary Stop observation before each effect/release. + None of these observations grants Goal, account or solver admission authority. + """ + _validate_plan(plan) + if (type(execute) is not bool or type(reconcile) is not bool or + not math.isfinite(timeout) or timeout <= 0 or + repair not in (None, "old", "new") or (repair is not None and not reconcile)): + raise ValueError("CLI activation requires explicit mode and finite timeout") + if not execute: + return {"planned": True, "source_change_allowed": False, "execution_authority_granted": False, + "plan_sha256": hashlib.sha256(json.dumps(plan, sort_keys=True).encode()).hexdigest()} + if receipt.parent.resolve() != receipt.parent or receipt.is_symlink(): + raise ValueError("Activation receipt must be controller-owned and physical") + parent = receipt.parent.stat() + if parent.st_uid != os.getuid() or parent.st_mode & 0o077: + raise ValueError("Activation receipt requires its private controller directory") + box = client.containers.get(plan["container_id"]) + lock = _worker_lock(client, box.id) + if str(lock) != plan["worker_lock"]: + raise ValueError("Activation must share the original Host's physical worker-lock realm") + with exclusive_file_lock(lock, timeout_seconds=timeout, operation="cli_source_activation"): + if reconcile: + info = receipt.lstat() + if (not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or + info.st_nlink != 1 or info.st_mode & 0o077): + raise ValueError("Activation recovery requires its private physical receipt") + value = json.loads(receipt.read_bytes()) + if value.get("schema") != "loopx_cli_source_activation_v0" or value.get("plan") != plan: + raise ValueError("Activation recovery differs from its original intent") + if value.get("phase") == "released": + return _historical(value) # Never pause or replay a released source. + # A durable request is not a pause acknowledgement. In particular, + # another actor can pause after this controller dies before its + # pause call; observing Paused cannot transfer that actor's barrier. + if (value.get("phase") not in ("paused", "qualified", "deferred", "switch_requested", + "repair_requested", "release_requested") or value.get("pause_acknowledged") is not True or + type(value.get("paused_at_epoch")) not in (int, float) or + not math.isfinite(value["paused_at_epoch"]) or value["paused_at_epoch"] <= 0): + raise SourceActivationHeld("Original pause acknowledgement is not durable; ownership recovery is required") + else: + if receipt.exists(): + raise ValueError("CLI activation intent already exists; reconcile the original") + state = _state(box, plan) + if not state["Running"] or state["Paused"]: + raise ValueError("CLI activation needs its exact running, unpaused source") + expected_link = {"launcher_target": plan["old_target"], "launcher_uid": plan["launcher_uid"], + "launcher_gid": plan["launcher_gid"], "launcher_mode": 0o777} + if control() != "continue" or _link_record(box, plan["launcher"]) != expected_link: + return {"applied": False, "reason": "control_or_launcher_changed", "execution_authority_granted": False} + initial = preflight(box, plan) + initial.update(container_id=box.id, started_at=state["StartedAt"], running=True, + paused=False, control=control(), source_control=_goal_control(box, plan)) + initial_decision = effect_runtime_result("benchmark.cli_source_activation", { + "operation": "preflight", "scope": plan, "observation": initial}) + if not initial_decision.get("pause_allowed", False): + return {"applied": False, "reason": initial_decision["reason"], + "execution_authority_granted": False} + verify_release(box, plan) + value = {"schema": "loopx_cli_source_activation_v0", "plan": plan, + "phase": "pause_requested", "prepared_at_epoch": time.time(), + "execution_authority_granted": False, "work_or_clock_mutations": 0} + write_json_atomic(receipt, value, mode=0o600, fsync_directory=True) + try: + box.pause() + except Exception: + # A rejected/uncertain pause never authorizes release of somebody + # else's barrier. Preserve the intent for explicit observation. + value["phase"] = "pause_uncertain" + write_json_atomic(receipt, value, mode=0o600, fsync_directory=True) + raise SourceActivationHeld("Pause acknowledgement unavailable; inspect original source") from None + value.update(phase="paused", pause_acknowledged=True, paused_at_epoch=time.time()) + write_json_atomic(receipt, value, mode=0o600, fsync_directory=True) + state = _state(box, plan) + if value["phase"] == "release_requested" and state["Running"] and not state["Paused"]: + target = plan["new_target"] if value["released_target"] == "new" else plan["old_target"] + if _link_record(box, plan["launcher"]) != {"launcher_target": target, "launcher_uid": plan["launcher_uid"], + "launcher_gid": plan["launcher_gid"], "launcher_mode": 0o777}: + raise SourceActivationHeld("Released source launcher changed; no effect was replayed") + value.update(phase="released", release_observed_at_epoch=time.time(), + applied=value["released_target"] == "new", model_receipt_adoption="unobserved") + write_json_atomic(receipt, value, mode=0o600, fsync_directory=True) + return _historical(value) + if not state["Running"] or not state["Paused"] or value["phase"] == "pause_uncertain": + raise SourceActivationHeld("Original owned pause is not confirmed") + if "qualification" not in value: + facts = probe(box, plan) + state = _held_state(box, plan) + facts.update(container_id=box.id, started_at=state["StartedAt"], running=True, + paused=True, owned_barrier=True, control=control(), + source_control=_goal_control(box, plan), + **_link_record(box, plan["launcher"])) + decision = effect_runtime_result("benchmark.cli_source_activation", { + "operation": "qualify", "scope": plan, "observation": facts}) + value.update(phase="qualified" if decision["qualified"] else "deferred", qualification=decision, + source_witness=facts["source_witness_after"], + source_stopped=facts.get("source_stopped", False)) + write_json_atomic(receipt, value, mode=0o600, fsync_directory=True) + if decision["qualified"]: + verify_release(box, plan) + if control() != "continue" or _goal_control(box, plan) != "continue": + raise SourceActivationHeld("Current control stopped before the launcher write") + _held_state(box, plan) + value["phase"] = "switch_requested" + write_json_atomic(receipt, value, mode=0o600, fsync_directory=True) + try: + _put_link(box, plan["launcher"], plan["new_target"], uid=plan["launcher_uid"], gid=plan["launcher_gid"]) + except Exception: + value["write_acknowledgement_lost"] = True + write_json_atomic(receipt, value, mode=0o600, fsync_directory=True) + if repair is not None: + state = _held_state(box, plan) + repair_decision = effect_runtime_result("benchmark.cli_source_activation", { + "operation": "repair", "scope": plan, "observation": { + "container_id": box.id, "started_at": state["StartedAt"], + "running": True, "paused": True, "owned_barrier": True, "control": control(), + "source_control": _goal_control(box, plan), + "source_stopped": value.get("source_stopped", False), "repair_target": repair, + "qualified_source": value.get("qualification", {}).get("qualified") is True, + "unchanged_source_witness": witness(box, plan) == value["source_witness"]}}) + if repair_decision.get("source_change_allowed") is not True: + raise SourceActivationHeld("Explicit launcher repair lacks its unchanged qualified source") + verify_release(box, plan) + value.update(phase="repair_requested", repair_target=repair) + write_json_atomic(receipt, value, mode=0o600, fsync_directory=True) + if control() != "continue" or _goal_control(box, plan) != "continue": + raise SourceActivationHeld("Current control stopped before the launcher repair") + _held_state(box, plan) + _put_link(box, plan["launcher"], plan["old_target"] if repair == "old" else plan["new_target"], + uid=plan["launcher_uid"], gid=plan["launcher_gid"]) + if control() != "continue": + raise SourceActivationHeld("Current control does not permit source release") + state = _held_state(box, plan) + decision = effect_runtime_result("benchmark.cli_source_activation", { + "operation": "readback", "scope": plan, "observation": { + "container_id": box.id, "started_at": state["StartedAt"], + "running": True, "paused": True, "owned_barrier": True, "control": control(), + "source_control": _goal_control(box, plan), + "source_stopped": value.get("source_stopped", False), + "qualified_source": value.get("qualification", {}).get("qualified") is True, + **_link_record(box, plan["launcher"]), + "unchanged_source_witness": witness(box, plan) == value["source_witness"]}}) + value["readback"] = decision + write_json_atomic(receipt, value, mode=0o600, fsync_directory=True) + if not decision["resume_allowed"]: + raise SourceActivationHeld("Launcher/source readback is uncertain; original source remains paused") + value.update(phase="release_requested", released_target=decision["disposition"]) + write_json_atomic(receipt, value, mode=0o600, fsync_directory=True) + _held_state(box, plan) + if control() != "continue" or _goal_control(box, plan) != "continue": + raise SourceActivationHeld("Current control stopped before releasing the original source") + box.unpause() + state = _state(box, plan) + if state["Paused"] or not state["Running"]: + raise SourceActivationHeld("Source release is not confirmed") + value.update(phase="released", release_observed_at_epoch=time.time(), + applied=decision["disposition"] == "new", + pause_wall_seconds=time.time() - value["paused_at_epoch"], + model_receipt_adoption="unobserved") + write_json_atomic(receipt, value, mode=0o600, fsync_directory=True) + return value diff --git a/benchmark/edgebench/cli_source_probe.py b/benchmark/edgebench/cli_source_probe.py new file mode 100644 index 0000000000..89b538dfe4 --- /dev/null +++ b/benchmark/edgebench/cli_source_probe.py @@ -0,0 +1,315 @@ +"""Read a paused source in an isolated, credential-free observer namespace. + +The source itself executes no probe while paused. Existing native and settlement +owners read their copied paths; a passive native reader is not a model request. +This adapter never creates a checkpoint point, registration, or work allowance. +""" +from __future__ import annotations + +from contextlib import contextmanager +import hashlib +import io +import json +from pathlib import Path, PurePosixPath +import posixpath +import re +import tarfile +import tempfile + +from .checkpoint_worker import _guest_path + + +def _canonical(value): + return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest() + + +@contextmanager +def _archive_response(container, path, *, metadata_only=False): + """Own the actual HTTP response, including before the first body read. + + Docker's GET holds its container archive lock until EOF or response close. + Closing an unstarted SDK generator does neither. HEAD is the native stat + route; GET here retains the caller's socket timeout and always closes. + """ + from docker.utils import decode_json_header + + api = container.client.api + url = api._url("/containers/{0}/archive", container.id) + response = (api.head(url, params={"path": path}, timeout=api.timeout, allow_redirects=False) + if metadata_only else api._get(url, params={"path": path}, stream=True, + headers={"Accept-Encoding": "identity"}, allow_redirects=False)) + try: + api._raise_for_status(response) + stat = decode_json_header(response.headers["x-docker-container-path-stat"]) + yield (() if metadata_only else response.iter_content(2 * 1024 * 1024), stat) + finally: + response.close() + + +def _path_stat(container, path): + with _archive_response(container, path, metadata_only=True) as (_, stat): + return stat + + +def archive_witness(container, components: list[dict]) -> str: + """Hash declared bytes and link identities without following task symlinks.""" + result = [] + for component in components: + remote = _guest_path(component["source_path"]) + members = [] + with _archive_response(container, str(remote)) as (chunks, stat), tempfile.TemporaryFile() as spool: + for block in chunks: + spool.write(block) + spool.seek(0) + with tarfile.open(fileobj=spool) as archive: + names = set() + for entry in archive: + name = PurePosixPath(entry.name) + if (name.is_absolute() or ".." in name.parts or not name.parts or + name.parts[0] != remote.name or str(name) in names or + not (entry.isfile() or entry.isdir() or entry.issym() or entry.islnk())): + raise ValueError("Source observation archive has invalid members") + names.add(str(name)) + item = {"path": str(name), "mode": entry.mode, "kind": entry.type.decode("ascii")} + if entry.isfile(): + item.update(bytes=entry.size, sha256=hashlib.file_digest(archive.extractfile(entry), "sha256").hexdigest()) + elif entry.issym() or entry.islnk(): + item["target"] = entry.linkname # Data only, never resolve or extract it on the controller. + members.append(item) + result.append({"id": component["id"], "source_path": str(remote), + "root_link": stat.get("linkTarget", ""), "members": sorted(members, key=lambda m: m["path"])}) + return _canonical(result) + + +def _copy(container, observer, remote: str, immutable_roots=()) -> None: + path = _guest_path(remote) + if observer.exec_run(["mkdir", "-p", str(path.parent)], user="root").exit_code: + raise RuntimeError("Observer parent preparation failed") + with _archive_response(container, str(path)) as (chunks, _), tempfile.TemporaryFile() as spool: + for block in chunks: + spool.write(block) + spool.seek(0) + with tarfile.open(fileobj=spool) as archive: + names = set() + for entry in archive: + name = PurePosixPath(entry.name) + if (name.is_absolute() or ".." in name.parts or not name.parts or + name.parts[0] != path.name or str(name) in names or + not (entry.isfile() or entry.isdir() or entry.issym() or entry.islnk())): + raise ValueError("Observer archive contains an invalid member") + names.add(str(name)) + if entry.issym() or entry.islnk(): + absolute = posixpath.normpath(posixpath.join(str(path.parent / name.parent), entry.linkname)) + if (not immutable_roots or not any(absolute == root or absolute.startswith(root + "/") + for root in immutable_roots)): + raise ValueError("Observer link leaves its declared immutable source roots") + spool.seek(0) + if not observer.put_archive(str(path.parent), spool): + raise RuntimeError("Observer source transport was not confirmed") + + +def _file_bytes(container, path): + with _archive_response(container, path) as (chunks, _), tempfile.TemporaryFile() as spool: + for block in chunks: + spool.write(block) + spool.seek(0) + with tarfile.open(fileobj=spool) as archive: + entries = archive.getmembers() + if len(entries) != 1 or not entries[0].isfile() or entries[0].size > 2 * 1024 * 1024: + raise ValueError("Source observation requires one bounded physical JSON file") + return archive.extractfile(entries[0]).read() + + +def _json_file(container, path): + return json.loads(_file_bytes(container, path)) + + +def source_goal_control(container, plan, registry_path): + """Existing current Goal activation, never a cached profile's on flag. + + The operation caller must also compose its ordinary resource/phase/owner + Stop observation. This read neither grants work nor overrides that control. + """ + from loopx.control_plane.projects.registry_codec import decode_registry_snapshot + from loopx.control_plane.goals.activation import goal_is_stopped + + document = decode_registry_snapshot(Path(registry_path), _file_bytes(container, registry_path)) + matches = [g for g in document.get("goals", []) if g.get("id") == plan["goal_id"]] + if len(matches) != 1: + return "unknown" + return "stop" if goal_is_stopped(matches[0]) else "continue" + + +def process_observation(container, allowed_command_sha256: list[str]) -> list[int]: + """Only exact controller-declared idle command identities are accepted. + + Do not infer model, tool, or idle roles from process-name substrings. The + native owner separately verifies a completed, stable source conversation. + """ + table = container.top(ps_args="-eo pid,args") + titles = table["Titles"] + pid_index = titles.index("PID") + commands = [i for i, value in enumerate(titles) if value in {"COMMAND", "CMD", "ARGS"}] + if len(commands) != 1: + raise ValueError("Source process observation is unavailable") + unknown = [] + for row in table["Processes"]: + digest = hashlib.sha256(row[commands[0]].encode()).hexdigest() + if digest not in allowed_command_sha256: + unknown.append(int(row[pid_index])) + return sorted(unknown) + + +class PausedSourceProbe: + """Concrete observer for the existing heartbeat CLI source boundary. + + All paths and idle command identities come from the original owned runtime. + Immutable roots supply its Python, Node, native executable and installed + kernel; they must not include a credential home. Native store participants + are explicit, matching the existing checkpoint reader's closed allowlist. + """ + + def __init__(self, client, *, declaration: dict, components: list[dict], + immutable_roots: list[str], allowed_idle_commands: list[str], + user: str, observer_memory: str = "512m"): + self.client, self.declaration = client, declaration + self.components, self.immutable_roots = components, immutable_roots + self.allowed = allowed_idle_commands + self.user, self.observer_memory = user, observer_memory + for row in components: + _guest_path(row["source_path"]) + for path in immutable_roots: + _guest_path(path) + home = PurePosixPath(declaration["native_home"]) + # Copy only the native owner's declared stores/config/hooks, never auth + # or the entire credential home, even if the caller supplied that root. + paths = [PurePosixPath(r["source_path"]) for r in components] + list(map(PurePosixPath, immutable_roots)) + if any(p == home or p in home.parents for p in paths): + raise ValueError("Observer must not copy the native credential home") + for path in paths: + if home not in path.parents: + continue + name = path.relative_to(home).parts[0] + if name not in {"sessions", "archived_sessions", "session_index.jsonl", "config.toml", "hooks.json", "loopx-context"} and not re.fullmatch( + r"(?:state|thread_history)_[0-9]+\.sqlite(?:-wal|-journal)?", name): + raise ValueError("Observer native paths must use the checkpoint reader's store allowlist") + required = [declaration[key] for key in ("registry", "runtime_root", "scheduler_state", "workspace", "profile_path", "phase_path")] + required += [declaration["scheduler_state"] + ".host-attempt", str(home / "config.toml"), str(home / "hooks.json")] + required += declaration["authority_paths"] + for required_path in required: + if not any(required_path == row["source_path"] or required_path.startswith(row["source_path"] + "/") for row in components): + raise ValueError("Source witness omits a required work, authority, profile or phase participant") + self.native_paths = declaration["native_store_paths"] + if (not isinstance(self.native_paths, list) or not self.native_paths or + not any(re.fullmatch(r"state_[0-9]+\.sqlite", PurePosixPath(p).name) for p in self.native_paths)): + raise ValueError("Observer needs the qualified native checkpoint store inventory") + for path in self.native_paths: + if PurePosixPath(path).parent != home or not any(row["source_path"] == path for row in components): + raise ValueError("Native store inventory differs from the declared physical participants") + + def _components(self, container): + from docker.errors import NotFound + + result = [*self.components, *({"id": "immutable_" + str(i), "source_path": path} + for i, path in enumerate(self.immutable_roots))] + for row in self.components: + if not row["source_path"].endswith(".sqlite"): + continue + for suffix in ("-wal", "-journal"): + path = row["source_path"] + suffix + try: + _path_stat(container, path) + except NotFound: + continue + result.append({"id": row["id"] + suffix, "source_path": path}) + return result + + def witness(self, container, _plan): + if self.declaration["registry"] != _plan["registry_path"]: + raise ValueError("Source observation must use the exact owned registry route") + return archive_witness(container, self._components(container)) + + def preflight(self, container, _plan): + from docker.errors import NotFound + + if self.declaration["registry"] != _plan["registry_path"]: + raise ValueError("Source preflight differs from its owned registry route") + try: + _path_stat(container, self.declaration["ready_path"]) + except NotFound: + ready = False + else: + ready = True + state = _json_file(container, self.declaration["scheduler_state"]) + pending = _json_file(container, self.declaration["scheduler_state"] + ".host-attempt") + return {"source_control": source_goal_control(container, _plan, self.declaration["registry"]), + "ready_present": ready, "unknown_process_ids": process_observation(container, self.allowed), + "scheduler": {"host_attempt": state.get("host_attempt"), "host_pending": pending["pending"], + "last_wake_status": state.get("last_wake_status"), "last_wake_rc": state.get("last_wake_rc")}} + + def __call__(self, container, plan): + container.reload() + if not container.attrs["State"]["Paused"]: + raise ValueError("Source observation requires the physical writer barrier") + for row in self._components(container): + path = row["source_path"] + if any(target == path or target.startswith(path + "/") for target in (plan["launcher"], plan["new_target"])): + raise ValueError("Protected source participants must exclude the owned launcher and new release") + before = self.witness(container, plan) + from docker.errors import NotFound + + ready = self.declaration["ready_path"] + try: + _path_stat(container, ready) + except NotFound: + ready_present = False + else: + ready_present = True + unknown = process_observation(container, self.allowed) + facts = {"probe_status": "unavailable", "ready_present": ready_present, + "unknown_process_ids": unknown, "source_witness_before": before} + if ready_present or unknown: + facts["source_witness_after"] = self.witness(container, plan) + return facts + observer = self.client.containers.run(container.attrs["Image"], ["sleep", "180"], + entrypoint=[], detach=True, network_disabled=True, mem_limit=self.observer_memory, + pids_limit=128, nano_cpus=1_000_000_000, + labels={"loopx.source-activation-observer": plan["activation_id"]}) + try: + for path in self.immutable_roots: + _copy(container, observer, path, self.immutable_roots) + reads = [self.declaration[key] for key in ("registry", "runtime_root", "scheduler_state")] + reads += self.declaration["authority_paths"] + reads += [self.declaration["scheduler_state"] + ".host-attempt", + str(PurePosixPath(self.declaration["native_home"]) / "config.toml")] + reads += [row["source_path"] for row in self._components(container) + if row["source_path"] in self.native_paths or any(row["source_path"] == p + suffix + for p in self.native_paths for suffix in ("-wal", "-journal"))] + for path in dict.fromkeys(reads): + _copy(container, observer, path) + if observer.exec_run(["mkdir", "-p", self.declaration["workspace"]], user="root").exit_code: + raise RuntimeError("Observer workspace namespace was not prepared") + script = Path(__file__).with_name("cli_source_probe_entry.py") + declaration = {**self.declaration, **{key: plan[key] for key in ( + "goal_id", "agent_id", "todo_id", "turn_instance_id", "session_id")}} + with tempfile.TemporaryFile() as spool: + with tarfile.open(fileobj=spool, mode="w") as archive: + for name, payload in (("probe.py", script.read_bytes()), + ("declaration.json", json.dumps(declaration).encode())): + entry = tarfile.TarInfo(name) + entry.size, entry.mode = len(payload), 0o444 + archive.addfile(entry, io.BytesIO(payload)) + spool.seek(0) + if not observer.put_archive("/tmp", spool): + raise RuntimeError("Observer read program was not staged") + result = observer.exec_run([declaration["python"], "-I", "-B", "/tmp/probe.py", "/tmp/declaration.json"], + user=self.user, environment={"PATH": declaration["path"], "HOME": declaration["home"], + "PYTHONDONTWRITEBYTECODE": "1", "LOOPX_USAGE_PING": "0"}) + if result.exit_code == 0: + facts.update(json.loads(result.output)) + else: + facts["observer_exit_code"] = result.exit_code # Never expose raw native output or credentials. + finally: + observer.remove(force=True) + facts["source_witness_after"] = self.witness(container, plan) + return facts diff --git a/benchmark/edgebench/cli_source_probe_entry.py b/benchmark/edgebench/cli_source_probe_entry.py new file mode 100644 index 0000000000..4b06a68a3f --- /dev/null +++ b/benchmark/edgebench/cli_source_probe_entry.py @@ -0,0 +1,66 @@ +"""Isolated read program; imports only the source's explicitly copied kernel.""" +from __future__ import annotations + +import json +import hashlib +import os +from pathlib import Path +import sys + + +def main(): + declaration = json.loads(Path(sys.argv[1]).read_bytes()) + root = Path(declaration["kernel_root"]) + sys.path[:] = [str(root), *(p for p in sys.path if p not in ("", "."))] + import loopx + from loopx.control_plane.quota.settlement import _read_heartbeat_settlement_payload + from loopx.control_plane.turn_driver.codex_sessions import load_codex_cli_session + from loopx.control_plane.projects.registry_codec import load_registry + from loopx.control_plane.goals.activation import goal_is_stopped + from loopx.hosts.codex_checkpoint import native_session_checkpoint, _complete_history + from loopx.control_plane.turn_driver.codex_stdio import native_checkpoint_reader + + if Path(loopx.__file__).resolve() != root / "loopx/__init__.py": + raise ValueError("Source observer imported a different kernel") + goal = next(g for g in load_registry(Path(declaration["registry"]))["goals"] + if g["id"] == declaration["goal_id"]) + if goal_is_stopped(goal): + print(json.dumps({"probe_status": "stopped", "source_stopped": True})) + return + runtime = Path(declaration["runtime_root"]) + binding = load_codex_cli_session(runtime, lineage={key: declaration[key] for key in ("goal_id", "agent_id")}, + session_scope="agent") + if not binding or binding["session_id"] != declaration["session_id"]: + raise ValueError("Source native session binding differs") + scheduler = json.loads(Path(declaration["scheduler_state"]).read_bytes()) + scheduler = {"host_attempt": scheduler.get("host_attempt"), + "last_wake_status": scheduler.get("last_wake_status"), + "last_wake_rc": scheduler.get("last_wake_rc"), + "host_pending": json.loads(Path(declaration["scheduler_state"] + ".host-attempt").read_bytes())["pending"]} + settlement = _read_heartbeat_settlement_payload(runtime, goal_id=declaration["goal_id"], fields={ + "agent_id": declaration["agent_id"], "todo_id": declaration["todo_id"], + "turn_instance_id": declaration["turn_instance_id"], "replan_obligation_id": None, + "infer_turn_instance_id": False, "allow_unbound_binding": False}, + registry_path=Path(declaration["registry"]), goal_ref=None) + env = {**os.environ, "CODEX_BIN": declaration["codex"], "CODEX_HOME": declaration["native_home"], + "LOOPX_PROJECT": declaration["workspace"]} + with native_checkpoint_reader(env, timeout=30) as reader: + native = native_session_checkpoint(env, declaration["session_id"], timeout=30, transport=reader) + confirmed = _complete_history(reader.request, declaration["session_id"]) + encoded = json.dumps(confirmed, sort_keys=True, separators=(",", ":")).encode() + if (hashlib.sha256(encoded).hexdigest() != native["metadata"]["visible_history_sha256"] or + any(item.get("status") in {"inProgress", "running", "pending"} + for turn in confirmed for item in turn.get("items", []))): + raise ValueError("Source native history changed or retains pending effects") + native["metadata"]["quiescent_readback"] = True + print(json.dumps({"probe_status": "ready", "scheduler": scheduler, "settlement": settlement, + "native": native["metadata"], "observer_kernel": loopx.__file__})) + + +if __name__ == "__main__": + try: + main() + except Exception as error: + # The existing owner refused or its transport was unavailable. Keep + # raw native/configuration output inside the observer namespace. + print(json.dumps({"probe_status": "unavailable", "observer_error_kind": type(error).__name__})) diff --git a/benchmark/edgebench/cli_source_release.py b/benchmark/edgebench/cli_source_release.py new file mode 100644 index 0000000000..d64b37dd66 --- /dev/null +++ b/benchmark/edgebench/cli_source_release.py @@ -0,0 +1,125 @@ +"""Stage a canonical product archive without promoting an installed profile. + +The caller first creates it with benchmark.runtime.source.archive_source from a +qualified clean revision. No installer, hook, skill, credential or clock runs in +the active worker. The existing launcher is changed only by the barrier owner. +""" +from __future__ import annotations + +import hashlib +from pathlib import Path, PurePosixPath +import posixpath +import shutil +import tarfile +import tempfile + +from .cli_source_activation import _validate_plan +from .cli_source_probe import _archive_response, _path_stat + + +def _inventory(archive, prefix=""): + result = {} + for entry in archive: + path = PurePosixPath(entry.name) + if path.is_absolute() or ".." in path.parts: + raise ValueError("Versioned release archive contains an invalid path") + if prefix: + if not path.parts or path.parts[0] != prefix: + raise ValueError("Versioned release archive differs from its root") + path = PurePosixPath(*path.parts[1:]) + if str(path) == ".": + if not entry.isdir(): + raise ValueError("Versioned release root must be physical") + continue + name = str(path) + if name in result or name == "." or not (entry.isfile() or entry.isdir() or entry.issym()): + raise ValueError("Versioned release archive contains unsupported members") + item = {"mode": entry.mode & 0o777, "kind": entry.type} + if entry.isfile(): + item["sha256"] = hashlib.file_digest(archive.extractfile(entry), "sha256").hexdigest() + if entry.issym(): + # Git's archive owns the exact link bytes. Verification never + # follows them or interprets source material as a command. + item["target"] = entry.linkname + target = posixpath.normpath(posixpath.join(str(path.parent), entry.linkname)) + if target.startswith("../") or target == ".." or target.startswith("/"): + raise ValueError("Versioned release link leaves its source root") + result[name] = item + if any(result.get(str(parent), {}).get("kind") == tarfile.SYMTYPE + for name in result for parent in PurePosixPath(name).parents): + raise ValueError("Versioned release member traverses a symbolic parent") + return result + + +def _physical_directory(container, path): + for ancestor in reversed((PurePosixPath(path), *PurePosixPath(path).parents)): + stat = _path_stat(container, str(ancestor)) + if stat.get("linkTarget") or not stat["mode"] & (1 << 31): + raise ValueError("Versioned release ancestors must be physical directories") + + +def _matches(expected, actual): + # archive_source appends validated generated assets as files. Tar extraction + # creates any omitted physical parent directories; those are not extra + # executable/source members and have no archive-declared permission mode. + parents = {str(parent) for name in expected for parent in PurePosixPath(name).parents if str(parent) != "."} + if any(actual.get(name) != item for name, item in expected.items()): + return False + return all(name in parents and item["kind"] == tarfile.DIRTYPE + for name, item in actual.items() if name not in expected) + + +class StagedCliRelease: + """Exact canonical archive verifier used before and inside the pause.""" + def __init__(self, archive: Path, plan: dict): + _validate_plan(plan) + if archive.resolve(strict=True) != archive or not archive.is_file(): + raise ValueError("CLI source archive must be physical") + with archive.open("rb") as stream: + if hashlib.file_digest(stream, "sha256").hexdigest() != plan["target_archive_sha256"]: + raise ValueError("CLI source archive digest differs") + with tarfile.open(archive) as source: + self.inventory = _inventory(source) + if source.pax_headers.get("comment") != plan["target_revision"]: + raise ValueError("CLI source archive does not bind its Git revision") + launcher = self.inventory.get("scripts/loopx", {}) + if launcher.get("kind") != tarfile.REGTYPE or launcher.get("mode", 0) & 0o111 == 0: + raise ValueError("CLI source archive has no executable launcher") + self.archive, self.plan = archive, plan.copy() + self.root = str(PurePosixPath(plan["new_target"]).parent.parent) + + def verify(self, container, plan): + if plan != self.plan: + raise ValueError("CLI release verification differs from its declared source") + _physical_directory(container, self.root) + with _archive_response(container, self.root) as (chunks, _), tempfile.TemporaryFile() as spool: + for block in chunks: + spool.write(block) + spool.seek(0) + with tarfile.open(fileobj=spool) as archive: + actual = _inventory(archive, PurePosixPath(self.root).name) + if not _matches(self.inventory, actual): + raise ValueError("Staged CLI release bytes, modes or members differ") + + def stage(self, container, *, execute=False): + """Explicit preparation only; never switches an existing launcher.""" + if type(execute) is not bool: + raise ValueError("CLI release staging requires explicit execute") + if not execute: + return {"staged": False, "launcher_changed": False, "execution_authority_granted": False} + with tempfile.TemporaryFile() as snapshot: + with self.archive.open("rb") as stream: + shutil.copyfileobj(stream, snapshot, 2 * 1024 * 1024) + snapshot.seek(0) + if hashlib.file_digest(snapshot, "sha256").hexdigest() != self.plan["target_archive_sha256"]: + raise ValueError("CLI source archive changed after qualification") + snapshot.seek(0) + _physical_directory(container, self.plan["release_root"]) + # mkdir (without -p) rejects a prior/partial release; never overwrite + # a directory or turn an uncertain upload into success. + if container.exec_run(["mkdir", "--", self.root], user="root").exit_code: + raise ValueError("CLI release destination already exists or is unavailable") + if not container.put_archive(self.root, snapshot): + raise RuntimeError("CLI source staging transport was not confirmed") + self.verify(container, self.plan) + return {"staged": True, "launcher_changed": False, "execution_authority_granted": False} diff --git a/benchmark/tests/test_cli_source_activation.py b/benchmark/tests/test_cli_source_activation.py new file mode 100644 index 0000000000..50318a19c0 --- /dev/null +++ b/benchmark/tests/test_cli_source_activation.py @@ -0,0 +1,321 @@ +"""Real daemon transport and failure recovery, without models or work admission. + +Native/settlement semantics retain their owners; these fixtures deliberately +isolate physical pause, immutable source installation and dispatch readback. +""" +from __future__ import annotations + +from copy import deepcopy +import hashlib +import io +import json +import os +import subprocess +import tarfile +import time +import uuid + +import pytest + +from benchmark.edgebench import cli_source_activation as activation +from benchmark.edgebench.cli_source_probe import archive_witness, _archive_response, _path_stat +from benchmark.edgebench.cli_source_release import StagedCliRelease + + +pytestmark = pytest.mark.skipif(os.environ.get("LOOPX_CLI_ACTIVATION_DOCKER_SMOKE") != "1", + reason="Requires explicit isolated Docker CLI-activation qualification") + + +@pytest.fixture +def source(tmp_path): + import docker + + client = docker.from_env(timeout=15) + image = os.environ.get("LOOPX_CHECKPOINT_WRITER_IMAGE", "python:3.12-slim") + client.images.get(image) # No implicit pull or image build. + box = client.containers.run(image, ["sleep", "180"], detach=True, network_disabled=True, + mem_limit="64m", pids_limit=32, labels={"loopx.cli-activation-test": uuid.uuid4().hex}) + try: + result = box.exec_run(["sh", "-c", "mkdir -p /profile/bin /profile/releases/old/scripts /work /native /control; " + "printf '#!/bin/sh\\nprintf old' > /profile/releases/old/scripts/loopx; " + "chmod 755 /profile/releases/old/scripts/loopx; " + "ln -s /profile/releases/old/scripts/loopx /profile/bin/loopx; " + "printf preserved >/work/incumbent; printf original-session >/native/session; " + "printf original-finite-phase >/phase; chmod 400 /phase"]) + assert result.exit_code == 0, result.output + with io.BytesIO() as stream: + payload = json.dumps({"goals": [{"id": "fixture-goal"}]}).encode() + with tarfile.open(fileobj=stream, mode="w") as archive: + entry = tarfile.TarInfo("registry.json") + entry.size = len(payload) + archive.addfile(entry, io.BytesIO(payload)) + stream.seek(0) + assert box.put_archive("/control", stream) + repo = tmp_path / "product" + (repo / "scripts").mkdir(parents=True) + (repo / "scripts/loopx").write_text("#!/bin/sh\nprintf new\n") + (repo / "scripts/loopx").chmod(0o755) + for args in (["init", "-q"], ["add", "scripts/loopx"], ["-c", "user.name=Fixture", "-c", + "user.email=fixture@example.test", "commit", "-qs", "-m", "Synthetic executable release"]): + subprocess.run(["git", "-C", str(repo), *args], check=True, capture_output=True) + revision = subprocess.check_output(["git", "-C", str(repo), "rev-parse", "HEAD"], text=True).strip() + from benchmark.runtime.source import archive_source + + archive = tmp_path / "source.tar" + archive_source(repo, revision, archive) + box.reload() + plan = {"activation_id": uuid.uuid4().hex, "container_id": box.id, + "started_at": box.attrs["State"]["StartedAt"], "goal_id": "fixture-goal", "agent_id": "fixture-agent", + "todo_id": "todo_fixture0001", "turn_instance_id": "original-turn", "session_id": "original-native-session", + "registry_path": "/control/registry.json", "launcher": "/profile/bin/loopx", "launcher_uid": 0, "launcher_gid": 0, + "old_target": "/profile/releases/old/scripts/loopx", + "new_target": "/profile/releases/new/scripts/loopx", "release_root": "/profile/releases", + "target_revision": revision, "target_archive_sha256": hashlib.sha256(archive.read_bytes()).hexdigest(), + "worker_lock": str(activation._worker_lock(client, box.id))} + release = StagedCliRelease(archive.resolve(), plan) + assert release.stage(box)["staged"] is False + assert release.stage(box, execute=True)["staged"] is True + receipt_dir = tmp_path / "controller" + receipt_dir.mkdir(mode=0o700) + receipt = receipt_dir / "activation.json" + components = [{"id": "work", "source_path": "/work"}, {"id": "native", "source_path": "/native"}, + {"id": "phase", "source_path": "/phase"}] + control = ["continue"] + def witness(container, _): + return archive_witness(container, components) + original_witness = witness(box, plan) + + def facts(container, _): + identity = {"schema_version": "quota_settlement_identity_v0", + **{k: plan[k] for k in ("goal_id", "agent_id", "todo_id", "turn_instance_id")}, + "effect_id": ":".join(plan[k] for k in ("goal_id", "agent_id", "todo_id", "turn_instance_id"))} + return {"ready_present": False, "unknown_process_ids": [], "probe_status": "ready", + "scheduler": {"host_attempt": None, "host_pending": False, "last_wake_status": "wake_ok", "last_wake_rc": 0}, + "settlement": {"identity": {"result": {"value": identity, "failure": None}}, "progress": {"state": "settled"}}, + "native": {"session_id": plan["session_id"], "turn_ids": ["native-original"], + "completed_turn_id": "native-original", "visible_history_sha256": "e" * 64, + "model_calls": 0, "execution_authority_granted": False, "quiescent_readback": True}, + "source_witness_before": witness(container, plan), "source_witness_after": witness(container, plan)} + + params = {"client": client, "plan": plan, "receipt": receipt, "preflight": facts, "probe": facts, + "witness": witness, "verify_release": release.verify, "control": lambda: control[0]} + yield box, params, control + assert witness(box, plan) == original_witness + finally: + box.remove(force=True) + assert not client.containers.list(all=True, filters={"id": box.id}) + client.close() + + +def test_preview_and_exact_activation_preserve_work_session_phase_and_do_not_replay(source, monkeypatch): + box, params, _ = source + assert activation.activate_cli_source(**params)["source_change_allowed"] is False + assert not params["receipt"].exists() + writes = [] + original = activation._put_link + + def write(*args, **kwargs): + writes.append(args[2]) + return original(*args, **kwargs) + + monkeypatch.setattr(activation, "_put_link", write) + result = activation.activate_cli_source(**params, execute=True) + assert result["phase"] == "released" and result["applied"] is True + assert result["model_receipt_adoption"] == "unobserved" + assert result["work_or_clock_mutations"] == 0 and result["pause_wall_seconds"] > 0 + assert box.exec_run(["/profile/bin/loopx"]).output == b"new" + assert params["receipt"].stat().st_mode & 0o777 == 0o600 + original_receipt = params["receipt"].read_bytes() + historical = activation.activate_cli_source(**params, execute=True, reconcile=True) + assert historical["historical_receipt"] is True and historical["replayed"] is True + assert historical["current_source_verified"] is False and historical["applied"] is True + assert historical["model_receipt_adoption"] == "unobserved" + assert params["receipt"].read_bytes() == original_receipt + assert writes == [params["plan"]["new_target"]] + + +@pytest.mark.parametrize("failure", ["lost-write-ack", "rejected-before-write", "partial-launcher", "lost-release-ack"]) +def test_transport_ambiguity_is_read_back_before_release_and_never_repeats_dispatch(source, monkeypatch, failure): + box, params, _ = source + original = activation._put_link + + def faulty(container, path, target, **kwargs): + if failure == "lost-write-ack": + original(container, path, target, **kwargs) + elif failure == "partial-launcher": + with io.BytesIO() as stream: + with tarfile.open(fileobj=stream, mode="w") as archive: + member = tarfile.TarInfo("loopx") + member.size = 7 + archive.addfile(member, io.BytesIO(b"partial")) + stream.seek(0) + assert container.put_archive("/profile/bin", stream) + raise OSError("Injected bounded transport failure") + + if failure == "lost-release-ack": + from docker.models.containers import Container + + unpause = Container.unpause + + def lost_release(container): + unpause(container) + if container.id == box.id: + raise OSError("Injected lost release acknowledgement") + + monkeypatch.setattr(Container, "unpause", lost_release) + else: + monkeypatch.setattr(activation, "_put_link", faulty) + if failure in {"partial-launcher", "lost-release-ack"}: + with pytest.raises((activation.SourceActivationHeld, OSError)): + activation.activate_cli_source(**params, execute=True) + box.reload() + assert box.attrs["State"]["Paused"] is (failure == "partial-launcher") + if failure == "partial-launcher": + assert activation._link(box, params["plan"]["launcher"]) is None + monkeypatch.setattr(activation, "_put_link", original) + result = activation.activate_cli_source(**params, execute=True, reconcile=True, repair="new") + else: + result = activation.activate_cli_source(**params, execute=True, reconcile=True) + assert result["phase"] == "released" + assert box.exec_run(["/profile/bin/loopx"]).output == b"new" + else: + result = activation.activate_cli_source(**params, execute=True) + assert result["phase"] == "released" + expected = b"new" if failure == "lost-write-ack" else b"old" + assert box.exec_run(["/profile/bin/loopx"]).output == expected + assert result["applied"] is (failure == "lost-write-ack") + + +@pytest.mark.parametrize("case", ["busy-before-pause", "ready-after-pause", "Stop-before-pause", "Stop-after-pause", "unsettled-after-pause"]) +def test_busy_unknown_and_stop_cannot_change_the_launcher(source, case): + box, params, control = source + if case == "Stop-before-pause": + control[0] = "stop" + elif case == "busy-before-pause": + original = params["preflight"] + params["preflight"] = lambda *args: {**original(*args), "unknown_process_ids": [123]} + else: + original = params["probe"] + + def changed(*args): + value = deepcopy(original(*args)) + if case == "ready-after-pause": + value["ready_present"] = True + elif case == "Stop-after-pause": + control[0] = "stop" + else: + value["settlement"]["progress"]["state"] = "spend_receipt_required" + return value + + params["probe"] = changed + if case == "Stop-after-pause": + with pytest.raises(activation.SourceActivationHeld): + activation.activate_cli_source(**params, execute=True) + box.reload() + assert box.attrs["State"]["Paused"] is True + # Stop remains the release authority. A fresh explicit recovery after + # revocation can restore only the unchanged, original source here. + control[0] = "continue" + activation.activate_cli_source(**params, execute=True, reconcile=True) + else: + result = activation.activate_cli_source(**params, execute=True) + assert result["applied"] is False + assert activation._link(box, params["plan"]["launcher"]) == params["plan"]["old_target"] + assert box.exec_run(["/profile/bin/loopx"]).output == b"old" + + +def test_release_bytes_and_current_launcher_are_compared_not_overwritten(source): + box, params, _ = source + changed = box.exec_run(["sh", "-c", "printf changed >/profile/releases/new/scripts/loopx"]) + assert changed.exit_code == 0 + with pytest.raises(ValueError, match="differ"): + activation.activate_cli_source(**params, execute=True) + assert not params["receipt"].exists() + assert activation._link(box, params["plan"]["launcher"]) == params["plan"]["old_target"] + + +def test_recovery_rejects_another_intent_and_public_receipt_permissions(source): + _, params, _ = source + activation.activate_cli_source(**params, execute=True) + other = {**params, "plan": {**params["plan"], "activation_id": "f" * 32}} + with pytest.raises(ValueError, match="original intent"): + activation.activate_cli_source(**other, execute=True, reconcile=True) + params["receipt"].chmod(0o644) + with pytest.raises(ValueError, match="private physical receipt"): + activation.activate_cli_source(**params, execute=True, reconcile=True) + + +def test_native_head_and_an_unstarted_get_close_release_the_daemon_container_lock(source): + box, _, _ = source + assert box.exec_run(["python", "-c", "from pathlib import Path; Path('/archive-probe').write_bytes(b'x' * 4194304)"]).exit_code == 0 + box.pause() + started = time.monotonic() + for _ in range(3): + assert _path_stat(box, "/archive-probe")["size"] == 4194304 + # The caller can reject an archive before iterating a single body chunk. + # Own Response.close rather than closing Docker SDK's unstarted generator. + with _archive_response(box, "/archive-probe"): + pass + box.reload() + assert box.attrs["State"]["Paused"] is True + assert time.monotonic() - started < 5 + box.unpause() + + +@pytest.mark.parametrize("partial", [{"phase": phase} for phase in ( + "pause_requested", "pause_uncertain", "unknown-phase", None, 17)] + [ + {"phase": "paused", "pause_acknowledged": True}, + {"phase": "qualified", "pause_acknowledged": True, "paused_at_epoch": "unconfirmed"}, +]) +def test_unacknowledged_or_unknown_intent_cannot_acquire_another_actors_pause(source, monkeypatch, partial): + box, params, _ = source + from docker.models.containers import Container + + pause = Container.pause + + def controller_died_before_pause(container): + if container.id == box.id: + raise KeyboardInterrupt("Injected controller death before Docker pause") + return pause(container) + + monkeypatch.setattr(Container, "pause", controller_died_before_pause) + with pytest.raises(KeyboardInterrupt): + activation.activate_cli_source(**params, execute=True) + value = json.loads(params["receipt"].read_bytes()) + assert value["phase"] == "pause_requested" and "pause_acknowledged" not in value + value.update(partial) + params["receipt"].write_text(json.dumps(value)) + pause(box) # A different actor establishes its own real physical barrier. + with pytest.raises(activation.SourceActivationHeld, match="acknowledgement"): + activation.activate_cli_source(**params, execute=True, reconcile=True) + box.reload() + assert box.attrs["State"]["Paused"] is True + assert activation._link(box, params["plan"]["launcher"]) == params["plan"]["old_target"] + assert json.loads(params["receipt"].read_bytes())["phase"] == partial["phase"] + + +def test_repair_rechecks_stop_after_target_verification(source, monkeypatch): + box, params, control = source + write = activation._put_link + + def interrupted(*_args, **_kwargs): + raise KeyboardInterrupt("Injected controller death after confirmed pause") + + monkeypatch.setattr(activation, "_put_link", interrupted) + with pytest.raises(KeyboardInterrupt): + activation.activate_cli_source(**params, execute=True) + value = json.loads(params["receipt"].read_bytes()) + assert value["pause_acknowledged"] is True and value["phase"] == "switch_requested" + verify = params["verify_release"] + + def stop_after_verification(*args): + verify(*args) + control[0] = "stop" + + params["verify_release"] = stop_after_verification + monkeypatch.setattr(activation, "_put_link", write) + with pytest.raises(activation.SourceActivationHeld, match="stopped before"): + activation.activate_cli_source(**params, execute=True, reconcile=True, repair="new") + box.reload() + assert box.attrs["State"]["Paused"] is True + assert activation._link(box, params["plan"]["launcher"]) == params["plan"]["old_target"] diff --git a/docs/reference/protocols/README.md b/docs/reference/protocols/README.md index 443dc4f8cd..bb3a1ff16e 100644 --- a/docs/reference/protocols/README.md +++ b/docs/reference/protocols/README.md @@ -68,6 +68,7 @@ until their documented migration is qualified. - [`computer_use_runtime_v0`](computer-use-runtime-v0.md): Computer-use runtime v0 - [`host_integration_plugin_plan_v0`](host-integration-plugin-plan-v0.md): Host integration plugin plan v0 - [`host_integration_surface_v0`](host-integration-surface-v0.md): Host integration surface v0 +- [Host CLI source activation v0](host-cli-source-activation-v0.md): explicit paused Docker adapter for a qualified versioned CLI, with exact-intent recovery - [`host_mode_plan_v0`](host-mode-plan-v0.md): Host mode plan v0 - [`local_agent_launch_plan_v1`](local-agent-launch-plan-v1.md): Local agent launch plan v1 - [`openviking_session_memory_adapter_v0`](openviking-session-memory-adapter-v0.md): OpenViking session memory adapter v0 diff --git a/docs/reference/protocols/host-cli-source-activation-v0.md b/docs/reference/protocols/host-cli-source-activation-v0.md new file mode 100644 index 0000000000..95f7303e4d --- /dev/null +++ b/docs/reference/protocols/host-cli-source-activation-v0.md @@ -0,0 +1,130 @@ +# Host CLI source activation v0 + +This is an explicit Docker operation adapter for an existing benchmark-toolkit +controller. It changes the versioned CLI used by future subprocesses while the +running Host keeps its imported modules. It is not a new execution provider, +Goal operation, application setting or automatic update service. App, Lark, +ordinary CLI and existing checkpoint callers do not invoke it by default. + +## Owners and entrypoints + +`benchmark.edgebench.cli_source_activation.activate_cli_source` provides preview, +execute and exact-intent reconcile. The controller supplies its current ordinary +Stop, phase and resource observations; none of these methods grants a run slot, +work admission, account authority or additional allowance. + +The registered TypeScript handler `benchmark.cli_source_activation` qualifies the +boundary and chooses permitted readback/repair effects. Python adapts Docker +pause/archive and persists transport acknowledgements. Its journal phases remain +local to this adapter, separate from Goal, Todo, quota and native-session state. +The existing quota settlement, registry activation, native-history, source archive +and physical worker-lock owners retain their semantics. + +The operation requires an exact daemon/container/start identity, original +Goal/Agent/Todo/Turn/native session and registry route, the owned launcher path and uid/gid, distinct +versioned old/new targets, and the qualified new revision/archive digest. Targets +have the existing `releases//scripts/loopx` layout. The launcher must be a +physical Linux-profile symlink with its original ownership and mode. A different +launcher, namespace or ownership is a failed compare, never overwrite permission. + +1. Produce the canonical archive with `benchmark.runtime.source.archive_source` + from the independently qualified clean revision. `StagedCliRelease.stage` + previews by default. Explicit staging creates a fresh versioned directory, + verifies every declared byte/mode/member and never promotes the launcher or + runs the full installer. Implicit physical tar parent directories are allowed; + escaping links, symbolic parents and unexpected executable/source members + are rejected. Preserve a partial upload; do not overwrite it as a retry. +2. Compose a `PausedSourceProbe` from the original runtime's explicit participants, + native checkpoint store inventory, state-backup authority inventory and qualified + immutable reader dependencies. Its exact passive-process manifest comes from + the owned Host contract, not learning every process of a busy source as idle. + A passive app-server is not by itself an outstanding model request. +3. Preview `activate_cli_source(..., execute=False)`. On explicit execution it + takes the same daemon/full-container worker lock as the old checkpoint Host, + in the same controller user's physical filesystem realm. It does not create + a controller-ready request, checkpoint identity or capture attempt. +4. A bounded preflight rejects busy/ready/unknown candidates before pausing. After + the pause, the observer reads an isolated, network-disabled copy with existing + owners. It verifies the exact original settled identity, no pending Host + attempt, stable completed native history, no unfinished native effects, no + unknown processes and an unchanged source witness. SQLite WAL/journal files + participate. Credential homes/auth are excluded; source work never executes + in the observer. Work, research, authority, hooks, native configuration, + profile and original phase participants remain in place. +5. Persist the private intent before changing the launcher. Docker archive upload + is **not** a filesystem-atomic rename. The physical pause and shared lock make + this change atomic with respect to dispatch. Verify the complete target and + current Stop again, write the symlink with its original uid/gid, then read back + its exact identity and unchanged protected source before releasing the pause. + +The probe declaration is an integration input, not another saved product policy. +`source_goal_control` reads the existing registry codec/activation owner. The +caller must combine it with its ordinary current controller/phase/resource Stop +path. A cached profile flag or a prepared deadline is not a current guard. + +## Failure, Stop and recovery + +The private, physical, owner-only receipt is the operation's transport readback. +Keep it and the exact plan. `reconcile=True` never replays an already released +operation, starts a model, reruns a guard or rewrites work state. A released +receipt returns an explicitly historical/replayed projection with current source +ownership and model adoption unverified; it is not a fresh activation guard. + +- A busy or rejected paused observation can resume only the known original + launcher under current ordinary control, without an installation change. +- Stop before pause makes no effect. Stop during the transaction keeps the source + held for the ordinary Stop owner; a generic `finally` must not resume it. +- A lost write acknowledgement is resolved by actual old/new readback. A partial, + missing or foreign launcher stays paused. Explicit `repair="old"` or `"new"` + needs the same qualified unchanged source and current control; it repairs only + the launcher transport, never native history, work or a model dispatch. +- A lost release acknowledgement reconciles the running source and exact target + without another unpause or dispatch. An uncertain pause acknowledgement remains + held for explicit ownership recovery; the adapter cannot release another + actor's pause by guessing. A pending request, unknown/partial phase or missing + durable pause acknowledgement cannot acquire ownership of an observed pause. +- Process death or an unavailable probe leaves a durable intent. Preserve the + original namespace and reconcile it; do not create a second activation identity + or use an old checkpoint as rollback. + +A clean deferred/released-old result did not install the new CLI. A subsequent +candidate requires a fresh current boundary and intent after the prior operation +is known released. It cannot reuse historical qualification; an uncertain prior +operation remains held and must be reconciled first. + +All setup and pause time remains charged to the original finite allowance. +The adapter records actual pause wall time and never resets, refunds, extends or +regrants the phase. Native counters, cooldown, evaluator and incumbent are not +called or changed by installation. + +## Adoption and compatibility + +The released receipt is a source amendment for the controller's existing run +lineage: old imported Host plus the exact new CLI/kernel. It is not a whole-Host +upgrade, a native restore or proof that a model consumed the new behavior. Keep +the original frozen profile/digests and earlier source segment. The controller +must retain the amendment with future source/checkpoint provenance. + +Qualify the actual old Host's next wake, including checkpoint preflight, hook +preparation/SessionStart, child Python source, TypeScript runtime fingerprint and +the same-session model-visible input/readback. Preserve the actual publisher's +manual/composed settlement contract and prove original identity, one debit and +replay behavior. Do not select a new execution arm or invent a validator to make +an update appear enabled. A schema label, installed file or running container +alone is not semantic adoption or an effectiveness/cost improvement. + +## 中文边界 + +这是现有 benchmark-toolkit controller 的显式 Docker 操作适配器，不增加产品设置、 +执行臂、预算或 Goal/Todo owner。运行中的 Host 保留已导入代码；新 CLI 仅影响后续子进程。 +默认不执行；当前 App/Lark/普通 CLI 与旧 checkpoint 行为保持原调用边界。 + +必须先在同一物理 worker 锁下筛选，再暂停后重新用原生会话、真实 settlement 和状态 +owner 验证原身份已结算且无未决工作。不能将常驻 app-server、最近一次成功状态或旧 +checkpoint 当作当前静止证明。新源码先独立安装验证，只对 owned launcher 做精确比较 +后切换；不复制认证，不改原 profile、工作区、状态、原生机会或有限时钟。 + +写入中断或回执丢失必须读回原/新目标；未知目标保持暂停。Stop 优先，禁止在通用清理 +分支盲目恢复。恢复只处理同一操作的物理传输，不重发模型、guard、准入或结算。实际 +暂停成本不返还。交付记录明确混合版本，并由原 controller 保存 source amendment； +下一自然输入和真实读回另证采用，不能把安装成功、v1/v2 标签或测试数当作收益。 diff --git a/loopx/control_plane/capabilities/benchmark_cli_source_activation.ts b/loopx/control_plane/capabilities/benchmark_cli_source_activation.ts new file mode 100644 index 0000000000..bd322193a2 --- /dev/null +++ b/loopx/control_plane/capabilities/benchmark_cli_source_activation.ts @@ -0,0 +1,127 @@ +/** Explicit, quiescent activation of an already staged CLI release. + * The host observes its physical writer barrier and persists these decisions. + * This neither grants execution nor changes work, native history, or clocks. + */ +import type {JsonObject, SettlementIdentityInput} from "../effect_program.ts"; +import {settlementIdentity} from "../effect_program.ts"; +import {hasExactAuthorityKeys} from "../coordination/authority_store_codec.ts"; +import {requireJsonObject, requireNonEmptyString} from "../runtime_decode.ts"; +import {BARE_SHA256_PATTERN} from "../content_digest.ts"; +import {posix} from "node:path"; + +function sha(value: unknown, label: string): string { + const text = requireNonEmptyString(value, label); + if (!BARE_SHA256_PATTERN.test(text)) throw new Error(`${label} must be SHA-256`); + return text; +} + +function denied(reason: string, held: boolean): JsonObject { + return {qualified: false, reason, source_change_allowed: false, + resume_allowed: false, reconcile_required: held, execution_authority_granted: false}; +} + +/** Native/settlement facts come from their existing owners over the paused copy. + * In particular, a passive app-server is not an active model request. Unknown + * processes and an unqualified native observation cannot establish idleness. + */ +export function qualifyCliSourceActivation(value: unknown): JsonObject { + const input = requireJsonObject(value, "CLI source activation"); + if (!hasExactAuthorityKeys(input, ["operation", "scope", "observation"])) { + throw new Error("invalid CLI source activation fields"); + } + const scope = requireJsonObject(input.scope, "activation scope"); + if (!hasExactAuthorityKeys(scope, ["activation_id", "container_id", "started_at", "goal_id", "agent_id", "todo_id", "turn_instance_id", "session_id", + "registry_path", "launcher", "launcher_uid", "launcher_gid", "old_target", "new_target", "release_root", "target_revision", "target_archive_sha256", "worker_lock"])) { + throw new Error("invalid CLI source activation scope"); + } + for (const key of ["goal_id", "agent_id", "todo_id", "turn_instance_id", "session_id", "started_at"]) requireNonEmptyString(scope[key], key); + for (const [key, width] of [["activation_id", 32], ["container_id", 64], ["target_revision", 40]] as const) { + if (!new RegExp(`^[a-f0-9]{${width}}$`).test(requireNonEmptyString(scope[key], key))) throw new Error(`invalid ${key}`); + } + sha(scope.target_archive_sha256, "target source archive"); + for (const key of ["launcher_uid", "launcher_gid"]) { + if (!Number.isSafeInteger(scope[key]) || (scope[key] as number) < 0) throw new Error(`invalid ${key}`); + } + for (const key of ["registry_path", "launcher", "old_target", "new_target", "release_root", "worker_lock"]) { + const path = requireNonEmptyString(scope[key], key); + if (!posix.isAbsolute(path) || posix.normalize(path) !== path) throw new Error(`invalid ${key}`); + } + const releaseRoot = scope.release_root as string; + for (const key of ["old_target", "new_target"]) { + const relative = posix.relative(releaseRoot, scope[key] as string).split("/"); + if (relative.length !== 3 || relative[0] === ".." || relative[1] !== "scripts" || relative[2] !== "loopx") { + throw new Error("CLI activation must select an owned versioned launcher"); + } + } + if (scope.old_target === scope.new_target || (scope.launcher as string).startsWith(releaseRoot + "/")) { + throw new Error("CLI activation needs distinct releases and a separate launcher"); + } + const observed = requireJsonObject(input.observation, "activation observation"); + const held = observed.paused === true; + if (observed.control !== "continue" || observed.source_stopped === true || + ((input.operation !== "preflight" || observed.source_control !== undefined) && observed.source_control !== "continue")) { + return denied(observed.control === "stop" || observed.source_stopped === true || observed.source_control === "stop" ? "stop_requested" : "control_unknown", held); + } + if (observed.container_id !== scope.container_id || observed.started_at !== scope.started_at || + observed.running !== true) return denied("source_namespace_unverified", held); + if (input.operation === "preflight") { + const scheduler = requireJsonObject(observed.scheduler, "source scheduler"); + const unknown = observed.unknown_process_ids; + const ready = observed.paused === false && observed.ready_present === false && + Array.isArray(unknown) && unknown.length === 0 && scheduler.host_pending === false && + scheduler.host_attempt === null && scheduler.last_wake_status === "wake_ok" && scheduler.last_wake_rc === 0; + return {qualified: false, pause_allowed: ready, reason: ready ? "candidate_idle_boundary" : "source_busy_or_unknown", + source_change_allowed: false, resume_allowed: false, execution_authority_granted: false}; + } + if (!held || observed.owned_barrier !== true) return denied("writer_barrier_unverified", held); + if (input.operation === "readback" || input.operation === "repair") { + const oldTarget = requireNonEmptyString(scope.old_target, "old launcher target"); + const newTarget = requireNonEmptyString(scope.new_target, "new launcher target"); + if (oldTarget === newTarget) throw new Error("CLI source activation needs distinct releases"); + if (observed.unchanged_source_witness !== true) return denied("source_changed_under_barrier", true); + if (input.operation === "repair") { + if (observed.qualified_source !== true || !["old", "new"].includes(observed.repair_target as string)) { + return denied("repair_source_unqualified", true); + } + return {qualified: true, source_change_allowed: true, resume_allowed: false, + repair_target: observed.repair_target, execution_authority_granted: false}; + } + const target = observed.launcher_target; + const disposition = target === newTarget ? "new" : target === oldTarget ? "old" : "unknown"; + const identityMatches = observed.launcher_uid === scope.launcher_uid && observed.launcher_gid === scope.launcher_gid && + observed.launcher_mode === 0o777; + const known = identityMatches && (disposition === "old" || (disposition === "new" && observed.qualified_source === true)); + return {qualified: known, disposition, source_change_allowed: false, + resume_allowed: known, reconcile_required: !known, + execution_authority_granted: false}; + } + if (input.operation !== "qualify") throw new Error("unsupported CLI source activation operation"); + if (observed.ready_present !== false) return denied("checkpoint_handoff_pending", true); + if (observed.launcher_target !== scope.old_target || observed.launcher_uid !== scope.launcher_uid || + observed.launcher_gid !== scope.launcher_gid || observed.launcher_mode !== 0o777) return denied("launcher_compare_failed", true); + if (observed.probe_status !== "ready") return denied("source_quiescence_unavailable", true); + const unknown = observed.unknown_process_ids; + if (!Array.isArray(unknown) || unknown.length) return denied("process_quiescence_unknown", true); + const scheduler = requireJsonObject(observed.scheduler, "source scheduler"); + if (scheduler.host_pending !== false || scheduler.host_attempt !== null || + scheduler.last_wake_status !== "wake_ok" || scheduler.last_wake_rc !== 0) return denied("host_turn_pending", true); + const settlement = requireJsonObject(observed.settlement, "source settlement"); + const identityResult = requireJsonObject(requireJsonObject(settlement.identity, "source identity").result, "source identity result"); + if (identityResult.failure !== null || identityResult.value === null) return denied("settlement_identity_unknown", true); + const identity = settlementIdentity(requireJsonObject(identityResult.value, "settlement identity") as unknown as SettlementIdentityInput); + if (identity.goal_id !== scope.goal_id || identity.agent_id !== scope.agent_id || + identity.todo_id !== scope.todo_id || identity.turn_instance_id !== scope.turn_instance_id || + requireJsonObject(settlement.progress, "source settlement progress").state !== "settled") return denied("original_turn_unsettled", true); + const native = requireJsonObject(observed.native, "native source cursor"); + if (native.execution_authority_granted !== false || native.model_calls !== 0 || + native.quiescent_readback !== true || + native.session_id !== scope.session_id || !Array.isArray(native.turn_ids) || !native.turn_ids.length || + native.completed_turn_id !== native.turn_ids.at(-1) || native.interrupted_guard !== undefined) return denied("native_cursor_not_idle", true); + sha(native.visible_history_sha256, "native visible history"); + const before = sha(observed.source_witness_before, "source witness"); + if (sha(observed.source_witness_after, "confirmed source witness") !== before) return denied("source_changed_during_observation", true); + return {qualified: true, reason: "settled_native_idle_at_owned_barrier", source_change_allowed: true, + resume_allowed: false, reconcile_required: false, execution_authority_granted: false, + source_witness_sha256: before, original_settlement_identity: identity, + native_session_id: native.session_id, native_completed_turn_id: native.completed_turn_id}; +} diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index c810e79047..d5d5fde390 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -289,6 +289,7 @@ export function createEffectRuntimeHandlers( ["benchmark.trajectory.qualify", lazyHandler(() => import("./capabilities/trajectory_assessment.ts"), ({qualifyTrajectoryAssessment}) => qualifyTrajectoryAssessment)], ["benchmark.trajectory.continuation", lazyHandler(() => import("./capabilities/trajectory_assessment_continuation.ts"), ({compareTrajectoryAssessments}) => compareTrajectoryAssessments)], ["benchmark.terminal-followthrough.prepare", lazyHandler(() => import("./capabilities/benchmark_terminal_followthrough.ts"), ({prepareBenchmarkTerminalFollowthrough}) => prepareBenchmarkTerminalFollowthrough)], + ["benchmark.cli_source_activation", lazyHandler(() => import("./capabilities/benchmark_cli_source_activation.ts"), ({qualifyCliSourceActivation}) => qualifyCliSourceActivation)], ["agent.preferences", lazyHandler(() => import("./capabilities/agent_preferences.ts"), ({agentPreferences}) => agentPreferences)], ["todo.archive.capture_dependencies", lazyHandler(() => Promise.all([import("./todos/archive_capture.ts"), import("./coordination/source_transfer.ts")]), ([{captureArchivedTodoDependencies}, {withCoordinationSourceTransfer}]) => withCoordinationSourceTransfer("todo.archive.capture_dependencies", captureArchivedTodoDependencies))], ["agent.supervisor.plan_append", lazyHandler(() => import("./agents/supervisor_event_append.ts"), ({planSupervisorEventAppend}) => planSupervisorEventAppend)], diff --git a/tests/control_plane/test_cli_source_activation_readback.py b/tests/control_plane/test_cli_source_activation_readback.py new file mode 100644 index 0000000000..8ba75f1650 --- /dev/null +++ b/tests/control_plane/test_cli_source_activation_readback.py @@ -0,0 +1,75 @@ +"""Use real CLI authority receipts; native metadata is a separate transport gate.""" +import pytest + +from loopx.control_plane.effect_runtime import effect_runtime_result +from loopx.control_plane.quota.settlement import _read_heartbeat_settlement_payload +from loopx.control_plane.projects.registry_codec import load_registry +from loopx.control_plane.goals.activation import goal_is_stopped +from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime +from tests.control_plane.test_quota_settlement_cli import ( + GOAL_ID, AGENT_ID, TODO_ID, TURN_ID, _write_fixture, _configure_read_only_todo, + _initialize_fixture_authority, _run_cli, _run_generated_cli, _spend_run_count, +) + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_activation_reads_the_original_real_settlement_without_writing_or_replaying_it(tmp_path, monkeypatch, provider): + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_fixture(tmp_path / "original") + _configure_read_only_todo(project) + _initialize_fixture_authority(project, runtime, provider=provider) + rc, packet = _run_cli(registry, runtime, "quota", "should-run", "--runtime-profile", "generic_cli", + "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--todo-id", TODO_ID, "--turn-instance-id", TURN_ID) + assert rc == 0, packet + scope = {"activation_id": "a" * 32, "container_id": "b" * 64, "started_at": "original-start", + "goal_id": GOAL_ID, "agent_id": AGENT_ID, "todo_id": TODO_ID, "turn_instance_id": TURN_ID, + "session_id": "original-session", "registry_path": str(registry), "launcher": "/profile/bin/loopx", "launcher_uid": 1000, "launcher_gid": 1000, + "old_target": "/profile/releases/old/scripts/loopx", "new_target": "/profile/releases/new/scripts/loopx", + "release_root": "/profile/releases", "target_revision": "c" * 40, + "target_archive_sha256": "d" * 64, "worker_lock": "/controller/worker"} + + def decision(): + goal = next(g for g in load_registry(registry)["goals"] if g["id"] == GOAL_ID) + readback = _read_heartbeat_settlement_payload(runtime, goal_id=GOAL_ID, + fields={"agent_id": AGENT_ID, "todo_id": TODO_ID, "turn_instance_id": TURN_ID, + "replan_obligation_id": None, "infer_turn_instance_id": False, "allow_unbound_binding": False}, + registry_path=registry, goal_ref=None) + return effect_runtime_result("benchmark.cli_source_activation", {"operation": "qualify", "scope": scope, + "observation": {"container_id": scope["container_id"], "started_at": scope["started_at"], + "running": True, "paused": True, "owned_barrier": True, "control": "continue", + "source_control": "stop" if goal_is_stopped(goal) else "continue", + "ready_present": False, "launcher_target": scope["old_target"], "launcher_uid": 1000, + "launcher_gid": 1000, "launcher_mode": 0o777, "probe_status": "ready", "unknown_process_ids": [], + "scheduler": {"host_pending": False, "host_attempt": None, "last_wake_status": "wake_ok", "last_wake_rc": 0}, + "settlement": readback, "source_witness_before": "e" * 64, "source_witness_after": "e" * 64, + "native": {"execution_authority_granted": False, "model_calls": 0, "quiescent_readback": True, + "session_id": scope["session_id"], "turn_ids": ["completed-native-turn"], + "completed_turn_id": "completed-native-turn", "visible_history_sha256": "f" * 64}}}) + + assert decision()["source_change_allowed"] is False + assert _spend_run_count(runtime) == 0 + plan = packet["interaction_contract"]["cli_channel"]["settlement_plan"] + writeback = plan["ordered_steps"][1]["command_template"] + for old, new in (("", "verified_exclusion"), ("", "single_surface"), + ("", "outcome_progress")): + writeback = writeback.replace(old, new) + rc, written = _run_generated_cli(writeback + " --progress-result-class advanced --progress-evidence-id exclusion-proof" + " --delivery-boundary in_flight_continuation --no-global-sync --suppress-external-sinks", registry_path=registry) + assert rc == 0, written + assert decision()["source_change_allowed"] is False + rc, spent = _run_generated_cli(written["settlement_owed"]["command"], registry_path=registry) + assert rc == 0, spent + result = decision() + assert result["source_change_allowed"] is True + assert result["original_settlement_identity"]["turn_instance_id"] == TURN_ID + assert decision() == result + assert _spend_run_count(runtime) == 1 + stop = ("goal-lifecycle", "--goal-id", GOAL_ID, "--operation", "stop") + rc, preview = _run_cli(registry, runtime, *stop, cwd=project) + assert rc == 0, preview + rc, stopped = _run_cli(registry, runtime, *stop, "--actor-kind", "owner", + "--expected-state-fingerprint", preview["observed_state_fingerprint"], "--execute", cwd=project) + assert rc == 0 and stopped["readback"]["verified"], stopped + denied = decision() + assert denied["source_change_allowed"] is False and denied["reason"] == "stop_requested" + assert _spend_run_count(runtime) == 1 diff --git a/tests/control_plane_ts/cli_source_activation.test.ts b/tests/control_plane_ts/cli_source_activation.test.ts new file mode 100644 index 0000000000..6ffd9bb904 --- /dev/null +++ b/tests/control_plane_ts/cli_source_activation.test.ts @@ -0,0 +1,90 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import {qualifyCliSourceActivation as decide} from "../../loopx/control_plane/capabilities/benchmark_cli_source_activation.ts"; + +const scope = {activation_id: "a".repeat(32), container_id: "b".repeat(64), started_at: "original-start", + goal_id: "fixture-goal", agent_id: "fixture-agent", todo_id: "todo_fixture0001", turn_instance_id: "original-turn", + session_id: "original-native-session", registry_path: "/control/registry.json", launcher: "/profile/bin/loopx", launcher_uid: 0, launcher_gid: 0, release_root: "/profile/releases", + old_target: "/profile/releases/old/scripts/loopx", new_target: "/profile/releases/new/scripts/loopx", + target_revision: "c".repeat(40), target_archive_sha256: "d".repeat(64), worker_lock: "/controller/worker.lock"}; +const identity = {schema_version: "quota_settlement_identity_v0", goal_id: scope.goal_id, agent_id: scope.agent_id, + todo_id: scope.todo_id, turn_instance_id: scope.turn_instance_id, + effect_id: `${scope.goal_id}:${scope.agent_id}:${scope.todo_id}:${scope.turn_instance_id}`}; +function fixture() { + return {container_id: scope.container_id, started_at: scope.started_at, running: true, paused: true, + owned_barrier: true, control: "continue", source_control: "continue", ready_present: false, launcher_target: scope.old_target, + launcher_uid: 0, launcher_gid: 0, launcher_mode: 0o777, + probe_status: "ready", unknown_process_ids: [] as number[], + scheduler: {host_pending: false, host_attempt: null, last_wake_status: "wake_ok", last_wake_rc: 0}, + settlement: {identity: {result: {value: identity, failure: null}}, progress: {state: "settled"}}, + native: {execution_authority_granted: false, model_calls: 0, session_id: scope.session_id, + turn_ids: ["native-original"], completed_turn_id: "native-original", visible_history_sha256: "e".repeat(64), + quiescent_readback: true}, source_witness_before: "f".repeat(64), source_witness_after: "f".repeat(64)}; +} +test("only an owned pause with the same settled identity and stable native idle source qualifies", () => { + const result = decide({operation: "qualify", scope, observation: fixture()}); + assert.equal(result.qualified, true); + assert.equal(result.source_change_allowed, true); + assert.equal(result.resume_allowed, false); + assert.equal(result.execution_authority_granted, false); + assert.deepEqual(result.original_settlement_identity, {goal_id: scope.goal_id, agent_id: scope.agent_id, + todo_id: scope.todo_id, turn_instance_id: scope.turn_instance_id, replan_obligation_id: null, + binding_kind: "todo", binding_id: scope.todo_id, effect_id: identity.effect_id}); +}); +for (const [name, mutation] of [ + ["Stop", {control: "stop"}], ["unknown control", {control: "unknown"}], + ["typed source Stop", {source_stopped: true}], ["unpaused source", {paused: false}], + ["missing Goal-control proof", {source_control: undefined}], ["unknown Goal-control proof", {source_control: "unknown"}], + ["unowned barrier", {owned_barrier: false}], ["other namespace", {started_at: "replacement-start"}], + ["ready checkpoint", {ready_present: true}], ["unknown ready", {ready_present: null}], + ["unknown processes", {unknown_process_ids: [19]}], ["failed probe", {probe_status: "unavailable"}], + ["unfinished Host", {scheduler: {...fixture().scheduler, host_pending: true}}], + ["unresolved attempt", {scheduler: {...fixture().scheduler, host_attempt: {status: "started"}}}], + ["failed wake", {scheduler: {...fixture().scheduler, last_wake_rc: 1}}], + ["unsettled Turn", {settlement: {...fixture().settlement, progress: {state: "spend_receipt_required"}}}], + ["other Turn", {settlement: {identity: {result: {failure: null, value: {...identity, turn_instance_id: "another-turn", + effect_id: `${scope.goal_id}:${scope.agent_id}:${scope.todo_id}:another-turn`}}}, progress: {state: "settled"}}}], + ["native model request", {native: {...fixture().native, model_calls: 1}}], + ["no native idle proof", {native: {...fixture().native, quiescent_readback: false}}], + ["unfinished native cursor", {native: {...fixture().native, completed_turn_id: "old-cursor"}}], + ["other session", {native: {...fixture().native, session_id: "another-session"}}], + ["moving source", {source_witness_after: "a".repeat(64)}], + ["launcher CAS mismatch", {launcher_target: scope.new_target}], + ["launcher owner mismatch", {launcher_uid: 23}], +] as const) { + test(`${name} cannot authorize a launcher write`, () => { + const result = decide({operation: "qualify", scope, observation: {...fixture(), ...mutation}}); + assert.equal(result.source_change_allowed, false); + assert.equal(result.resume_allowed, false); + }); +} +test("preflight permits only a candidate; the paused native and settlement checks remain mandatory", () => { + const observed = {...fixture(), paused: false, owned_barrier: false}; + const result = decide({operation: "preflight", scope, observation: observed}); + assert.equal(result.pause_allowed, true); + assert.equal(result.source_change_allowed, false); + assert.equal(decide({operation: "qualify", scope, observation: observed}).qualified, false); + assert.equal(decide({operation: "preflight", scope, observation: {...observed, ready_present: true}}).pause_allowed, false); +}); +test("readback reconciles exactly old or qualified new; partial and unqualified new remain held", () => { + for (const [target, qualified, allowed] of [[scope.old_target, false, true], [scope.new_target, true, true], + [scope.new_target, false, false], [null, true, false], ["/profile/releases/partial", true, false]] as const) { + const result = decide({operation: "readback", scope, observation: {...fixture(), launcher_target: target, + qualified_source: qualified, unchanged_source_witness: true}}); + assert.equal(result.resume_allowed, allowed); + assert.equal(result.source_change_allowed, false); + } +}); +test("explicit transport repair never overrides changed work or Stop", () => { + const observed = {...fixture(), repair_target: "old", qualified_source: true, unchanged_source_witness: true}; + assert.equal(decide({operation: "repair", scope, observation: observed}).source_change_allowed, true); + for (const changed of [{control: "stop"}, {unchanged_source_witness: false}, {qualified_source: false}]) { + assert.equal(decide({operation: "repair", scope, observation: {...observed, ...changed}}).source_change_allowed, false); + } +}); +test("malformed or nonversioned scopes are rejected, not interpreted as permission", () => { + for (const changed of [{new_target: "/profile/releases/not-a-launcher"}, {new_target: scope.old_target}, + {launcher: scope.old_target}, {target_archive_sha256: "invalid"}, {container_id: "short"}, {extra: true}]) { + assert.throws(() => decide({operation: "qualify", scope: {...scope, ...changed}, observation: fixture()})); + } +});