fix(heartbeat): hint deferred prompt upgrades through budgeted turn-start hooks

docs/reference/automation-prompt-upgrades.md

@@ -68,6 +68,38 @@ 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.
 
+## Deferred upgrade hint
+
+When an automatically eligible migration is deferred, reconciliation records
+only its identity, old prompt digest and CLI route under the private runtime root's
+`automation-prompt-upgrades/` directory. Records are scoped to the registry and
+Codex home; they contain no prompt body or saved host update request.
+
+Codex App heartbeat decisions load a fixed, read-only turn-start hook from the
+existing heartbeat lifecycle. It uses the existing typed capability-hook
+observation and required-read contracts, without adding a standalone capability,
+provider package, prompt template or scheduler action. For one unambiguous pending
+entry whose old prompt and thread still match both host stores, the hook inserts
+an `automation-prompts plan --automation-id ...` read into the existing Agent/CLI
+channel. Read that fresh plan, review the prompt-only adoption through the App,
+and read back the result. Repair alone spends no quota; normal work keeps its
+existing decision and permission boundaries.
+
+The active hint declares `prompt_budget_bytes=1536` in its required read.
+The typed hook validates this optional allowance (at most 2048 bytes per read).
+Only emitted hook reads extend the envelope's 8192-byte budget and their command
+projection allowance; inactive hooks contribute zero. Existing unbudgeted reads
+retain their 360-character projection and the normal envelope budget. This is
+prompt capacity, not execution, quota or adoption authority.
+
+No pending entry, an adopted prompt, customization, a changed thread, ambiguous
+identity or unavailable host evidence produces no adoption hint. With no pending receipt
+the hook does not open the host database or dispatch a capability call.
+An unrelated RRULE change does not hide a pending prompt. The next reconciliation
+removes resolved records; the turn never needs a new ACK or state write.
+Detection remains update-time: edits made outside LoopX between updates are not
+new migration candidates. Re-run `automation-prompts plan` for explicit review.
+
 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 +234,24 @@ gh 登录，仍失败则明确要求已核验 SHA，不切换分支或静默覆
 日程、暂停状态、模型、线程、通知偏好和历史均不迁移。
 不支持的存储仍需原生 API；运行中的本轮不热切换。普通测试不消耗模型 token，
 真实模型发布资格仍需独立评测，不能由迁移成功推断。
+
+自动升级候选未能应用时，对账只在 runtime root 的
+`automation-prompt-upgrades/` 中记录按 registry 和 Codex home 隔离的身份、
+prompt 摘要与 CLI 路由，不保存 prompt 正文或宿主更新请求。Codex App heartbeat
+固定加载现有 heartbeat 生命周期内的只读 turn-start hook，复用已有的类型化
+capability-hook 观察与 required-read 契约，不新增独立 capability、provider 包、
+prompt 模板或 scheduler action。只有唯一未完成项的旧 prompt 和线程仍与两个
+宿主存储一致时，才在现有 Agent/CLI 通道插入指定 automation 的最新 plan 读取提示。
+按最新计划审阅、通过 App 仅更新 prompt 并读回；修复本身不消耗额度，正常工作仍按
+原有决策和权限执行。
+
+无未完成项、已升级、自定义修改、线程变化、身份歧义或无法核验时不注入采纳提示；
+无未完成记录时不打开宿主数据库、不调用 capability 分发器。单独的 RRULE 变化不影响提示。
+完成升级即停止提示，下次对账清除记录，无需新的 ACK 或按轮状态写入。发现仍发生在
+升级时；两次升级之间的外部修改不会自动成为迁移候选，可显式运行
+`automation-prompts plan` 审阅。
+
+激活的 hint 同时声明 `prompt_budget_bytes=1536`，由类型化 hook 校验（单条最多
+2048 字节）。只有实际输出的 hook read 才增加 envelope 原有 8192 字节预算及该条
+命令的投影空间；未激活时增加量为零。未声明预算的 read 保留原有 360 字符投影和
+默认 envelope 预算。这仅增加 prompt 容量，不增加执行、额度或采纳权限。

