From fc8e3739f5e2856a1d8c8b859066fff962160553 Mon Sep 17 00:00:00 2001 From: huangruiteng <14976749+huangruiteng@users.noreply.github.com> Date: Tue, 15 Sep 2026 13:49:15 +0800 Subject: [PATCH] fix(heartbeat): observe the installed automation prompt binding every turn Installed automation bodies were reconciled once, at `update --apply` time. A running App cannot be written through the offline adapter, so the pending adoption was reported only inside that batch: a lane that was repaired, resumed, or activated later silently kept applying a frozen body, and nothing in the recurring contract ever re-observed the installed prompt. `quota should-run` now observes the caller's own installed automation and projects it as `scheduler_hint.app_automation.prompt_binding` (`codex_app_automation_prompt_binding_v0`: current, adoption_required, blocked, ambiguous, absent, unavailable). A stale entry carries `host_action=adopt_managed_bootstrap`, its no-spend policy, both prompt digests, and the same reviewed prompt-only `automation_update` request that update-time reconciliation returns, so the obligation survives in the live contract instead of a completed report. Update-time and turn-time requests are built by one renderer so their shapes cannot drift. The observation is read-only, bounded, and fail-open: an unreadable store reports a status instead of failing the turn, and a lane without an installed automation projects no field. A recognized loader is always compared with its own binding, because a turn must never retarget another home, registry, runtime root, or CLI binary; only an unrecognized body is reviewed against the caller's registry. The thin/compact/full scheduler-hint rules name the new host action within their existing output budgets. Signed-off-by: huangruiteng <14976749+huangruiteng@users.noreply.github.com> --- docs/heartbeat-automation-prompt.md | 13 +- docs/reference/automation-prompt-upgrades.md | 37 +++ .../control_plane/heartbeat-prompt-smoke.py | 2 + .../heartbeat/automation_upgrade.py | 232 +++++++++++++++--- .../heartbeat/installed_prompt_update.py | 20 +- loopx/control_plane/heartbeat/rules.py | 7 +- .../quota/app_automation_observation.py | 81 ++++++ .../control_plane/quota/should_run_packet.py | 52 ++-- .../control_plane/scheduler/scheduler_hint.py | 7 + skills/loopx-project/SKILL.md | 8 +- .../test_automation_prompt_upgrade.py | 203 +++++++++++++++ ...est_codex_app_prompt_binding_projection.py | 124 ++++++++++ 12 files changed, 704 insertions(+), 82 deletions(-) create mode 100644 loopx/control_plane/quota/app_automation_observation.py create mode 100644 tests/control_plane/test_codex_app_prompt_binding_projection.py diff --git a/docs/heartbeat-automation-prompt.md b/docs/heartbeat-automation-prompt.md index 463b7e64ec..f38760fe44 100644 --- a/docs/heartbeat-automation-prompt.md +++ b/docs/heartbeat-automation-prompt.md @@ -637,7 +637,13 @@ heartbeats should search/use `automation_update` when available. If `scheduler_hint.app_automation.host_action=pause_or_delete_current_heartbeat`: in that terminal case, call `automation_update` once to pause the current heartbeat (delete only if pause is unavailable), verify the host result, spend -no quota, and end the turn without a scheduler ACK. Otherwise call it only when +no quota, and end the turn without a scheduler ACK. When the same lane reports +`scheduler_hint.app_automation.prompt_binding.host_action=adopt_managed_bootstrap`, +its installed body is not the current managed loader: apply the reported +prompt-only `api_update_request` through `automation_update` once after checking +the reported prompt hash, read the automation back, and spend no quota. That +request preserves the binding, schedule and status and is independent of +`automation_update` cadence handling. Otherwise call it only when `scheduler_hint.app_automation.stateful_backoff.apply_needed=true` and `scheduler_hint.app_automation.recommended_rrule` is present. After a successful RRULE update, run `loopx` with @@ -815,7 +821,10 @@ automations can all share the same LoopX quota guard without hard-coding different wait loops. Host implementations should first honor a terminal `app_automation.host_action=pause_or_delete_current_heartbeat` by stopping the current heartbeat once, verifying the result, and ending without scheduler ACK -or quota spend. Otherwise they should read the compact +or quota spend, and they should honor +`app_automation.prompt_binding.host_action=adopt_managed_bootstrap` by applying +that reported prompt-only request once, verifying the readback, and spending no +quota. Otherwise they should read the compact `app_automation.stateful_backoff` packet, call `automation_update` only when `apply_needed=true`, and then let `quota scheduler-ack-current` persist the applied RRULE state from the latest scheduler hint without spending quota. A diff --git a/docs/reference/automation-prompt-upgrades.md b/docs/reference/automation-prompt-upgrades.md index 22d4edbab9..2c24cb2ad9 100644 --- a/docs/reference/automation-prompt-upgrades.md +++ b/docs/reference/automation-prompt-upgrades.md @@ -68,6 +68,31 @@ read-only preview, not the upgrade executor. Do not infer a manual-only policy from its `adoption_required` status. Custom or inconsistent entries still need review; automatic prompt migration never grants scheduler or thread authority. +## Live turn observation + +Install-time reconciliation is one-shot: a pending adoption that nobody applies +would otherwise never reappear, and a repaired or resumed lane can keep running +yesterday's frozen body. `quota should-run` therefore reports the caller's own +installed automation in `scheduler_hint.app_automation.prompt_binding` (schema +`codex_app_automation_prompt_binding_v0`: `current`, `adoption_required`, +`blocked`, `ambiguous`, `absent`, or `unavailable`). A stale entry adds +`host_action=adopt_managed_bootstrap`, its no-spend policy, both prompt digests, +and the same reviewed prompt-only `automation_update` request that update-time +reconciliation returns, so the obligation survives in the live contract instead +of a completed report. + +The observation is bounded, read-only, and fail-open: an unreadable or +disagreeing store reports a status and never fails the turn, and a lane without +an installed automation projects no field at all. Several installed automations +can claim one Goal/agent, so the observer follows the one bound to the current +thread and classifies only a body whose TOML and App records agree; unconfirmed +look-alikes are named only when nothing else is confirmable. A recognized loader +is always compared with its own binding, because a turn must never retarget another Codex +home, registry, runtime root, or CLI binary; only an unrecognized body is +reviewed against the caller's registry. Adoption stays explicit and +non-blocking: it is not delivery permission, not a scheduler authority, and a +deliberate owner-pinned body is reviewed rather than overwritten. + On the qualified macOS heartbeat schema, direct migration requires the App closed. The adapter holds a SQLite writer transaction through TOML delivery, compares the entire previewed manifest, preserves every non-prompt field, and @@ -202,3 +227,15 @@ gh 登录，仍失败则明确要求已核验 SHA，不切换分支或静默覆 日程、暂停状态、模型、线程、通知偏好和历史均不迁移。 不支持的存储仍需原生 API；运行中的本轮不热切换。普通测试不消耗模型 token， 真实模型发布资格仍需独立评测，不能由迁移成功推断。 + +安装期对账只报告一次，因此 `quota should-run` 每轮都把本轮 lane 已安装的 +automation 观测投影为 `scheduler_hint.app_automation.prompt_binding`：状态为 +`current`/`adoption_required`/`blocked`/`ambiguous`/`absent`/`unavailable`。 +非当前 loader 时附带 `host_action=adopt_managed_bootstrap`、no-spend 策略、 +两个 prompt 摘要，以及与升级期完全相同的、仅改 prompt 的 +`automation_update` 请求，使采纳义务留在实时契约里，而不只存在于一次性报告。 +该观测只读、有界、失败即降级，未安装 automation 的 lane 不投影该字段；已识别 +的 loader 一律按自身绑定比对，turn 不会改标到其他 home、registry、runtime root +或 CLI。同一 Goal/agent 可能装了多个 automation，因此观测跟随当前 thread 绑定的 +那一个，并且只对 TOML 与 App 记录一致的 body 分类；未确认的同名记录仅在没有可确认 +body 时才列出。采纳仍需显式执行，既不授予交付权限也不接管调度。 diff --git a/examples/control_plane/heartbeat-prompt-smoke.py b/examples/control_plane/heartbeat-prompt-smoke.py index 71167f43a2..dfe6d02678 100644 --- a/examples/control_plane/heartbeat-prompt-smoke.py +++ b/examples/control_plane/heartbeat-prompt-smoke.py @@ -594,6 +594,7 @@ def main() -> int: "具体user todo未投影", "Observed capabilities -> `--available-capability`; never user gates", "host_action=pause_or_delete_current_heartbeat->automation_update stop(no-spend)", + "prompt_binding=adopt(no-spend);", "else RRULE/projected-fallback_hint/ack/fail", "no-change=`surface_only`/no spend", "unchanged->`--vision-unchanged-reason`", @@ -696,6 +697,7 @@ def main() -> int: "NOTIFY缺动作→", "具体user todo未投影", "host_action=pause_or_delete_current_heartbeat->automation_update stop(no-spend)", + "prompt_binding=adopt(no-spend);", "else RRULE/projected-fallback_hint/ack/fail", "no-change=`surface_only`/no spend", "unchanged->`--vision-unchanged-reason`", diff --git a/loopx/control_plane/heartbeat/automation_upgrade.py b/loopx/control_plane/heartbeat/automation_upgrade.py index c1f8993fd6..d9b492a1ac 100644 --- a/loopx/control_plane/heartbeat/automation_upgrade.py +++ b/loopx/control_plane/heartbeat/automation_upgrade.py @@ -15,7 +15,7 @@ import sqlite3 import tempfile import tomllib -from typing import Any +from typing import Any, Mapping from types import SimpleNamespace from .bootstrap_prompt import ( @@ -28,10 +28,14 @@ from loopx.upgrade import ( codex_home, infer_agent_id_from_prompt, infer_goal_id_from_prompt, - infer_available_capabilities_from_prompt, + infer_available_capabilities_from_prompt, load_codex_app_automation_manifest, ) SCHEMA = "loopx_automation_prompt_upgrade_v0" +PROMPT_BINDING_SCHEMA_VERSION = "codex_app_automation_prompt_binding_v0" +PROMPT_BINDING_ADOPT_ACTION = "adopt_managed_bootstrap" +PROMPT_BINDING_HOST_ACTION_CONTRACT = "codex_app_automation_prompt_adoption" +PROMPT_BINDING_SPEND_POLICY = "no_spend_for_automation_prompt_adoption" BOOTSTRAP = HEARTBEAT_BOOTSTRAP _LEGACY_BOOTSTRAP = "LoopX managed heartbeat bootstrap v1" _LEGACY_INSTRUCTION = ( @@ -182,53 +186,209 @@ def _read(home: Path, automation_id: str, connection: sqlite3.Connection) -> tup 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]: +def _desired_prompt(*, registry: Path, prompt: str, goal_id: str, agent_id: str, + runtime_root: str | None, cli_bin: str) -> str: + """Desired installed body: keep an exact loader's own binding, else the loader.""" + 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) + loaded_binding = host_bootstrap_binding(prompt) + if (loaded_binding and loaded_binding["registry"].resolve() == registry.resolve() + and (loaded_binding.get("codex_app") or + loaded_binding.get("runtime_profile") == "codex_app_heartbeat")): + # Upgrade the wrapper while retaining explicit owner + # policy and scheduler inputs from the exact loader. + desired = goal_bootstrap(SimpleNamespace(**loaded_binding), registry=registry) + return desired + + +def _classify_entry(*, home: Path, connection: sqlite3.Connection, automation_id: str, + goals: dict[str, Any], registry: Path, runtime_root: str | None, + cli_bin: str) -> dict[str, Any]: + """Classify one installed automation; shared by the batch plan and one-lane reads.""" from loopx.agent_registry import registered_agent_ids_for_goal + + entry: dict[str, Any] = {"automation_id": automation_id} + try: + source, item, row = _read(home, automation_id, 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 = _desired_prompt(registry=registry, prompt=prompt, goal_id=goal_id, + agent_id=agent_id, runtime_root=runtime_root, 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)) + return entry + + +def _registry_goals(registry: Path) -> dict[str, Any]: from loopx.history import load_registry from loopx.registry import registry_goals + return {str(goal["id"]): goal for goal in registry_goals(load_registry(registry))} + + +def build_plan(*, registry: Path, home: Path | None = None, + runtime_root: str | None = None, cli_bin: str = "loopx") -> dict[str, Any]: home = (home or codex_home()).expanduser().resolve() - goals = {str(goal["id"]): goal for goal in registry_goals(load_registry(registry))} + goals = _registry_goals(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) - loaded_binding = host_bootstrap_binding(prompt) - if (loaded_binding and loaded_binding["registry"].resolve() == registry.resolve() - and (loaded_binding.get("codex_app") or - loaded_binding.get("runtime_profile") == "codex_app_heartbeat")): - # Upgrade the wrapper while retaining explicit owner - # policy and scheduler inputs from the exact loader. - desired = goal_bootstrap(SimpleNamespace(**loaded_binding), registry=registry) - 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) + entries.append(_classify_entry(home=home, connection=connection, + automation_id=path.parent.name, goals=goals, registry=registry, + runtime_root=runtime_root, cli_bin=cli_bin)) 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 automation_update_request(*, automation_id: str, manifest: Mapping[str, Any], + expected_prompt_sha256: str, desired_prompt: str) -> dict[str, Any]: + """Complete prompt-only App request; scheduling and thread binding are preserved. + + The CLI cannot call an in-App tool itself, so every entrypoint that finds a + stale installed body hands the host this exact reviewed request. + """ + return {"tool": "automation_update", + "expected_prompt_sha256": expected_prompt_sha256, + "precondition": "View the same automation; verify this prompt hash and all " + "preserved fields before update; read back afterward.", + "arguments": {"mode": "update", "id": automation_id, "kind": "heartbeat", + "name": manifest["name"], "status": manifest["status"], + "rrule": manifest["rrule"], + "targetThreadId": manifest["target_thread_id"], + "notificationPolicy": manifest.get("notification_policy"), + "prompt": desired_prompt}} + + +def installed_prompt_binding(*, registry: Path, goal_id: str, agent_id: str, + home: Path | None = None, thread_id: str | None = None, + runtime_root: str | None = None, + cli_bin: str = "loopx") -> dict[str, Any]: + """Read-only: is the automation driving this lane the current managed loader? + + A frozen execution body keeps applying the policy of the day it was installed + to every later wake, and update-time reconciliation reports that only once, so + the live turn contract has to carry the observation. Bounded and fail-open: an + unreadable store reports a status instead of failing the turn. + + Several installed automations can claim one Goal/agent, so discovery is + TOML-only and the automation bound to the current thread wins. Only a body both + stores confirm is classified: a cron row or an unbound look-alike cannot + describe this lane. + """ + home = (home or codex_home()).expanduser().resolve() + observed: dict[str, Any] = {"schema_version": PROMPT_BINDING_SCHEMA_VERSION, + "status": "absent", "automation_id": None, "host_action": "none", + "host_action_contract": "none", "spend_policy": PROMPT_BINDING_SPEND_POLICY} + installed = load_codex_app_automation_manifest(home) + if not installed.get("available"): + return {**observed, "status": "unavailable", "reason": str(installed.get("reason") + or "no installed Codex App automation store")[:200]} + lane = [item for item in installed.get("entries") or [] if isinstance(item, dict) + and item.get("installed") is True and item.get("goal_id") == goal_id + and item.get("agent_id") == agent_id] + if not lane: + return observed + thread_id = str(thread_id or "").strip() + bound = {str(item.get("automation_id")) for item in lane if thread_id + and str(item.get("target_thread_id") or "") == thread_id} + unresolved: list[str] = [] + confirmed: list[tuple[str, str]] = [] + try: + with closing(_connect(home)) as connection: + for item in lane: + automation_id = str(item.get("automation_id") or "") + try: + _, installed_item, _ = _read(home, automation_id, connection) + except ValueError: + unresolved.append(automation_id) + continue + confirmed.append((automation_id, installed_item["prompt"])) + chosen = [pair for pair in confirmed if pair[0] in bound] or confirmed + if len(chosen) > 1: + return {**observed, "status": "ambiguous", "goal_id": goal_id, + "agent_id": agent_id, + "automation_ids": sorted(pair[0] for pair in chosen)[:8], + "reason": "several installed automations claim this Goal/agent; " + "reconcile through the App"} + entry = (_installed_body_entry(chosen[0][1], home=home, connection=connection, + automation_id=chosen[0][0], registry=registry, + goal_id=goal_id, agent_id=agent_id, runtime_root=runtime_root, + cli_bin=cli_bin) if chosen else None) + except (OSError, ValueError, sqlite3.Error) as error: + return {**observed, "status": "unavailable", "goal_id": goal_id, "agent_id": agent_id, + "reason": str(error)[:200]} + if entry is None: + return {**observed, "status": "blocked", "goal_id": goal_id, "agent_id": agent_id, + "unresolved_automation_ids": sorted(unresolved)[:8], + "reason": "no installed automation for this Goal/agent is confirmed by both " + "stores; reconcile through the App"} + status = str(entry.get("status") or "unknown") + if status not in {"current", "adoption_required"}: + # An unmanaged or unconfirmable body is not adoption authority, so it + # reports its own classification instead of a request the host must not + # apply. + return {**observed, "status": status, "automation_id": str(entry["automation_id"]), + "goal_id": goal_id, "agent_id": agent_id, "reason": entry.get("reason")} + result = {**observed, "status": status, "automation_id": str(entry["automation_id"]), + "goal_id": goal_id, "agent_id": agent_id, "prompt_sha256": entry["prompt_sha256"], + "desired_sha256": entry["desired_sha256"]} + if status == "current": + return result + try: + manifest = tomllib.loads((home / "automations" / str(entry["automation_id"]) + / "automation.toml").read_text(encoding="utf-8")) + if {"name", "status", "rrule", "target_thread_id"} - manifest.keys(): + raise ValueError("installed automation is missing fields the App request must preserve") + except (OSError, UnicodeError, tomllib.TOMLDecodeError, ValueError) as error: + return {**result, "status": "blocked", "reason": str(error)[:200]} + return {**result, "automation_status": str(manifest.get("status") or ""), + "host_action": PROMPT_BINDING_ADOPT_ACTION, + "host_action_contract": PROMPT_BINDING_HOST_ACTION_CONTRACT, + "reason": "the installed automation body is not the current managed loader; " + "review this prompt-only request through the App", + "api_update_request": automation_update_request( + automation_id=str(entry["automation_id"]), manifest=manifest, + expected_prompt_sha256=entry["prompt_sha256"], + desired_prompt=entry["desired_prompt"])} + + +def _installed_body_entry(prompt: str, *, home: Path, connection: sqlite3.Connection, + automation_id: str, registry: Path, goal_id: str, agent_id: str, + runtime_root: str | None, cli_bin: str) -> dict[str, Any]: + """Classify one confirmed body; an exact loader settles its own binding. + + A turn must never retarget another Codex home, registry, or CLI binary, so a + recognized loader is reviewed against the registry it already names; only an + unrecognized body is reviewed against the caller's registry. + """ + loaded = host_bootstrap_binding(prompt) + if loaded is None: + return _classify_entry(home=home, connection=connection, automation_id=automation_id, + goals=_registry_goals(registry), registry=registry, runtime_root=runtime_root, + cli_bin=cli_bin) + desired = _desired_prompt(registry=loaded["registry"], prompt=prompt, goal_id=goal_id, + agent_id=agent_id, runtime_root=runtime_root, cli_bin=cli_bin) + return {"automation_id": automation_id, + "status": "current" if prompt == desired else "adoption_required", + "prompt_sha256": digest(prompt), "desired_prompt": desired, + "desired_sha256": digest(desired)} + + def apply_offline(*, home: Path, automation_id: str, expected_prompt_sha256: str, desired_prompt: str, expected_source_sha256: str | None = None) -> dict[str, Any]: """Journaled prompt-only write, also used by qualified update-time migration. diff --git a/loopx/control_plane/heartbeat/installed_prompt_update.py b/loopx/control_plane/heartbeat/installed_prompt_update.py index 94569f89f3..ca1ea7bfac 100644 --- a/loopx/control_plane/heartbeat/installed_prompt_update.py +++ b/loopx/control_plane/heartbeat/installed_prompt_update.py @@ -14,7 +14,14 @@ import sys import tempfile -from .automation_upgrade import SCHEMA, _atomic, apply_offline, bootstrap_binding, build_plan +from .automation_upgrade import ( + SCHEMA, + _atomic, + apply_offline, + automation_update_request, + bootstrap_binding, + build_plan, +) from .bootstrap_prompt import host_bootstrap_binding @@ -114,14 +121,9 @@ def reconcile(*, before: dict, registry: Path, home: Path, manifest = {} required = {"name", "status", "rrule", "target_thread_id"} if required <= manifest.keys() and manifest.get("prompt") == now["current_prompt"]: - api_updates.append({"tool": "automation_update", - "expected_prompt_sha256": now["prompt_sha256"], - "precondition": "View the same automation; verify this prompt hash and all preserved fields before update; read back afterward.", - "arguments": {"mode": "update", "id": identifier, "kind": "heartbeat", - "name": manifest["name"], "status": manifest["status"], - "rrule": manifest["rrule"], "targetThreadId": manifest["target_thread_id"], - "notificationPolicy": manifest.get("notification_policy"), - "prompt": now["desired_prompt"]}}) + api_updates.append(automation_update_request(automation_id=identifier, + manifest=manifest, expected_prompt_sha256=now["prompt_sha256"], + desired_prompt=now["desired_prompt"])) results.append(result) pending = any(result["status"] not in {"current", "updated", "unmanaged", "missing"} for result in results) return {"ok": not pending, "status": "attention_required" if pending else "current", "results": results, diff --git a/loopx/control_plane/heartbeat/rules.py b/loopx/control_plane/heartbeat/rules.py index 264bd5f2b9..9e16e10d45 100644 --- a/loopx/control_plane/heartbeat/rules.py +++ b/loopx/control_plane/heartbeat/rules.py @@ -41,7 +41,9 @@ ) SCHEDULER_HINT_APPLICATION_RULE = ( "`scheduler_hint` no-spend. host_action=pause_or_delete_current_heartbeat -> " - "automation_update stop once, verify, end; else apply_needed -> RRULE via " + "automation_update stop once, verify, end; prompt_binding.host_action=" + "adopt_managed_bootstrap -> apply its prompt-only api_update_request once, " + "verify, then rerun quota; else apply_needed -> RRULE via " "automation_update; unavailable -> use fallback_hint.cli_args only when projected " "(SQLite/app API " "bypass - fallback only), then ack; further failure -> failure_hint; " @@ -49,11 +51,14 @@ ) SCHEDULER_HINT_COMPACT_RULE = ( "host_action=pause_or_delete_current_heartbeat: automation_update stop; " + "prompt_binding.host_action=adopt_managed_bootstrap: prompt-only " + "automation_update, no spend; " "else RRULE apply via automation_update, projected fallback_hint when unavailable, " "then ack/fail. No spend." ) SCHEDULER_HINT_THIN_RULE = ( "host_action=pause_or_delete_current_heartbeat->automation_update stop(no-spend); " + "prompt_binding=adopt(no-spend); " "else RRULE/projected-fallback_hint/ack/fail." ) RUNTIME_CAPABILITY_PROJECTION_THIN_RULE = ( diff --git a/loopx/control_plane/quota/app_automation_observation.py b/loopx/control_plane/quota/app_automation_observation.py new file mode 100644 index 0000000000..66b428a375 --- /dev/null +++ b/loopx/control_plane/quota/app_automation_observation.py @@ -0,0 +1,81 @@ +"""Observe this lane's Codex App automation for one turn contract. + +The turn packet has to describe the automation the host actually runs, not what +the last update intended: the cadence state lives under the runtime root, and a +frozen installed body keeps applying the policy of the day it was installed. +Both reads are bounded and fail-open, so a missing host store degrades to "not +observed" instead of failing the turn. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import os +from pathlib import Path +from typing import Any + +from ..scheduler.state import load_app_automation_scheduler_state +from ..todos.contract import normalize_todo_claimed_by + + +@dataclass(frozen=True) +class LaneAppAutomationObservation: + """The lane's App automation as the host has it, with unobserved parts None.""" + + scheduler_state: dict[str, Any] | None + prompt_binding: dict[str, Any] | None + + +def observe_lane_app_automation( + status_payload: dict[str, Any], + *, + goal_id: str, + agent_id: str | None, + surface: str, +) -> LaneAppAutomationObservation: + """Read the lane's cadence state and installed prompt body once per turn.""" + + safe_agent_id = normalize_todo_claimed_by(agent_id) + raw_runtime_root = status_payload.get("runtime_root") + return LaneAppAutomationObservation( + scheduler_state=( + load_app_automation_scheduler_state( + Path(str(raw_runtime_root)).expanduser(), + goal_id=goal_id, + agent_id=safe_agent_id, + surface=surface, + ) + if raw_runtime_root and safe_agent_id + else None + ), + prompt_binding=_installed_prompt_binding( + status_payload, + goal_id=goal_id, + agent_id=safe_agent_id, + ), + ) + + +def _installed_prompt_binding( + status_payload: dict[str, Any], + *, + goal_id: str, + agent_id: str | None, +) -> dict[str, Any] | None: + """Report only bindings the host has to act on, and never fail the turn.""" + + registry = str(status_payload.get("registry") or "").strip() + if not agent_id or not registry: + return None + from ..heartbeat.automation_upgrade import installed_prompt_binding + + try: + binding = installed_prompt_binding( + registry=Path(registry).expanduser(), + goal_id=goal_id, + agent_id=agent_id, + thread_id=str(os.environ.get("CODEX_THREAD_ID") or "").strip(), + ) + except (OSError, ValueError): + return None + return binding if binding.get("status") not in {"absent", "unavailable"} else None diff --git a/loopx/control_plane/quota/should_run_packet.py b/loopx/control_plane/quota/should_run_packet.py index ab7eb108d0..863f9a63f0 100644 --- a/loopx/control_plane/quota/should_run_packet.py +++ b/loopx/control_plane/quota/should_run_packet.py @@ -31,6 +31,7 @@ from ..goals.goal_frontier import ( AUTONOMOUS_REPLAN_REQUIRED_MODE, ) +from ..quota.app_automation_observation import observe_lane_app_automation from ..quota.decision_summary import ( quota_decision_agent_id, refine_quota_recommended_action, @@ -87,9 +88,6 @@ build_external_evidence_observation_obligation, ) from ..scheduler.scheduler_hint import build_scheduler_hint -from ..scheduler.state import ( - load_app_automation_scheduler_state, -) from ..todos.contract import ( normalize_todo_claimed_by, ) @@ -187,6 +185,7 @@ def _scheduler_hint( payload: dict[str, Any], *, include_detail: bool = False, codex_app_scheduler_state: dict[str, Any] | None = None, available_capabilities: Any = None, codex_app_current_rrule: Any = None, codex_app_automation_id: Any = None, + codex_app_prompt_binding: dict[str, Any] | None = None, scheduler_execution_context: Mapping[str, Any] | SchedulerExecutionContextResolution | None = None, ) -> dict[str, Any]: return build_scheduler_hint( @@ -197,29 +196,11 @@ def _scheduler_hint( codex_app_scheduler_state=codex_app_scheduler_state, available_capabilities=available_capabilities, codex_app_current_rrule=codex_app_current_rrule, codex_app_automation_id=codex_app_automation_id, + codex_app_prompt_binding=codex_app_prompt_binding, scheduler_execution_context=scheduler_execution_context, ) -def _load_app_automation_scheduler_state( - status_payload: dict[str, Any], - *, - goal_id: str, - agent_id: str | None, - surface: str, -) -> dict[str, Any] | None: - raw_runtime_root = status_payload.get("runtime_root") - safe_agent_id = normalize_todo_claimed_by(agent_id) - if not raw_runtime_root or not safe_agent_id: - return None - return load_app_automation_scheduler_state( - Path(str(raw_runtime_root)).expanduser(), - goal_id=goal_id, - agent_id=safe_agent_id, - surface=surface, - ) - - def _execution_obligation( *, should_run: bool, @@ -1463,25 +1444,30 @@ def _build_quota_should_run_payload( turn_instance_id=turn_instance_id, runtime_root=_interaction_runtime_root(runtime_root, prepared.status_payload), ) + lane_automation = ( + observe_lane_app_automation( + prepared.status_payload, + goal_id=prepared.safe_goal_id, + agent_id=quota_decision_agent_id(payload) or prepared.requested_agent_id, + surface=prepared.resolved_scheduler_context.context.host_surface.value, + ) + if prepared.resolved_scheduler_context.ok + and prepared.resolved_scheduler_context.context is not None + and prepared.resolved_scheduler_context.context.app_automation_applicable + else None + ) payload["scheduler_hint"] = _scheduler_hint( payload, include_detail=prepared.include_scheduler_detail, available_capabilities=prepared.runtime_available_capabilities, codex_app_scheduler_state=( - _load_app_automation_scheduler_state( - prepared.status_payload, - goal_id=prepared.safe_goal_id, - agent_id=quota_decision_agent_id(payload) - or prepared.requested_agent_id, - surface=prepared.resolved_scheduler_context.context.host_surface.value, - ) - if prepared.resolved_scheduler_context.ok - and prepared.resolved_scheduler_context.context is not None - and prepared.resolved_scheduler_context.context.app_automation_applicable - else None + lane_automation.scheduler_state if lane_automation else None ), codex_app_current_rrule=prepared.codex_app_current_rrule, codex_app_automation_id=prepared.codex_app_automation_id, + codex_app_prompt_binding=( + lane_automation.prompt_binding if lane_automation else None + ), scheduler_execution_context=prepared.resolved_scheduler_context, ) finalize_user_gate_notification_cooldown( diff --git a/loopx/control_plane/scheduler/scheduler_hint.py b/loopx/control_plane/scheduler/scheduler_hint.py index f385ce5f05..dd6a62e370 100644 --- a/loopx/control_plane/scheduler/scheduler_hint.py +++ b/loopx/control_plane/scheduler/scheduler_hint.py @@ -634,6 +634,7 @@ class _SchedulerHintBuilder: codex_app_scheduler_state: dict[str, Any] | None codex_app_current_rrule: Any codex_app_automation_id: Any + codex_app_prompt_binding: Any include_detail: bool def _identity_value(self, path: str) -> Any: @@ -881,6 +882,10 @@ def build( }, "no_spend_for_cadence_change": True, } + if self.codex_app_prompt_binding: + # The installed body decides which rules the next wake follows, so a + # stale wrapper is part of the observed App automation state. + app_automation["prompt_binding"] = dict(self.codex_app_prompt_binding) stateful_backoff = app_automation["stateful_backoff"] if host_update_failures: stateful_backoff["host_update_failures"] = [ @@ -1153,6 +1158,7 @@ def build_scheduler_hint( available_capabilities: Any = None, codex_app_current_rrule: Any = None, codex_app_automation_id: Any = None, + codex_app_prompt_binding: Any = None, scheduler_execution_context: ( Mapping[str, Any] | SchedulerExecutionContextResolution | None ) = None, @@ -1342,6 +1348,7 @@ def build_scheduler_hint( codex_app_scheduler_state=codex_app_scheduler_state, codex_app_current_rrule=codex_app_current_rrule, codex_app_automation_id=codex_app_automation_id, + codex_app_prompt_binding=codex_app_prompt_binding, include_detail=include_detail, ) if arbitration.disposition == SchedulerDisposition.AGENT_MONITOR_ONLY_WAIT: diff --git a/skills/loopx-project/SKILL.md b/skills/loopx-project/SKILL.md index 5e13884959..779a80455f 100644 --- a/skills/loopx-project/SKILL.md +++ b/skills/loopx-project/SKILL.md @@ -582,7 +582,13 @@ search/use `automation_update` when available. If `automation_update` once to pause the current heartbeat (delete only when the host cannot pause), verify the host result, spend no quota, and end the turn. This terminal host action takes precedence over RRULE handling and requires no -scheduler ACK. Otherwise use `automation_update` only when +scheduler ACK. When the same lane reports +`scheduler_hint.app_automation.prompt_binding.host_action=adopt_managed_bootstrap`, +the installed automation body is a frozen or older wrapper: re-read the live +hint, apply its prompt-only `api_update_request` once through `automation_update` +after checking the reported prompt hash, read the automation back, and spend no +quota. Prompt adoption preserves the binding, schedule, and status, and it is +separate from cadence handling. Otherwise use `automation_update` only when `scheduler_hint.app_automation.stateful_backoff.apply_needed=true` and `scheduler_hint.app_automation.recommended_rrule` is present. After a successful RRULE update, run `loopx` with diff --git a/tests/control_plane/test_automation_prompt_upgrade.py b/tests/control_plane/test_automation_prompt_upgrade.py index a771c8df6c..94ace92f76 100644 --- a/tests/control_plane/test_automation_prompt_upgrade.py +++ b/tests/control_plane/test_automation_prompt_upgrade.py @@ -537,3 +537,206 @@ def test_exact_legacy_host_loader_upgrades_to_v2_without_dropping_explicit_polic _set_fixture_prompt(path, database, malformed) rejected = lifecycle.snapshot(registry=registry, home=home)["entries"][0] assert rejected["status"] != "current" and not rejected["automatic_eligible"] + + +def test_installed_prompt_binding_reviews_a_frozen_body_without_writing(tmp_path): + home, path, database, registry, prompt = fixture(tmp_path) + original = path.read_bytes() + binding = upgrade.installed_prompt_binding(registry=registry, home=home, + goal_id="fixture-goal", agent_id="agent-a") + assert binding["status"] == "adoption_required" + assert binding["automation_id"] == "watch" + assert binding["goal_id"] == "fixture-goal" and binding["agent_id"] == "agent-a" + assert binding["host_action"] == upgrade.PROMPT_BINDING_ADOPT_ACTION + assert binding["host_action_contract"] == upgrade.PROMPT_BINDING_HOST_ACTION_CONTRACT + assert binding["spend_policy"] == upgrade.PROMPT_BINDING_SPEND_POLICY + assert binding["prompt_sha256"] == upgrade.digest(prompt) + request = binding["api_update_request"] + assert request["tool"] == "automation_update" + assert request["expected_prompt_sha256"] == upgrade.digest(prompt) + assert request["arguments"] == { + "mode": "update", "id": "watch", "kind": "heartbeat", + "name": "Fixture watch", "status": "PAUSED", "rrule": "FREQ=HOURLY", + "targetThreadId": "thread-a", "notificationPolicy": "failed_runs_only", + "prompt": upgrade.bootstrap_prompt(registry=registry, goal_id="fixture-goal", + agent_id="agent-a"), + } + assert binding["desired_sha256"] == upgrade.digest(request["arguments"]["prompt"]) + # The stale body is the private local store; only its digest travels. + assert prompt not in json.dumps(binding) + assert path.read_bytes() == original + + +def test_installed_prompt_binding_keeps_a_bound_loader_current(tmp_path): + registry_other = tmp_path / "other" / "registry.json" + prompt = upgrade.bootstrap_prompt(registry=registry_other, goal_id="fixture-goal", + agent_id="agent-a") + home, path, database, registry, _ = fixture(tmp_path) + _set_fixture_prompt(path, database, prompt) + binding = upgrade.installed_prompt_binding(registry=registry, home=home, + goal_id="fixture-goal", agent_id="agent-a") + # A recognized loader is never retargeted to the caller's registry. + assert binding["status"] == "current" + assert binding["host_action"] == "none" + assert "api_update_request" not in binding + + +def test_installed_prompt_binding_offers_only_the_v2_wrapper_upgrade(tmp_path): + registry_path = tmp_path / "registry.json" + prompt = upgrade.bootstrap_prompt(registry=registry_path, goal_id="fixture-goal", + agent_id="agent-a").replace( + upgrade.BOOTSTRAP, upgrade._LEGACY_BOOTSTRAP, 1).removesuffix( + upgrade._BOOTSTRAP_INSTRUCTION) + upgrade._LEGACY_INSTRUCTION + home, path, database, registry, _ = fixture(tmp_path) + _set_fixture_prompt(path, database, prompt) + binding = upgrade.installed_prompt_binding(registry=registry, home=home, + goal_id="fixture-goal", agent_id="agent-a") + assert binding["status"] == "adoption_required" + assert binding["api_update_request"]["arguments"]["prompt"].startswith( + "LoopX managed heartbeat bootstrap v2\n" + ) + + +def test_turn_binding_reuses_the_update_time_request(tmp_path, monkeypatch): + from loopx.control_plane.heartbeat import installed_prompt_update as lifecycle + from loopx.heartbeat_prompt import build_heartbeat_prompt + home, path, database, registry, _ = fixture(tmp_path) + owned = build_heartbeat_prompt(goal_id="fixture-goal", agent_id="agent-a", + registered_agents=["agent-a"], runtime_profile="codex_app_heartbeat", + thin=True)["task_body"] + _set_fixture_prompt(path, database, owned) + binding = upgrade.installed_prompt_binding(registry=registry, home=home, + goal_id="fixture-goal", agent_id="agent-a") + before = lifecycle.snapshot(registry=registry, home=home) + assert before["entries"][0]["automatic_eligible"] is True + monkeypatch.setattr(lifecycle.sys, "platform", "linux") + deferred = lifecycle.reconcile(before=before, registry=registry, home=home) + api_updates = deferred["api_updates"] + # One reviewed prompt-only request shape, whichever entrypoint finds it. + assert len(api_updates) == 1 + assert api_updates[0]["arguments"] == binding["api_update_request"]["arguments"] + assert api_updates[0]["expected_prompt_sha256"] == binding["prompt_sha256"] + + +def test_installed_prompt_binding_fails_open_per_lane(tmp_path): + # No installed host store at all. + empty = upgrade.installed_prompt_binding(registry=tmp_path / "registry.json", + home=tmp_path / "missing", goal_id="fixture-goal", agent_id="agent-a") + assert empty["status"] == "unavailable" and empty["host_action"] == "none" + home, path, database, registry, _ = fixture(tmp_path) + # Another registered agent is not this lane. + other = upgrade.installed_prompt_binding(registry=registry, home=home, + goal_id="fixture-goal", agent_id="agent-other") + assert other["status"] == "absent" and "api_update_request" not in other + # Disagreeing stores are reconciled through the App, never by prompt adoption. + with sqlite3.connect(database) as connection: + connection.execute("UPDATE automations SET target_thread_id='thread-b'") + conflicted = upgrade.installed_prompt_binding(registry=registry, home=home, + goal_id="fixture-goal", agent_id="agent-a") + assert conflicted["status"] == "blocked" + assert "api_update_request" not in conflicted + + +def test_installed_prompt_binding_clears_after_the_reviewed_request_is_applied(tmp_path): + home, path, database, registry, _ = fixture(tmp_path) + binding = upgrade.installed_prompt_binding(registry=registry, home=home, + goal_id="fixture-goal", agent_id="agent-a") + upgrade.apply_offline(home=home, automation_id="watch", + expected_prompt_sha256=binding["prompt_sha256"], + desired_prompt=binding["api_update_request"]["arguments"]["prompt"]) + settled = upgrade.installed_prompt_binding(registry=registry, home=home, + goal_id="fixture-goal", agent_id="agent-a") + assert settled["status"] == "current" + assert settled["prompt_sha256"] == binding["desired_sha256"] + + +def _add_look_alike(home: Path, database: Path, automation_id: str, prompt: str, *, + row_kind: str = "heartbeat", thread_id: str = "thread-a") -> None: + """Add a second record that names the lane but is not a confirmable heartbeat.""" + + path = home / "automations" / automation_id / "automation.toml" + path.parent.mkdir(parents=True) + path.write_text('version = 1\n' + f'id = "{automation_id}"\nname = "Look alike"\n' + 'kind = "heartbeat"\nstatus = "PAUSED"\n' + f'target_thread_id = "{thread_id}"\nrrule = "FREQ=HOURLY"\n' + 'prompt = ' + json.dumps(prompt) + "\n", encoding="utf-8") + with sqlite3.connect(database) as connection: + connection.execute("INSERT INTO automations VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + (automation_id, row_kind, prompt, "PAUSED", thread_id, "FREQ=HOURLY", None, 1, 2)) + + +def test_installed_prompt_binding_keeps_the_confirmable_record_in_front(tmp_path): + home, path, database, registry, prompt = fixture(tmp_path) + view = upgrade.installed_prompt_binding(registry=registry, home=home, + goal_id="fixture-goal", agent_id="agent-a") + assert view["status"] == "adoption_required" + assert "unresolved_automation_ids" not in view + # A converted cron row that only mentions the lane must not hide the live record. + _add_look_alike(home, database, "legacy-lookalike", prompt, row_kind="cron") + view = upgrade.installed_prompt_binding(registry=registry, home=home, + goal_id="fixture-goal", agent_id="agent-a") + assert view["status"] == "adoption_required" + assert view["automation_id"] == "watch" + assert view["api_update_request"]["arguments"]["id"] == "watch" + + +def test_installed_prompt_binding_reports_two_confirmable_records_as_ambiguous(tmp_path): + home, path, database, registry, prompt = fixture(tmp_path) + _add_look_alike(home, database, "watch-2", prompt) + view = upgrade.installed_prompt_binding(registry=registry, home=home, + goal_id="fixture-goal", agent_id="agent-a") + assert view["status"] == "ambiguous" + assert view["automation_ids"] == ["watch", "watch-2"] + assert "api_update_request" not in view + + +def test_installed_prompt_binding_follows_the_automation_bound_to_the_turn(tmp_path): + home, path, database, registry, prompt = fixture(tmp_path) + _add_look_alike(home, database, "watch-2", prompt, thread_id="thread-b") + # Several confirmable records claim the lane: only the bound one drives it. + bound = upgrade.installed_prompt_binding(registry=registry, home=home, + goal_id="fixture-goal", agent_id="agent-a", thread_id="thread-a") + assert bound["status"] == "adoption_required" + assert bound["automation_id"] == "watch" + unbound = upgrade.installed_prompt_binding(registry=registry, home=home, + goal_id="fixture-goal", agent_id="agent-a", thread_id="thread-c") + assert unbound["status"] == "ambiguous" + assert "api_update_request" not in unbound + + +def test_installed_prompt_binding_names_the_unconfirmed_records(tmp_path): + home, path, database, registry, prompt = fixture(tmp_path) + _add_look_alike(home, database, "legacy-lookalike", prompt, row_kind="cron") + with sqlite3.connect(database) as connection: + connection.execute("UPDATE automations SET kind='cron' WHERE id='watch'") + blocked = upgrade.installed_prompt_binding(registry=registry, home=home, + goal_id="fixture-goal", agent_id="agent-a") + assert blocked["status"] == "blocked" + assert blocked["unresolved_automation_ids"] == ["legacy-lookalike", "watch"] + assert "api_update_request" not in blocked + + +def test_installed_prompt_binding_reports_a_confirmed_but_unmanaged_body(tmp_path): + home, path, database, registry, prompt = fixture(tmp_path) + registered = json.loads(registry.read_text()) + registered["goals"][0]["registered_agents"] = [] + registry.write_text(json.dumps(registered)) + view = upgrade.installed_prompt_binding(registry=registry, home=home, + goal_id="fixture-goal", agent_id="agent-a") + assert view["status"] == "unmanaged" and view["automation_id"] == "watch" + assert "api_update_request" not in view and "desired_sha256" not in view + + +def test_installed_prompt_binding_survives_an_ambiguous_installed_body(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,)) + # A body the owner refuses to classify must degrade to a status, never raise + # out of the read-only observation the live turn contract depends on. + view = upgrade.installed_prompt_binding(registry=registry, home=home, + goal_id="fixture-goal", agent_id="agent-a") + assert view["status"] == "blocked" and view["automation_id"] == "watch" + assert "ambiguous Goal/agent bindings" in view["reason"] + assert "api_update_request" not in view and "desired_sha256" not in view diff --git a/tests/control_plane/test_codex_app_prompt_binding_projection.py b/tests/control_plane/test_codex_app_prompt_binding_projection.py new file mode 100644 index 0000000000..64a3788234 --- /dev/null +++ b/tests/control_plane/test_codex_app_prompt_binding_projection.py @@ -0,0 +1,124 @@ +from __future__ import annotations + +import json +from pathlib import Path +import sqlite3 + +from loopx.control_plane.heartbeat import automation_upgrade as upgrade +from loopx.control_plane.quota.app_automation_observation import ( + observe_lane_app_automation, +) +from loopx.control_plane.quota.should_run import build_quota_should_run +from loopx.control_plane.scheduler.execution_context import ( + scheduler_execution_context_for_runtime_profile, +) +from loopx.control_plane.testing.quota_fixtures import ( + quota_status_payload, + quota_todo_item, +) + + +APP_CONTEXT = scheduler_execution_context_for_runtime_profile("codex_app_heartbeat") +CLI_CONTEXT = scheduler_execution_context_for_runtime_profile("codex_cli") +GOAL_ID = "fixture-goal" +AGENT_ID = "agent-a" +THREAD_ID = "thread-a" +FROZEN_BODY = f"Advance `{GOAL_ID}` from registry. --agent-id {AGENT_ID}" + + +def _host_with_frozen_body(tmp_path: Path) -> tuple[Path, Path]: + home = tmp_path / "host" + path = home / "automations/watch/automation.toml" + path.parent.mkdir(parents=True) + path.write_text('version = 1\nid = "watch"\nname = "Fixture watch"\nkind = "heartbeat"\n' + 'status = "ACTIVE"\ntarget_thread_id = "' + THREAD_ID + '"\n' + 'rrule = "FREQ=MINUTELY;INTERVAL=3"\n' + 'prompt = ' + json.dumps(FROZEN_BODY) + "\n", 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", FROZEN_BODY, "ACTIVE", THREAD_ID, + "FREQ=MINUTELY;INTERVAL=3", "fixture-model", 123, 456)) + 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": GOAL_ID, "repo": str(tmp_path), + "state_file": str(state), "registered_agents": [AGENT_ID]}]}), encoding="utf-8") + return home, registry + + +def _lane_payload(tmp_path: Path, registry: Path) -> dict: + """One runnable lane whose host context the packet has to observe.""" + + return { + **quota_status_payload( + goal_id=GOAL_ID, + status="active", + agent_todo_items=[ + quota_todo_item( + todo_id="todo_current001", + index=1, + priority="P1", + title="Advance the reviewed slice.", + claimed_by=AGENT_ID, + ) + ], + recommended_action="Advance the reviewed slice.", + coordination={"agent_model": "peer_v1", "registered_agents": [AGENT_ID]}, + claim_scope_agent_id=AGENT_ID, + ), + "registry": str(registry), + "runtime_root": str(tmp_path / "runtime"), + } + + +def test_app_lane_projects_the_installed_prompt_binding(tmp_path, monkeypatch): + home, registry = _host_with_frozen_body(tmp_path) + monkeypatch.setenv("CODEX_HOME", str(home)) + monkeypatch.setenv("CODEX_THREAD_ID", THREAD_ID) + packet = build_quota_should_run( + _lane_payload(tmp_path, registry), + goal_id=GOAL_ID, + agent_id=AGENT_ID, + scheduler_execution_context=APP_CONTEXT, + ) + hint = packet["scheduler_hint"] + binding = hint["app_automation"]["prompt_binding"] + assert binding["status"] == "adoption_required" + assert binding["automation_id"] == "watch" + assert binding["host_action"] == upgrade.PROMPT_BINDING_ADOPT_ACTION + assert binding["host_action_contract"] == upgrade.PROMPT_BINDING_HOST_ACTION_CONTRACT + assert binding["api_update_request"]["arguments"]["prompt"].startswith( + "LoopX managed heartbeat bootstrap v2\n" + ) + # The legacy Codex App projection carries the same observed automation. + assert hint["codex_app"]["prompt_binding"] == binding + assert hint["app_automation"]["no_spend_for_cadence_change"] is True + + +def test_non_app_hosts_never_observe_a_prompt_binding(tmp_path, monkeypatch): + home, registry = _host_with_frozen_body(tmp_path) + monkeypatch.setenv("CODEX_HOME", str(home)) + monkeypatch.setenv("CODEX_THREAD_ID", THREAD_ID) + packet = build_quota_should_run( + _lane_payload(tmp_path, registry), + goal_id=GOAL_ID, + agent_id=AGENT_ID, + scheduler_execution_context=CLI_CONTEXT, + ) + assert "prompt_binding" not in json.dumps(packet["scheduler_hint"]) + + +def test_lane_without_an_installed_automation_observes_nothing(tmp_path): + observation = observe_lane_app_automation( + {"registry": str(tmp_path / "registry.json"), "runtime_root": str(tmp_path)}, + goal_id=GOAL_ID, + agent_id=AGENT_ID, + surface="codex_app", + ) + assert observation.prompt_binding is None + assert observation.scheduler_state is None