diff --git a/docs/reference/automation-prompt-upgrades.md b/docs/reference/automation-prompt-upgrades.md new file mode 100644 index 0000000000..22c7808c73 --- /dev/null +++ b/docs/reference/automation-prompt-upgrades.md @@ -0,0 +1,106 @@ +# Managed Codex heartbeat prompt upgrades + +## Product contract + +Default execution remains `heartbeat-prompt --thin`. An adopted automation +stores a small, stable bootstrap instead of a copy of the current execution +rules. At **every wake**, it requests the complete JSON thin prompt from the +installed LoopX, with an explicit registry, Goal, agent and host capability +binding, then follows that prompt. After a normal LoopX upgrade, the next wake +reads the new rules; no per-release Codex database rewrite is needed. An already +running wake keeps its current instructions. This is model-mediated loading, +not a guarantee that a model will obey every instruction. + +The bootstrap is a transport wrapper, not a new execution mode. It does not +default to full/compact prompts, contain a parallel work policy, or create +another scheduler. Failure to load a complete successful response stops work +and spending; it must not fall back to remembered rules. Project watches and +business policy belong in LoopX state, not in this bootstrap. + +For a one-agent trial, pass `--cli-bin loopx-canary` to preview and apply. The +bootstrap and the thin prompt's generated commands both use that executable; +other automations continue to use their existing runtime. Do not promote the +canary as the global default merely to test one task. + +The owner is the existing heartbeat/upgrade boundary; there is no new optional +capability or extension provider. SQLite/TOML handling is a local host adapter, +not Todo, quota or scheduler authority. + +## Preview and adopt existing tasks + +```sh +loopx --format json automation-prompts plan --plan-file ./private-prompt-plan.json +``` + +The plan contains current and proposed prompts and is **private local data**. +Do not commit, upload, or paste it into public issues. The file is owner-only. +Use repeatable `--automation-id` selectors to narrow the plan. Discovery uses +registered Goal/agent identities only to propose candidates; it never grants +permission to overwrite them. Unrelated, ambiguous and inconsistent host +records are not adopted. Review custom instructions before replacement: move +durable project policy to its LoopX owner, or leave that task unadopted. There +is no automatic prose merge or substring-based permission to replace a prompt. + +While the Codex App is running, prefer its `automation_update` interface: +view each selected task, verify its current prompt against the preview, update +only the prompt to `desired_prompt` while preserving every other field, then +read it back. Do not recreate the task or rebind its thread. A single user +request can authorize this reviewed batch; the plan itself does not execute it. + +When that API is unavailable, the qualified macOS offline fallback is: + +```sh +# Review the plan, then close the Codex App first. +loopx automation-prompts apply --plan-file ./private-prompt-plan.json --offline --execute +``` + +The same registry/runtime-root and Codex home must be used for preview and +apply. `--codex-home` explicitly selects one home; the command never searches +other homes or copies sessions between them. No new automation is created. +Schedule, pause state, model, notification preferences, thread binding and run +history are preserved. Each selected task commits independently; the command +reports partial failure instead of claiming the whole batch succeeded. + +Only existing heartbeat records with matching SQLite/TOML identity, prompt, +status and thread binding are eligible. Unknown schemas and stale previews fail +closed. The fallback checks that the macOS App is closed; keep it closed until +readback completes. Restart afterward. This adapter targets the observed local +schema, **not an official stable Codex storage API**; no Windows/cloud support +is claimed. Use the native API if the host changes its storage contract. + +## Recovery and rollback + +The fallback stores a private per-task journal before writing. SQLite commit +and TOML replacement are not one transaction: a crash can leave a mirror pending. +While the App remains closed, recover that exact task: + +```sh +loopx automation-prompts recover --automation-id TASK_ID --offline --execute +# Or restore the previous prompt: +loopx automation-prompts rollback --automation-id TASK_ID --offline --execute +``` + +Recovery is idempotent. It refuses to overwrite later prompt/metadata edits. +The journal remains under the selected Codex home's `loopx-automation-backups` +directory; do not publish it. Rollback restores only the recorded prompt, +never an entire historical database or session table. Native API migrations +should retain the reviewed previous prompt privately and use that same API +for rollback. + +Disable automatic rule adoption by replacing the bootstrap with an explicitly +pinned prompt using the App, or pause the task there. LoopX runtime rollback +also changes the rules loaded on the next wake. Future incompatible bootstrap +revisions still require an explicit migration; the v1 wrapper does not silently +rewrite itself. `upgrade-plan` recognizes exact v1 wrappers as runtime-loaded +thin prompts, rather than repeatedly reporting their body as stale. + +## 中文摘要 + +默认仍是 thin。一次迁移后，automation 每次唤醒读取已安装 LoopX 的最新 +thin 指令；之后升级 LoopX 即可让下一轮采用新版规则，无需逐版本改 SQLite。 +正在运行的轮次不热切换。启动器不是另一套执行规则，也不增加权限。 + +先批量预览，再明确接管；自定义内容不猜测合并，不自动删除。App 运行时用 +原生更新接口；离线兼容通道要求关闭 App、精确预览校验、双存储读回，并保留 +私有恢复记录。日程、暂停状态、模型、线程、通知偏好和历史均不迁移。 +本批提供命令行批量迁移，不新增 Dashboard 按钮；也不保证模型行为已通过在线评测。 diff --git a/loopx/cli_commands/automation_prompts.py b/loopx/cli_commands/automation_prompts.py new file mode 100644 index 0000000000..68a5668398 --- /dev/null +++ b/loopx/cli_commands/automation_prompts.py @@ -0,0 +1,88 @@ +"""One reviewed migration; subsequent runtime upgrades need no host-store write.""" +from __future__ import annotations + +import argparse +import json +import sqlite3 +from pathlib import Path +import subprocess +import sys + +from loopx.control_plane.heartbeat.automation_upgrade import ( + SCHEMA, _atomic, apply_offline, build_plan, recover_offline, +) +from loopx.upgrade import codex_home + + +def register_automation_prompts(subparsers, add_format) -> None: + parser = subparsers.add_parser("automation-prompts", help="Preview and migrate existing Codex heartbeats to live LoopX rules.") + add_format(parser) + parser.add_argument("action", choices=("plan", "apply", "recover", "rollback")) + parser.add_argument("--codex-home", type=Path, help="One explicit host home; never discovers or migrates other homes.") + parser.add_argument("--plan-file", type=Path, help="Private reviewed plan file; plan saves it, apply reads it.") + parser.add_argument("--automation-id", action="append", default=[]) + parser.add_argument("--cli-bin", default="loopx", help="Stable installed executable; choose a canary binary for a single-task trial.") + parser.add_argument("--execute", action="store_true") + parser.add_argument("--offline", action="store_true", help="Acknowledge the Codex App is closed; use its automation API while running.") + + +def _require_offline() -> None: + if sys.platform != "darwin": + raise ValueError("offline adapter is qualified only on macOS; use the App automation API") + for name in ("Codex", "ChatGPT"): + observed = subprocess.run(["/usr/bin/pgrep", "-x", name], capture_output=True, check=False) + if observed.returncode != 1: + raise ValueError("close the Codex/ChatGPT App before offline migration; otherwise use automation_update") + + +def run(args: argparse.Namespace, registry: Path) -> dict: + home = (args.codex_home or codex_home()).expanduser().resolve() + if args.action == "plan": + if args.execute: + raise ValueError("plan cannot execute") + payload = build_plan(registry=registry, home=home, runtime_root=args.runtime_root, cli_bin=args.cli_bin) + if args.automation_id: + payload["entries"] = [entry for entry in payload["entries"] if entry["automation_id"] in args.automation_id] + if args.plan_file: + _atomic(args.plan_file.expanduser(), json.dumps(payload, ensure_ascii=False, indent=2)) + return payload + if not args.execute or not args.offline: + raise ValueError("use the App API, or close the App and explicitly pass --offline --execute") + _require_offline() + if args.action in ("recover", "rollback"): + if not args.automation_id: + raise ValueError("recovery requires explicit --automation-id") + return {"ok": True, "results": [recover_offline(home=home, automation_id=identifier, + rollback=args.action == "rollback") for identifier in args.automation_id]} + if not args.plan_file: + raise ValueError("apply requires a reviewed --plan-file") + plan = json.loads(args.plan_file.expanduser().read_text(encoding="utf-8")) + if plan.get("schema_version") != SCHEMA or plan.get("codex_home") != str(home): + raise ValueError("plan schema or host-home mismatch") + # Regenerate using current registry/profile inputs: a saved plan is not a + # license to install stale rules or execute arbitrary prompt text. + current = {entry["automation_id"]: entry for entry in build_plan( + registry=registry, home=home, runtime_root=args.runtime_root, cli_bin=args.cli_bin)["entries"]} + results = [] + for entry in plan["entries"]: + identifier = entry["automation_id"] + if args.automation_id and identifier not in args.automation_id: + continue + if entry["status"] != "adoption_required": + continue + now = current.get(identifier) + if not now or any(now.get(key) != entry.get(key) for key in + ("prompt_sha256", "source_sha256", "desired_prompt", "target_thread_id")): + results.append({"automation_id": identifier, "ok": False, "status": "preview_stale"}) + continue + try: + results.append(apply_offline(home=home, automation_id=identifier, + expected_prompt_sha256=entry["prompt_sha256"], desired_prompt=entry["desired_prompt"])) + except (ValueError, OSError, sqlite3.Error) as error: + results.append({"automation_id": identifier, "ok": False, "status": "failed", "reason": str(error)}) + return {"ok": all(item["ok"] for item in results), "results": results, + "scope": "per-automation commit; other automations and sessions unchanged"} + + +def render(payload: dict) -> str: + return "# Automation prompt upgrade\n\n```json\n" + json.dumps(payload, ensure_ascii=False, indent=2) + "\n```" diff --git a/loopx/cli_commands/support_control.py b/loopx/cli_commands/support_control.py index 9eac156399..9731a4360e 100644 --- a/loopx/cli_commands/support_control.py +++ b/loopx/cli_commands/support_control.py @@ -88,6 +88,7 @@ AddFormat = Callable[[argparse.ArgumentParser], None] SUPPORT_CONTROL_COMMANDS = { + "automation-prompts", "backup-state", "chat", "chat-endpoint", @@ -108,6 +109,8 @@ def register_support_control_commands( subparsers: argparse._SubParsersAction, add_subcommand_format: AddFormat, ) -> None: + from .automation_prompts import register_automation_prompts + register_automation_prompts(subparsers, add_subcommand_format) register_backup_state_command(subparsers, add_subcommand_format) register_heartbeat_control_commands(subparsers, add_subcommand_format) @@ -501,6 +504,15 @@ def handle_support_control_command( if args.command not in SUPPORT_CONTROL_COMMANDS: return None + if args.command == "automation-prompts": + from .automation_prompts import run, render + try: + payload = run(args, registry_path) + except Exception as error: + payload = {"ok": False, "error": str(error)} + print_payload(payload, output_format(args), render) + return 0 if payload.get("ok") else 1 + if args.command == "chat-endpoint": return handle_chat_endpoint_command( args, diff --git a/loopx/control_plane/heartbeat/automation_upgrade.py b/loopx/control_plane/heartbeat/automation_upgrade.py new file mode 100644 index 0000000000..460df5123b --- /dev/null +++ b/loopx/control_plane/heartbeat/automation_upgrade.py @@ -0,0 +1,272 @@ +"""Installed prompt lifecycle; execution policy stays in heartbeat-prompt. + +The App API is the preferred writer. The SQLite writer is an explicit offline +compatibility adapter, not a public Codex API or a scheduler implementation. +""" +from __future__ import annotations + +import hashlib +from contextlib import closing +import json +import os +from pathlib import Path +import re +import shlex +import sqlite3 +import tempfile +import tomllib +from typing import Any + +from loopx.upgrade import ( + codex_home, infer_agent_id_from_prompt, infer_goal_id_from_prompt, + infer_available_capabilities_from_prompt, +) + +SCHEMA = "loopx_automation_prompt_upgrade_v0" +BOOTSTRAP = "LoopX managed heartbeat bootstrap v1" + + +def digest(value: str) -> str: + return hashlib.sha256(value.encode()).hexdigest() + + +def bootstrap_prompt(*, registry: Path, goal_id: str, agent_id: str, + runtime_root: str | None = None, + capabilities: list[str] | None = None, + cli_bin: str = "loopx") -> str: + args = [cli_bin, "--format", "json", "--registry", str(registry.resolve())] + if runtime_root: + args += ["--runtime-root", str(Path(runtime_root).expanduser().resolve())] + args += ["heartbeat-prompt", "--thin", "--codex-app", "--goal-id", goal_id, + "--agent-id", agent_id] + if cli_bin != "loopx": + args += ["--cli-bin", cli_bin] + for capability in capabilities or []: + args += ["--available-capability", capability] + return ( + f"{BOOTSTRAP}\n" + "每次唤醒先执行：\n" + f"```sh\n{shlex.join(args)}\n```\n" + "读取完整结果；仅 ok=true 时按本次 task_body 执行，不复用旧指令；" + "失败或结果不完整则停止并报告，不执行任务或记账。" + ) + + +def _atomic(path: Path, text: str) -> None: + path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + fd, temporary = tempfile.mkstemp(dir=path.parent, prefix=".loopx-") + try: + with os.fdopen(fd, "w", encoding="utf-8") as stream: + stream.write(text) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, path) + directory = os.open(path.parent, os.O_RDONLY) + try: + os.fsync(directory) + finally: + os.close(directory) + finally: + if os.path.exists(temporary): + os.unlink(temporary) + + +def bootstrap_binding(prompt: str) -> dict | None: + if not prompt.startswith(BOOTSTRAP + "\n"): + return None + try: + command = prompt.split("```sh\n", 1)[1].split("\n```", 1)[0] + tokens = shlex.split(command) + if tokens[1:3] != ["--format", "json"]: + return None + values: dict[str, Any] = {"capabilities": []} + flags = {"--registry": "registry", "--runtime-root": "runtime_root", + "--goal-id": "goal_id", "--agent-id": "agent_id", "--cli-bin": "cli_bin"} + index = 3 + while index < len(tokens): + token = tokens[index] + if token in ("heartbeat-prompt", "--thin", "--codex-app"): + index += 1 + continue + if token == "--available-capability": + values["capabilities"].append(tokens[index + 1]) + elif token in flags and flags[token] not in values: + values[flags[token]] = tokens[index + 1] + else: + return None + index += 2 + values["registry"] = Path(values["registry"]) + if tokens[0] != values.get("cli_bin", "loopx"): + return None + return values if bootstrap_prompt(**values) == prompt else None + except (IndexError, KeyError, TypeError, ValueError): + return None + + +def _replace_prompt(source: str, prompt: str) -> str: + """Retain unknown TOML fields/comments; prove only prompt changed by parsing. + + Candidate spans are not trusted lexically: multiline strings can contain + fake assignments. Both prefix and full-document semantic checks must pass. + """ + original = tomllib.loads(source) + desired = {**original, "prompt": prompt} + lines = source.splitlines(keepends=True) + for start, line in enumerate(lines): + if not re.match(r"^prompt\s*=", line): + continue + try: + prefix = tomllib.loads("".join(lines[:start])) + except tomllib.TOMLDecodeError: + continue + if "prompt" in prefix: + continue + for end in range(start + 1, len(lines) + 1): + candidate = "".join(lines[:start]) + "prompt = " + json.dumps(prompt, ensure_ascii=False) + "\n" + "".join(lines[end:]) + try: + if tomllib.loads(candidate) == desired: + return candidate + except tomllib.TOMLDecodeError: + pass + raise ValueError("unsupported automation TOML prompt encoding; use the App API") + + +def _connect(home: Path, *, writable: bool = False) -> sqlite3.Connection: + path = home / "sqlite/codex-dev.db" + connection = sqlite3.connect(path.as_uri() + ("?mode=rw" if writable else "?mode=ro"), uri=True, timeout=5) + connection.row_factory = sqlite3.Row + columns = {row[1] for row in connection.execute("PRAGMA table_info(automations)")} + if not {"id", "kind", "prompt", "status", "target_thread_id"} <= columns: + connection.close() + raise ValueError("unsupported Codex automation schema; use the App API") + return connection + + +def _read(home: Path, automation_id: str, connection: sqlite3.Connection) -> tuple[str, dict, dict]: + if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,159}", automation_id): + raise ValueError("invalid automation id") + path = home / "automations" / automation_id / "automation.toml" + if path.is_symlink() or path.parent.is_symlink(): + raise ValueError("symlink automation stores are not supported") + source = path.read_text(encoding="utf-8") + if len(source.encode("utf-8")) > 256_000: + raise ValueError("automation manifest exceeds the supported offline size") + item = tomllib.loads(source) + row = connection.execute("SELECT * FROM automations WHERE id=?", (automation_id,)).fetchone() + if row is None or item.get("id") != automation_id: + raise ValueError("automation identity missing or mismatched") + row = dict(row) + if item.get("kind") != "heartbeat" or row["kind"] != "heartbeat": + raise ValueError("only existing heartbeat automations are supported") + if item.get("status") == "DELETED" or row["status"] == "DELETED": + raise ValueError("deleted automation cannot be upgraded") + for key in ("prompt", "status", "target_thread_id"): + if item.get(key) != row.get(key): + raise ValueError(f"automation stores disagree on {key}; reconcile through the App") + return source, item, row + + +def build_plan(*, registry: Path, home: Path | None = None, + runtime_root: str | None = None, cli_bin: str = "loopx") -> dict[str, Any]: + from loopx.agent_registry import registered_agent_ids_for_goal + from loopx.history import load_registry + from loopx.registry import registry_goals + + home = (home or codex_home()).expanduser().resolve() + goals = {str(goal["id"]): goal for goal in registry_goals(load_registry(registry))} + entries = [] + with closing(_connect(home)) as connection: + for path in sorted((home / "automations").glob("*/automation.toml")): + entry: dict[str, Any] = {"automation_id": path.parent.name} + try: + source, item, row = _read(home, path.parent.name, connection) + prompt = item["prompt"] + goal_id = infer_goal_id_from_prompt(prompt) + agent_id = infer_agent_id_from_prompt(prompt) + goal_mentions = set(re.findall(r"--goal-id\s+([A-Za-z0-9_.:-]+)", prompt)) + agent_mentions = set(re.findall(r"--agent-id\s+([A-Za-z0-9_.:-]+)", prompt)) + if (goal_mentions - {goal_id}) or (agent_mentions - {agent_id}): + raise ValueError("ambiguous Goal/agent bindings; select and migrate through the App") + if goal_id not in goals or agent_id not in registered_agent_ids_for_goal(goals[goal_id]): + entry.update(status="unmanaged", reason="no unique registered Goal/agent binding") + else: + desired = bootstrap_prompt(registry=registry, goal_id=goal_id, agent_id=agent_id, + runtime_root=runtime_root, capabilities=infer_available_capabilities_from_prompt(prompt), + cli_bin=cli_bin) + entry.update(status="current" if prompt == desired else "adoption_required", + goal_id=goal_id, agent_id=agent_id, prompt_sha256=digest(prompt), + current_prompt=prompt, + source_sha256=digest(source), desired_prompt=desired, + desired_sha256=digest(desired), target_thread_id=row["target_thread_id"]) + except (ValueError, OSError) as error: + entry.update(status="blocked", reason=str(error)) + entries.append(entry) + return {"schema_version": SCHEMA, "ok": True, "codex_home": str(home), + "entries": entries, "writes": False, + "policy": "Discovery is not adoption authority. Review each replacement; use the App API first."} + + +def apply_offline(*, home: Path, automation_id: str, expected_prompt_sha256: str, + desired_prompt: str) -> dict[str, Any]: + """Explicit offline fallback. Persist recovery before either host-store write. + + Scheduler/thread/history tables are never touched. The caller must close + the App: its external TOML writes cannot join this SQLite transaction. + """ + home = home.expanduser().resolve() + journal = home / "loopx-automation-backups" / (automation_id + ".json") + if journal.is_symlink() or journal.parent.is_symlink(): + raise ValueError("symlink backup stores are not supported") + with closing(_connect(home, writable=True)) as connection: + connection.execute("BEGIN IMMEDIATE") + source, item, row = _read(home, automation_id, connection) + if digest(item["prompt"]) != expected_prompt_sha256: + raise ValueError("prompt changed after preview; no writes performed") + if item["prompt"] == desired_prompt: + return {"ok": True, "status": "current", "automation_id": automation_id} + if journal.exists(): + raise ValueError("previous migration journal exists; recover or rollback it first") + replacement = _replace_prompt(source, desired_prompt) + # Entire originals stay private for recovery; no raw prompts in receipts. + _atomic(journal, json.dumps({"schema_version": SCHEMA, "automation_id": automation_id, + "before": source, "after": replacement, "row": row}, ensure_ascii=False)) + connection.execute("UPDATE automations SET prompt=? WHERE id=? AND prompt=?", + (desired_prompt, automation_id, item["prompt"])) + connection.commit() + _atomic(home / "automations" / automation_id / "automation.toml", replacement) + with closing(_connect(home)) as connection: + _read(home, automation_id, connection) + return {"ok": True, "status": "updated", "automation_id": automation_id, + "backup": str(journal), "future_policy": "read installed heartbeat-prompt on every wake"} + + +def recover_offline(*, home: Path, automation_id: str, rollback: bool = False) -> dict: + home = home.expanduser().resolve() + if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,159}", automation_id): + raise ValueError("invalid automation id") + journal = home / "loopx-automation-backups" / (automation_id + ".json") + if journal.is_symlink() or journal.parent.is_symlink(): + raise ValueError("symlink backup stores are not supported") + saved = json.loads(journal.read_text(encoding="utf-8")) + if saved.get("schema_version") != SCHEMA or saved.get("automation_id") != automation_id: + raise ValueError("invalid migration journal") + before, after = saved["before"], saved["after"] + prompts = [tomllib.loads(text)["prompt"] for text in (before, after)] + path = home / "automations" / automation_id / "automation.toml" + if path.is_symlink() or path.parent.is_symlink(): + raise ValueError("symlink automation stores are not supported") + with closing(_connect(home, writable=True)) as connection: + connection.execute("BEGIN IMMEDIATE") + row = connection.execute("SELECT * FROM automations WHERE id=?", (automation_id,)).fetchone() + if row is None or row["prompt"] not in prompts or path.read_text(encoding="utf-8") not in (before, after): + raise ValueError("automation changed since migration; recovery refuses to overwrite it") + original = saved["row"] + if any(row[key] != value for key, value in original.items() if key != "prompt"): + raise ValueError("automation metadata changed; reconcile through the App") + selected = before if rollback else after + connection.execute("UPDATE automations SET prompt=? WHERE id=?", (tomllib.loads(selected)["prompt"], automation_id)) + connection.commit() + _atomic(path, selected) + with closing(_connect(home)) as connection: + _read(home, automation_id, connection) + return {"ok": True, "status": "rolled_back" if rollback else "recovered", "automation_id": automation_id} diff --git a/loopx/upgrade.py b/loopx/upgrade.py index 98f6cb549d..d60d392eda 100644 --- a/loopx/upgrade.py +++ b/loopx/upgrade.py @@ -225,6 +225,7 @@ def infer_available_capabilities_from_prompt(prompt: str) -> list[str]: def load_codex_app_automation_manifest(root: Path | None = None) -> dict[str, Any]: + from .control_plane.heartbeat.automation_upgrade import bootstrap_binding home = root or codex_home() automations_root = home / "automations" if not automations_root.exists(): @@ -273,6 +274,7 @@ def load_codex_app_automation_manifest(root: Path | None = None) -> dict[str, An continue agent_id = infer_agent_id_from_prompt(prompt) status = str(automation.get("status") or "ACTIVE") + binding = bootstrap_binding(prompt) entries.append( { "automation_id": str(automation.get("id") or path.parent.name), @@ -293,6 +295,9 @@ def load_codex_app_automation_manifest(root: Path | None = None) -> dict[str, An "status": status, "installed": status.upper() != "DELETED", "source": "codex_app_automation_toml", + "runtime_thin_bootstrap": { + **binding, "registry": str(binding["registry"]), + } if binding else None, "path": str(path), } ) @@ -755,11 +760,21 @@ def build_upgrade_plan( expected_digest = str(summary.get("sha256") or "") not_installed = entry_declares_not_installed(entry) actual_digest = None if not_installed else installed_entry_digest(entry) if entry else None + bootstrap = entry.get("runtime_thin_bootstrap") if entry else None + live_thin = bool( + isinstance(bootstrap, dict) + and mode == "thin" + and bootstrap.get("goal_id") == goal_id + and bootstrap.get("agent_id") == agent_id + and bootstrap.get("cli_bin", "loopx") == cli_bin + and bootstrap.get("runtime_root") == (str(Path(runtime_root_override).expanduser().resolve()) if runtime_root_override else None) + and Path(str(bootstrap.get("registry"))).resolve() == Path(registry_path).resolve() + ) status = "unknown" if not_installed: status = "not_installed" elif entry: - status = "current" if actual_digest == expected_digest else "stale" + status = "current" if live_thin or actual_digest == expected_digest else "stale" policy_audit = ( { "available": False, diff --git a/tests/control_plane/test_automation_prompt_upgrade.py b/tests/control_plane/test_automation_prompt_upgrade.py new file mode 100644 index 0000000000..150a75a568 --- /dev/null +++ b/tests/control_plane/test_automation_prompt_upgrade.py @@ -0,0 +1,226 @@ +from __future__ import annotations + +import json +from pathlib import Path +import shlex +import sqlite3 +import subprocess +import sys +import tomllib + +import pytest + +from loopx.control_plane.heartbeat import automation_upgrade as upgrade + + +def fixture(tmp_path: Path): + home = tmp_path / "host" + path = home / "automations/watch/automation.toml" + path.parent.mkdir(parents=True) + prompt = "Advance `fixture-goal` from registry. --agent-id agent-a" + path.write_text('version = 1\nid = "watch"\nkind = "heartbeat"\n' + 'status = "PAUSED"\ntarget_thread_id = "thread-a"\n' + 'rrule = "FREQ=HOURLY"\nnotification_policy = "failed_runs_only"\n' + '# retain custom metadata\n[unused]\nvalue = 1\n', encoding="utf-8") + path.write_text('prompt = ' + json.dumps(prompt) + '\n' + path.read_text(), encoding="utf-8") + database = home / "sqlite/codex-dev.db" + database.parent.mkdir() + with sqlite3.connect(database) as connection: + connection.execute("CREATE TABLE automations (id TEXT PRIMARY KEY, kind TEXT, prompt TEXT, status TEXT, target_thread_id TEXT, rrule TEXT, model TEXT, updated_at INTEGER, next_run_at INTEGER)") + connection.execute("INSERT INTO automations VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + ("watch", "heartbeat", prompt, "PAUSED", "thread-a", "FREQ=HOURLY", "fixture-model", 123, 456)) + connection.execute("CREATE TABLE sessions (id TEXT)") + connection.execute("INSERT INTO sessions VALUES ('do-not-touch')") + registry = tmp_path / "registry.json" + state = tmp_path / "STATE.md" + state.write_text("# Fixture\n", encoding="utf-8") + registry.write_text(json.dumps({"goals": [{"id": "fixture-goal", "repo": str(tmp_path), + "state_file": str(state), "registered_agents": ["agent-a"]}]}), encoding="utf-8") + return home, path, database, registry, prompt + + +def test_real_sqlite_upgrade_preserves_schedule_binding_model_and_history(tmp_path): + home, path, database, registry, prompt = fixture(tmp_path) + original = path.read_text() + plan = upgrade.build_plan(registry=registry, home=home) + item = plan["entries"][0] + assert item["status"] == "adoption_required" + assert path.read_text() == original + with sqlite3.connect(database) as connection: + before = connection.execute("SELECT * FROM automations").fetchone() + result = upgrade.apply_offline(home=home, automation_id="watch", + expected_prompt_sha256=upgrade.digest(prompt), desired_prompt=item["desired_prompt"]) + assert result["status"] == "updated" + assert "# retain custom metadata" in path.read_text() + with sqlite3.connect(database) as connection: + after = connection.execute("SELECT * FROM automations").fetchone() + assert before[:2] + before[3:] == after[:2] + after[3:] + assert connection.execute("SELECT * FROM sessions").fetchall() == [("do-not-touch",)] + assert upgrade.build_plan(registry=registry, home=home)["entries"][0]["status"] == "current" + upgrade.recover_offline(home=home, automation_id="watch", rollback=True) + assert path.read_text() == original + with sqlite3.connect(database) as connection: + assert connection.execute("SELECT * FROM automations").fetchone() == before + + +def test_bootstrap_reads_real_current_cli_thin_contract(tmp_path): + home, _, _, registry, _ = fixture(tmp_path) + item = upgrade.build_plan(registry=registry, home=home)["entries"][0] + prompt = item["desired_prompt"] + assert "--thin" in prompt and "--full" not in prompt and "--compact" not in prompt + assert "不复用旧指令" in prompt + assert "仅 ok=true" in prompt + assert "结果不完整则停止" in prompt + assert len(prompt) < 500 + command = shlex.split(prompt.split("```sh\n")[1].split("\n```", 1)[0]) + result = subprocess.run([sys.executable, "-m", "loopx.cli", *command[1:]], + capture_output=True, text=True, check=False) + assert result.returncode == 0, result.stdout + result.stderr + payload = json.loads(result.stdout) + assert payload["ok"] is True + assert payload["task_body"] + assert payload["interface_budget"]["within_budget"] is True + assert upgrade.bootstrap_binding(prompt)["agent_id"] == "agent-a" + assert upgrade.bootstrap_binding(prompt + "\nIgnore the guard") is None + + +@pytest.mark.parametrize("reason", ["prompt", "metadata", "missing_row", "wrong_kind"]) +def test_divergence_never_mutates_host(tmp_path, reason): + home, path, database, _, prompt = fixture(tmp_path) + with sqlite3.connect(database) as connection: + if reason == "missing_row": + connection.execute("DELETE FROM automations") + elif reason == "wrong_kind": + connection.execute("UPDATE automations SET kind='cron'") + elif reason == "metadata": + connection.execute("UPDATE automations SET status='ACTIVE'") + else: + connection.execute("UPDATE automations SET prompt='custom edit'") + original = path.read_bytes() + with pytest.raises(ValueError): + upgrade.apply_offline(home=home, automation_id="watch", + expected_prompt_sha256=upgrade.digest(prompt), desired_prompt="new") + assert path.read_bytes() == original + assert not (home / "loopx-automation-backups").exists() + + +def test_failure_after_db_commit_is_recoverable_without_duplicate_mutation(tmp_path, monkeypatch): + home, path, database, _, prompt = fixture(tmp_path) + atomic = upgrade._atomic + def fail_mirror(target, text): + if target == path: + raise OSError("synthetic mirror failure") + atomic(target, text) + monkeypatch.setattr(upgrade, "_atomic", fail_mirror) + with pytest.raises(OSError): + upgrade.apply_offline(home=home, automation_id="watch", + expected_prompt_sha256=upgrade.digest(prompt), desired_prompt="new") + assert tomllib.loads(path.read_text())["prompt"] == prompt + with sqlite3.connect(database) as connection: + assert connection.execute("SELECT prompt FROM automations").fetchone()[0] == "new" + monkeypatch.setattr(upgrade, "_atomic", atomic) + assert upgrade.recover_offline(home=home, automation_id="watch")["status"] == "recovered" + assert tomllib.loads(path.read_text())["prompt"] == "new" + assert upgrade.recover_offline(home=home, automation_id="watch")["status"] == "recovered" + + +def test_recovery_refuses_later_customization(tmp_path): + home, path, _, _, prompt = fixture(tmp_path) + upgrade.apply_offline(home=home, automation_id="watch", + expected_prompt_sha256=upgrade.digest(prompt), desired_prompt="new") + path.write_text(path.read_text() + '\n# user edit\n') + with pytest.raises(ValueError, match="changed since migration"): + upgrade.recover_offline(home=home, automation_id="watch", rollback=True) + + +def test_toml_multiline_embedded_assignment_is_not_a_field(tmp_path): + source = 'name = "watch"\nprompt = """old\nprompt = \'fake\'\n"""\nrrule = "FREQ=HOURLY"\n' + updated = upgrade._replace_prompt(source, 'new "quotes"\nbody') + assert tomllib.loads(updated) == {**tomllib.loads(source), "prompt": 'new "quotes"\nbody'} + + +def test_cli_preview_private_file_and_no_implicit_apply(tmp_path): + home, path, _, registry, _ = fixture(tmp_path) + plan = tmp_path / "private-plan.json" + args = [sys.executable, "-m", "loopx.cli", "--format", "json", "--registry", str(registry), + "automation-prompts", "plan", "--codex-home", str(home), "--plan-file", str(plan)] + original = path.read_bytes() + result = subprocess.run(args, capture_output=True, text=True) + assert result.returncode == 0, result.stdout + result.stderr + assert plan.stat().st_mode & 0o077 == 0 + assert path.read_bytes() == original + args[args.index("plan")] = "apply" + result = subprocess.run(args, capture_output=True, text=True) + assert result.returncode == 1 + assert path.read_bytes() == original + + +def test_unsupported_schema_and_missing_database_fail_without_creating(tmp_path): + home = tmp_path / "host" + home.mkdir() + with pytest.raises(sqlite3.Error): + upgrade._connect(home) + assert list(home.iterdir()) == [] + + +def test_stale_preview_cas_and_cross_home_boundary(tmp_path): + home, path, _, _, _ = fixture(tmp_path) + before = path.read_bytes() + with pytest.raises(ValueError, match="changed after preview"): + upgrade.apply_offline(home=home, automation_id="watch", + expected_prompt_sha256="stale", desired_prompt="new") + assert path.read_bytes() == before + other = tmp_path / "other-home" + other.mkdir() + assert list(other.iterdir()) == [] + + +def test_upgrade_plan_recognizes_exact_live_thin_wrapper(tmp_path, monkeypatch): + from loopx.upgrade import build_upgrade_plan + home, _, _, registry, prompt = fixture(tmp_path) + desired = upgrade.bootstrap_prompt(registry=registry, goal_id="fixture-goal", agent_id="agent-a") + upgrade.apply_offline(home=home, automation_id="watch", + expected_prompt_sha256=upgrade.digest(prompt), desired_prompt=desired) + monkeypatch.setenv("CODEX_HOME", str(home)) + plan = build_upgrade_plan(registry_path=registry) + assert plan["summary"]["current_prompt_count"] == 1 + assert plan["summary"]["stale_prompt_count"] == 0 + + +def test_cli_offline_apply_checks_exact_saved_plan(tmp_path, monkeypatch): + from argparse import Namespace + from loopx.cli_commands import automation_prompts as cli + home, path, _, registry, _ = fixture(tmp_path) + plan = upgrade.build_plan(registry=registry, home=home) + saved = tmp_path / "plan.json" + saved.write_text(json.dumps(plan)) + monkeypatch.setattr(cli, "_require_offline", lambda: None) + args = Namespace(codex_home=home, action="apply", execute=True, offline=True, + plan_file=saved, automation_id=[], runtime_root=None, cli_bin="loopx") + assert cli.run(args, registry)["results"][0]["status"] == "updated" + assert upgrade.bootstrap_binding(tomllib.loads(path.read_text())["prompt"]) + # The old preview cannot replace a now-customized prompt or change homes. + assert cli.run(args, registry)["results"][0]["status"] == "preview_stale" + args.codex_home = tmp_path / "other-home" + with pytest.raises(ValueError, match="host-home mismatch"): + cli.run(args, registry) + + +def test_canary_keeps_generated_commands_on_the_same_runtime(tmp_path): + prompt = upgrade.bootstrap_prompt(registry=tmp_path / "registry.json", + goal_id="fixture-goal", agent_id="agent-a", cli_bin="loopx-canary") + binding = upgrade.bootstrap_binding(prompt) + assert binding["cli_bin"] == "loopx-canary" + assert "--cli-bin loopx-canary" in prompt + assert upgrade.bootstrap_binding(prompt.replace("--cli-bin loopx-canary", "--cli-bin loopx")) is None + + +def test_ambiguous_discovery_is_not_replacement_authority(tmp_path): + home, path, database, registry, prompt = fixture(tmp_path) + ambiguous = prompt + " --agent-id agent-b" + path.write_text(upgrade._replace_prompt(path.read_text(), ambiguous)) + with sqlite3.connect(database) as connection: + connection.execute("UPDATE automations SET prompt=?", (ambiguous,)) + entry = upgrade.build_plan(registry=registry, home=home)["entries"][0] + assert entry["status"] == "blocked" + assert "desired_prompt" not in entry