loopx/control_plane/capability_hooks.ts

@@ -351,6 +351,15 @@ export function validateInteractionProjectionHookInvocation(input: {
   };
 }
 
+/** Optional per-read prompt allowance; never execution or effect authority. */
+export function turnStartPromptBudgetBytes(value: unknown): number {
+  if (value === undefined) return 0;
+  if (typeof value !== "number" || !Number.isSafeInteger(value) || value < 1 || value > 2_048) {
+    throw new Error("turn-start prompt budget must be an integer from 1 to 2048 bytes");
+  }
+  return value;
+}
+
 export function validateTurnStartHookRegistration(
   value: unknown,
 ): JsonObject & {
@@ -413,11 +422,10 @@ export function validateTurnStartHookRegistration(
       registration.required_read,
       "turn-start hook required_read",
     );
-    requireExactFields(
-      candidate,
-      TURN_START_REQUIRED_READ_FIELDS,
-      "turn-start hook required_read",
-    );
+    const readFields = new Set(TURN_START_REQUIRED_READ_FIELDS);
+    if ("prompt_budget_bytes" in candidate) readFields.add("prompt_budget_bytes");
+    requireExactFields(candidate, readFields, "turn-start hook required_read");
+    const promptBudget = turnStartPromptBudgetBytes(candidate.prompt_budget_bytes);
     const kind = requiredString(candidate.kind, "turn-start hook required_read kind");
     const command = requiredString(
       candidate.command,
@@ -443,6 +451,12 @@ export function validateTurnStartHookRegistration(
       throw new Error("turn-start hook required_read ordering is invalid");
     }
     requiredRead = { kind, command, reason, ordering: "before_work" };
+    if (promptBudget) {
+      requiredRead.prompt_budget_bytes = promptBudget;
+      if (Buffer.byteLength(JSON.stringify(requiredRead), "utf8") > promptBudget) {
+        throw new Error("turn-start required read exceeds its declared prompt budget");
+      }
+    }
   }
   return {
     ...registration,

loopx/control_plane/heartbeat/installed_prompt_update.py

@@ -123,6 +123,9 @@ def reconcile(*, before: dict, registry: Path, home: Path,
                             "notificationPolicy": manifest.get("notification_policy"),
                             "prompt": now["desired_prompt"]}})
         results.append(result)
+    from .prompt_upgrade_hook import record_deferred_upgrades
+    record_deferred_upgrades(registry=registry, home=home, runtime_root=runtime_root,
+        cli_bin=cli_bin, entries=current, results=results)
     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,
             "api_updates": api_updates,

loopx/control_plane/heartbeat/prompt_upgrade_hook.py

@@ -0,0 +1,129 @@
+"""Content-free update receipts and a read-only, conditional turn-start hint.
+
+Installed-prompt reconciliation owns discovery. The existing typed hook owns
+observation admission; neither this receipt nor its hint authorizes adoption.
+"""
+from __future__ import annotations
+
+from contextlib import closing
+import json
+from pathlib import Path
+import shlex
+from typing import Any, Mapping
+
+from ...file_lock import exclusive_file_lock
+from ...history import load_registry
+from ...paths import resolve_runtime_root
+from ...upgrade import codex_home
+from ..capability_hooks import (
+    TURN_START_HOOK_RESULT_SCHEMA_VERSION,
+    TurnStartHookRegistration,
+    dispatch_turn_start_hooks,
+)
+from .automation_upgrade import _atomic, _connect, _read, digest
+
+_SCHEMA = "loopx_deferred_prompt_upgrade_v0"
+_HOOK = "heartbeat.prompt_upgrade"
+
+
+def receipt_path(runtime_root: Path, registry: Path, home: Path) -> Path:
+    scope = json.dumps([str(registry.resolve()), str(home.resolve())])
+    return runtime_root / "automation-prompt-upgrades" / (digest(scope) + ".json")
+
+
+def _read_receipts(path: Path) -> dict[str, Any]:
+    if not path.exists():
+        return {}
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    if (not isinstance(payload, dict) or payload.get("schema_version") != _SCHEMA
+            or not isinstance(payload.get("entries"), dict)):
+        raise ValueError("invalid deferred prompt upgrade receipt")
+    return dict(payload["entries"])
+
+
+def record_deferred_upgrades(*, registry: Path, home: Path, runtime_root: str | None,
+                             cli_bin: str, entries: dict[str, Any], results: list[dict[str, Any]]) -> None:
+    root = resolve_runtime_root(load_registry(registry), runtime_root, registry_path=registry)
+    path = receipt_path(root, registry, home)
+    if not path.exists() and not any(result["status"] == "deferred" for result in results):
+        return
+    # Independent sync-installed subsets cannot erase each other's reminders.
+    with exclusive_file_lock(path):
+        pending = _read_receipts(path)
+        for result in results:
+            identifier = result["automation_id"]
+            if result["status"] == "deferred":
+                entry = entries[identifier]
+                pending[identifier] = {key: entry[key] for key in (
+                    "goal_id", "agent_id", "prompt_sha256", "target_thread_id",
+                )}
+                pending[identifier].update(cli_bin=cli_bin, runtime_root=runtime_root)
+            else:
+                pending.pop(identifier, None)
+        if pending:
+            _atomic(path, json.dumps({"schema_version": _SCHEMA, "entries": pending}))
+        elif path.exists():
+            path.unlink()
+
+
+def prompt_upgrade_hook(*, registry: Path, runtime_root: Path, goal_id: str,
+                        agent_id: str) -> TurnStartHookRegistration | None:
+    home = codex_home().expanduser().resolve()
+    pending = _read_receipts(receipt_path(runtime_root, registry, home))
+    candidates = [(key, value) for key, value in pending.items()
+        if isinstance(value, dict) and value.get("goal_id") == goal_id
+        and value.get("agent_id") == agent_id]
+    if len(candidates) != 1:
+        return None
+    identifier, record = candidates[0]
+    command = [record["cli_bin"], "--format", "json", "--registry", str(registry)]
+    if record.get("runtime_root"):
+        command += ["--runtime-root", record["runtime_root"]]
+    command += ["automation-prompts", "plan", "--codex-home", str(home),
+        "--automation-id", identifier, "--cli-bin", record["cli_bin"]]
+    required_read = {
+        "kind": "automation_prompt_upgrade",
+        "command": shlex.join(command),
+        "reason": "A managed prompt upgrade is pending. Read the fresh plan; review and apply only the prompt through automation_update, then read back. Preserve other fields; no quota spend for repair. Continue normal work under its existing decision.",
+        "ordering": "before_work",
+        "prompt_budget_bytes": 1536,
+    }
+
+    def produce() -> dict[str, Any]:
+        with closing(_connect(home)) as connection:
+            _, item, _ = _read(home, identifier, connection)
+        applicable = (digest(item["prompt"]) == record.get("prompt_sha256")
+            and item["target_thread_id"] == record.get("target_thread_id"))
+        return {
+            "schema_version": TURN_START_HOOK_RESULT_SCHEMA_VERSION,
+            "hook_id": _HOOK, "capability_id": "automation-prompt-upgrade",
+            "phase": "turn_start", "status": "observed" if applicable else "empty",
+            "observation_count": int(applicable), "agent_read_required": applicable,
+            "external_reads_performed": False, "external_writes_performed": False,
+            "local_private_state_mutated": False, "private_content_returned": False,
+            "provider_payload_returned": False, "error_code": None,
+        }
+
+    return TurnStartHookRegistration(
+        hook_id=_HOOK, capability_id="automation-prompt-upgrade",
+        requested_read_scope=("deferred_prompt_upgrade", "installed_automation"),
+        requested_write_scope=(), producer=produce, required_read=required_read,
+    )
+
+
+def extend_prompt_upgrade_reads(
+    dispatch: Mapping[str, Any] | None, **kwargs: Any,
+) -> Mapping[str, Any] | None:
+    """Fixed hook loading; healthy lanes preserve their exact existing projection."""
+    try:
+        hook = prompt_upgrade_hook(**kwargs)
+        if hook is None:
+            return dispatch
+        extra = dispatch_turn_start_hooks((hook,))
+    except (OSError, ValueError, KeyError, TypeError):
+        return dispatch
+    if not extra["required_reads"]:
+        return dispatch
+    return {**(dispatch or {}), "required_reads": [
+        *(dispatch or {}).get("required_reads", []), *extra["required_reads"],
+    ]}

loopx/control_plane/quota/live_decision.py

@@ -73,7 +73,7 @@ def _turn_start_required_reads(
         projected.append(
             {
                 key: read[key]
-                for key in ("kind", "command", "reason", "source", "ordering")
+                for key in ("kind", "command", "reason", "source", "ordering", "prompt_budget_bytes")
                 if key in read
             }
         )
@@ -484,6 +484,13 @@ def build_live_quota_should_run_decision(
         available_capabilities = remembered_runtime
     if route_source.startswith("loopx_turn_"):
         payload["runtime_root"] = str(runtime_root)
+    if codex_app_host and agent_id:
+        from ..heartbeat.prompt_upgrade_hook import extend_prompt_upgrade_reads
+
+        turn_start_hook_dispatch = extend_prompt_upgrade_reads(
+            turn_start_hook_dispatch, registry=registry_path, runtime_root=runtime_root,
+            goal_id=goal_id, agent_id=agent_id,
+        )
     _project_turn_start_required_reads(
         payload,
         turn_start_hook_dispatch,

loopx/control_plane/quota/turn_envelope.ts

@@ -5,9 +5,10 @@ import {
   type JsonObject,
 } from "../effect_program.ts";
 import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts";
+import { turnStartPromptBudgetBytes } from "../capability_hooks.ts";
 import { requireJsonObject } from "../runtime_decode.ts";
 import { projectPendingCapabilityIntent } from "../work_items/pending_capability_intent.ts";
-import { measureTurnEnvelope, TURN_ENVELOPE_BUDGET_BYTES } from "./turn_envelope_budget.ts";
+import { measureTurnEnvelope, turnEnvelopeBudgetBytes } from "./turn_envelope_budget.ts";
 export { TURN_ENVELOPE_BUDGET_BYTES } from "./turn_envelope_budget.ts";
 
 export const TURN_ENVELOPE_SCHEMA_VERSION = "loopx_turn_envelope_v0";
@@ -294,9 +295,12 @@ function requiredReads(interaction: JsonObject, payload: JsonObject): JsonObject
   const result: JsonObject[] = [];
   for (const value of raw.slice(0, 5)) {
     const item = object(value);
-    const command = text(item.command, 360);
+    const promptBudget = item.source === "turn_start_capability_hook"
+      ? turnStartPromptBudgetBytes(item.prompt_budget_bytes) : 0;
+    const command = text(item.command, promptBudget || 360);
     if (!command) continue;
     const compact: JsonObject = { command };
+    if (promptBudget) compact.prompt_budget_bytes = promptBudget;
     for (const field of ["kind", "reason", "source"]) {
       const rendered = text(item[field], 240);
       if (rendered) compact[field] = rendered;
@@ -681,7 +685,7 @@ function turnActionProjection(payload: JsonObject, protocolActionFields: JsonObj
   // Guidance must not crowd out the actionable contract. Preserve a signed
   // content reference to the existing full-decision route under budget pressure.
   if (Object.keys(context).length > 0
-    && Buffer.byteLength(JSON.stringify(projection), "utf8") > TURN_ENVELOPE_BUDGET_BYTES - 1_400) {
+    && Buffer.byteLength(JSON.stringify(projection), "utf8") > turnEnvelopeBudgetBytes(projection) - 1_400) {
     projection.agent_context = {
       schema_version: context.schema_version, phase: context.phase, scope: context.scope,
       target: "coordinator", authority: "guidance_only", delivery: "projected",

loopx/control_plane/quota/turn_envelope_budget.ts

@@ -1,4 +1,5 @@
 /** Performance diagnostics, never Turn admission or execution authority. */
+import { turnStartPromptBudgetBytes } from "../capability_hooks.ts";
 import type { JsonObject } from "../effect_program.ts";
 
 export const TURN_ENVELOPE_BUDGET_BYTES = 8_192;
@@ -30,13 +31,25 @@ function sectionBytes(envelope: JsonObject): Record<Section, number> {
   return sizes;
 }
 
+export function turnEnvelopeBudgetBytes(envelope: JsonObject): number {
+  const reads = Array.isArray(envelope.required_reads) ? envelope.required_reads : [];
+  return TURN_ENVELOPE_BUDGET_BYTES + reads.slice(0, 5).reduce((total: number, value: unknown) => {
+    if (!value || typeof value !== "object" || Array.isArray(value)) return total;
+    const read = value as JsonObject;
+    return total + (read.source === "turn_start_capability_hook"
+      ? turnStartPromptBudgetBytes(read.prompt_budget_bytes) : 0);
+  }, 0);
+}
+
 export function measureTurnEnvelope(envelope: JsonObject, source: JsonObject): void {
   // Keep v0 *_json_bytes code-point metrics for compatibility. New diagnostics
   // and the performance target use actual compact JSON UTF-8 bytes.
   const sourceChars = [...JSON.stringify(source)].length;
+  const budgetBytes = turnEnvelopeBudgetBytes(envelope);
+  const hookBudget = budgetBytes - TURN_ENVELOPE_BUDGET_BYTES;
   envelope.compaction = {
     source_json_bytes: sourceChars, envelope_json_bytes: 0,
-    byte_reduction_ratio: 0, budget_bytes: TURN_ENVELOPE_BUDGET_BYTES,
+    byte_reduction_ratio: 0, budget_bytes: budgetBytes,
     within_budget: true, envelope_utf8_bytes: 0,
   };
   // Measurements include their own serialized metadata. Recompute to a fixed
@@ -56,18 +69,19 @@ export function measureTurnEnvelope(envelope: JsonObject, source: JsonObject): v
       byte_reduction_ratio: ratioLocked
         ? (envelope.compaction as JsonObject).byte_reduction_ratio : sourceChars
         ? Math.round((1 - chars / sourceChars) * 10_000) / 10_000 : 0,
-      budget_bytes: TURN_ENVELOPE_BUDGET_BYTES,
-      within_budget: bytes <= TURN_ENVELOPE_BUDGET_BYTES,
+      budget_bytes: budgetBytes,
+      within_budget: bytes <= budgetBytes,
       envelope_utf8_bytes: bytes,
     };
-    if (bytes > TURN_ENVELOPE_BUDGET_BYTES) {
+    if (hookBudget) metric.hook_prompt_budget_bytes = hookBudget;
+    if (bytes > budgetBytes) {
       const sections = sectionBytes(envelope);
       metric.warning = {
         code: "turn_envelope_budget_exceeded", severity: "warning",
-        excess_bytes: bytes - TURN_ENVELOPE_BUDGET_BYTES,
+        excess_bytes: bytes - budgetBytes,
         section_bytes: sections,
         over_target_sections: (Object.keys(sections) as Section[])
-          .filter((key) => sections[key] > TURN_ENVELOPE_SECTION_TARGETS[key]),
+          .filter((key) => sections[key] > TURN_ENVELOPE_SECTION_TARGETS[key] + (key === "action" ? hookBudget : 0)),
       };
     }
     envelope.compaction = metric;

loopx/control_plane/work_items/interaction_contract.py

@@ -959,7 +959,8 @@ def _interaction_required_reads(payload: dict[str, Any]) -> list[dict[str, Any]]
     for item in reads:
         if not isinstance(item, dict):
             continue
-        command = protocol_action_text(item.get("command"), limit=360)
+        command = protocol_action_text(item.get("command"), limit=(
+            item.get("prompt_budget_bytes", 360) if item.get("source") == "turn_start_capability_hook" else 360))
         if not command:
             continue
         result.append({**item, "command": command})