diff --git a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md index b3adbfccbe..53551621f4 100644 --- a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md +++ b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md @@ -471,6 +471,38 @@ shape merely because of its size. Optional memory participation now has one compact, signed envelope projection; the Codex CLI adapter preserves default-off isolation. This does not qualify installed heartbeat/App adoption or model value. +Memory participation now uses a registered post-selection context contribution, +composed in the capability layer for native Turn and executing quota/heartbeat +guards. Codex, DSH and generic CLI hosts transport the same bounded context, +instructions and optional result fields. The capability owns binding/mode checks; +adapters retain only host formatting. TypeScript bounds contributions, prevents +authority/result-field replacement and binds them to the existing governing +signature document plus their source readback. Adding the original capture's +location does not invalidate an unchanged decision; changed actors, authority, +configuration readback or contributions do invalidate optional context. + +The generic Python contribution IO lives with the existing registrations and +dispatchers in `loopx.control_plane.capability_hooks`; there is no parallel Python +Turn-context module. It calls the TypeScript decoder for validation and source +binding. Host adapters consume that same hook-owned projection. + +Registered `PostSettlement` callbacks call the existing ingest owner after +independent exact-reflection attestation and accountable writeback/spend. The +manual-host sidecar adapter uses that registration too. `post_writeback` remains +effect-free. Per-hook receipts preserve completed work while an uncertain peer +hook recovers; the provider owner retains write deduplication and reconciliation. +Terminal replay preserves the original journal/receipt and returns recovered +owner readback as `post_settlement_recovery`, without repeating settlement. +File/SQLite native host and isolated provider cases qualify this source path. +Installed App/Lark adoption, model benefit, complete costs and the three +convergence acceptances below remain open. + +Pre-registration journal recovery retains the already admitted bounded +reflection and independent validation digest without changing the saved plan, +reinvoking the Host/validator or activating legacy input for new work. Provider +effects still use the current ingest policy and original settlement owner; +cached content alone does not authorize a write. + PR dependency observation now uses the existing registered TurnStart lifecycle shared by executing quota guards, native Turns and compatibility prequota. Permission comes from the typed Goal/Agent capability owner, including remembered @@ -486,8 +518,8 @@ App/Lark's shared context backend consumes canonical readiness without polling. Plans, read-only/status/context paths, missing permission and recovery add no observation. Installed executor adoption, legitimate recovery of existing live owners, sustained scheduling, model outcomes and the three convergence -acceptances remain open. Generic memory lifecycle integration is a separate -remaining boundary; post-writeback hooks retain their effect-free contract. +acceptances remain open. Installed memory lifecycle adoption remains a separate +acceptance; post-writeback hooks retain their effect-free contract. The same captured decision now retains the selected Task's existing continuation condition and readiness through the full selection projection, signed envelope diff --git a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md index 6f12e2d792..40d96e997f 100644 --- a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md +++ b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md @@ -329,6 +329,27 @@ R5 短包投影也完整保留已有 CLI 结算计划，包括 effect identity envelope 投影，Codex CLI 保持关闭隔离；这不代表安装态 heartbeat/App 或模型收益 已验收。 +memory 参与现由能力层组合的选择后上下文 hook 注册接入 native Turn 和执行中的 +quota/heartbeat guard。Codex、DSH、generic CLI 传输相同的有界上下文、指导与可选 +结果字段；绑定和模式判断归能力，adapter 只保留宿主格式。TypeScript 限制贡献，拒绝 +替换权限／核心结果字段，复用既有 governing 签名文档及来源读回绑定上下文。只增加 +原捕获的定位信息不使未变决策失效；actor、权限、配置读回或贡献变化则丢弃可选上下文。 + +通用 Python 贡献 IO 与既有注册和 dispatcher 共同归属 +`loopx.control_plane.capability_hooks`，不再保留平行的 Python Turn-context 模块。 +校验与来源绑定继续调用 TypeScript decoder，宿主 adapter 消费同一份 hook 投影。 + +注册的 `PostSettlement` callback 在独立精确反思证明和可记账 writeback/spend 后调用 +既有 ingest owner，手动宿主的 sidecar adapter 也使用同一注册。`post_writeback` 继续 +无副作用；逐 hook 回执保留已完成工作，只恢复结果不确定的其他 hook，provider owner +继续负责写入去重与对账。终态重放保留原 journal 与回执，从原 owner 回读的恢复结果 +通过 `post_settlement_recovery` 返回，不重新结算。File/SQLite 原生宿主及隔离 provider 用例只资格源码路径； +安装态 App/Lark、模型收益、完整成本及以下三项收敛验收仍开放。 + +注册前 journal 的恢复保留已准入的有界反思及独立验证 digest，不改写原 plan、不重启 +Host／validator，也不为新工作激活旧输入。provider 效果仍由当前 ingest 策略及原结算 +owner 决定，缓存内容本身不授予写入权限。 + PR 依赖观察现复用既有注册 TurnStart 生命周期，由执行性 quota guard、原生 Turn 与兼容 prequota 共用。权限来自 typed Goal／Agent capability owner，包括已记住的 宿主观察与撤销；已有授权时不必逐命令补 flag。能力自有诊断使 PR 业务规则留在 core @@ -338,7 +359,7 @@ single-flight 与每项依赖的持久节奏避免并发／到期读取重复。 失败后的无关独立校验工作及原 journal 重放。App/Lark 共用上下文后端只消费 canonical 就绪状态，不轮询。plan、只读／status／上下文、缺少权限与恢复路径均不添加观察。 安装态执行器采用、既有真实 owner 的合法恢复、持续调度、模型结果与以下三批收敛验收 -仍开放。通用 memory 生命周期接入是另一项未完成边界；post-writeback hook 继续无效果。 +仍开放。安装态 memory 生命周期采用仍单独验收；post-writeback hook 继续无效果。 同次捕获决策中的所选 Task 继续条件与满足状态，现在沿完整选择投影、签名 envelope 到实际协议宿主输入完整保留。File/SQLite 验证区分不同仓库的同号 PR，保留原事件时间 diff --git a/docs/reference/protocols/turn-envelope-v0.md b/docs/reference/protocols/turn-envelope-v0.md index 073fe63d66..e2c9ec6b22 100644 --- a/docs/reference/protocols/turn-envelope-v0.md +++ b/docs/reference/protocols/turn-envelope-v0.md @@ -683,6 +683,61 @@ post-settlement ingest retain their existing rules. Provider failure remains fail-open. The host adapter is Python IO transport over the existing capability resolver and TypeScript envelope owner, not a new enablement policy. +The capability layer registers post-selection context once for executing native +Turn and quota/heartbeat guards. Internal `turn_context_contributions` carry +bounded context fields, instructions and optional text-result schemas; they are +not provider permissions. Hosts consume the generic `capability_context_fields`, +`capability_instructions` and `capability_result_fields` projection. The existing +`reward_memory_recall` readback and 2,400-character reflection limit remain. +Programmatic callers contributing memory use the registered context path; +inserting a legacy recall readback alone cannot activate a host contribution. +An inactive binding leaves ordinary host input, prompt and result schema intact. + +Recovery also reads pre-registration journals that already contain a bounded +reflection and its independent validation receipt. Without a contribution +annotation, only this cached result field retains its existing 2,400-character +wire decoding; new Host results still require registered context. Recovery +preserves the original result, validation digest and plan without repeating the +Host or task validator. Current ingest policy, independent evidence and the +original settlement/idempotency owners still decide any provider effect. +The cached reader and capability composition reuse the same verified automation +fact projection in `control_plane/reward_memory.py`; provider policy is not +copied into the executor. + +中文：恢复注册前的执行日志时，保留已准入的有界反思、原独立证明及原 plan；缺少 +贡献 annotation 的旧缓存仅沿用该结果字段原有的 2,400 字符解码，不重新启动 Host +或 validator，也不为新 Host 输入开启旧接线。provider 效果仍由当前 ingest 策略、 +独立证据及原结算／去重 owner 判定。 +缓存 reader 和能力组合复用 `control_plane/reward_memory.py` 的同一参与事实投影， +executor 不复制 provider 策略。 + +TypeScript binds each contribution to the existing governing signature document +and its source readback. Capture locations and size diagnostics may change; +actor, authority, binding readback or contribution changes discard that optional +context. Capabilities cannot overwrite core authority or result fields. +Ingest-only input omits recalled context. Host adapters preserve enabled optional +result values for outer bounds/material validation and independent attestation. +Registered `PostSettlement` callbacks invoke the original ingest/sidecar owners +after accountable settlement. `post_writeback` remains effect-free. Per-hook +receipts preserve completed callbacks during partial recovery; provider receipts +and idempotency remain with their existing owner. Unknown effects remain unknown. +After terminal settlement, a resume leaves the journal and its original +`post_settlement` receipt unchanged. When that receipt is uncertain, the owner +may reconcile the same effect and return `post_settlement_recovery` as current +readback. The owner retains the durable receipt; this projection grants no new +work, validation, quota debit or scheduler action. + +中文：执行中的 native Turn 与 quota/heartbeat guard 通过注册的选择后上下文路径接入 +记忆；三种宿主消费统一贡献，不再各自判断记忆模式。原 recall 读回和反思大小限制保留， +只有旧读回而无注册贡献不能开启宿主记忆输入。关闭保持普通输入／指导／schema 原样。 +TS 复用 governing 签名文档和来源绑定，允许捕获定位变化，拒绝 actor／权限／绑定／ +贡献变更及核心字段替换。独立精确证明和可记账结算后，注册的 `PostSettlement` 调用 +原 ingest owner；无副作用的 `post_writeback` 不变。逐 hook 回执与原 provider 去重 +共同恢复，不把不确定效果写成零。终态重放保留原 journal／`post_settlement`，通过原 +owner 对同一效果恢复的当前读回放入 `post_settlement_recovery`；持久回执仍由原 owner +保管，不触发业务工作、验证、扣额或调度。隔离 File/SQLite／宿主／provider 验证不代表安装态 +App/Lark、模型采用或真实服务资格。 + This is a bounded shared-facts step. Heartbeat still has its existing full and compact rendering paths; it is not yet the same installed host journey. The remaining convergence work is tracked in the existing diff --git a/loopx/capabilities/reward_memory/README.md b/loopx/capabilities/reward_memory/README.md index dbf8f94339..9d034ba57f 100644 --- a/loopx/capabilities/reward_memory/README.md +++ b/loopx/capabilities/reward_memory/README.md @@ -124,6 +124,16 @@ falls back to a shared corpus; provider degradation stays visible while base Goal work continues. Each host must report its actual coverage rather than generalizing from a direct CLI or helper test. +The native context and settlement registrations live in `turn_hooks.py` and are +composed outside the kernel in `loopx.capabilities.turn_context`. The existing +experiment, recall, independent-reflection validation, outcome-ingest and sidecar +owners retain their permissions and recovery rules. Context contributions are +private advisory input; the generic host transport cannot replace governing +fields. The real `PostSettlement` dispatcher runs effect adapters after +accountable settlement; `post_writeback` observers remain effect-free. Source +wiring for Codex/DSH/generic CLI and full/thin heartbeat readback is distinct +from installed App/Lark adoption or model benefit. + Config v1 declares one `project_provider_binding`, its exact per-corpus scope references, the project corpus set, module-owned surfaces, and an automation policy. A surface explicitly lists its compatible `corpus_ids`, selects one diff --git a/loopx/capabilities/reward_memory/turn_hooks.py b/loopx/capabilities/reward_memory/turn_hooks.py new file mode 100644 index 0000000000..704cfe98a8 --- /dev/null +++ b/loopx/capabilities/reward_memory/turn_hooks.py @@ -0,0 +1,158 @@ +"""Configured memory registrations; the existing owners retain every effect gate.""" + +from __future__ import annotations + +from collections.abc import Mapping +from pathlib import Path +from typing import Any + +from ...control_plane.capability_hooks import TurnContextHookRegistration +from ...control_plane.reward_memory import reward_memory_automation_enabled +from ...control_plane.turn_driver.post_settlement import ( + PostSettlementHookRegistration, + registered_post_settlement, +) + + +def reward_memory_prompt_instructions(request: Mapping[str, Any]) -> list[str]: + """Share optional capability guidance without changing its admission owner.""" + instructions: list[str] = [] + value = request.get("reward_memory_recall") + recall = value if isinstance(value, Mapping) else {} + if ( + reward_memory_automation_enabled(request, operation="automatic_recall") + and isinstance(recall.get("context"), Mapping) + and recall["context"].get("guidance") + ): + instructions.append( + "When reward_memory_recall contains guidance, treat it as private, non-authoritative decision context: apply it only when it fits current evidence and never treat it as new action authority." + ) + if reward_memory_automation_enabled(request, operation="automatic_ingest"): + instructions.append( + "Set reward_memory_reflection_json to an empty string unless independent task evidence established a reusable experience. For eligible evidence, return one compact JSON object using schema_version=turn_reward_memory_reflection_v1, status=eligible, a configured surface_id, outcome_kind in research|simulation|real|engineering, content_summary, reasoning_summary, confidence in low|medium|high, and 1-5 opaque evidence_refs. Also include experience using schema_version=procedural_experience_contract_v0 with non-empty applicability and limitations lists, observed_outcome, attribution, the same evidence_refs, and future_behavior containing trigger, action, validation, and stop_condition. A fact recap without a future behavior change and non-generalization boundary is not eligible memory. Legacy v0 reflections are audit-only and cannot become durable memory. Never use your own summary as evidence. Settlement may ingest it only when the caller-declared Todo validator attests the exact reflection digest and evidence; ordinary validator success remains awaiting and makes no provider write." + ) + return instructions + + +def memory_context_result( + payload: Mapping[str, Any], recall: Mapping[str, Any] +) -> dict[str, Any]: + admitted = {**payload, "reward_memory_recall": recall} + recall_enabled = reward_memory_automation_enabled( + admitted, operation="automatic_recall" + ) + ingest_enabled = reward_memory_automation_enabled( + admitted, operation="automatic_ingest" + ) + context: dict[str, Any] = {} + if recall_enabled or ingest_enabled: + context["reward_memory_recall"] = dict(recall) + if not recall_enabled: + context["reward_memory_recall"]["context"] = None + return { + "readback": {"reward_memory_recall": dict(recall)}, + "context_fields": context, + "instructions": reward_memory_prompt_instructions(admitted), + "result_fields": { + "reward_memory_reflection_json": {"type": "string", "maxLength": 2400} + } + if ingest_enabled + else {}, + } + + +def memory_context_registration( + payload: Mapping[str, Any], + *, + registry_path: Path, + goal_id: str, + agent_id: str, + decision: Mapping[str, Any], + turn_instance_id: str, +) -> TurnContextHookRegistration: + def produce() -> Mapping[str, Any]: + from ..agent_turn_recall import run_configured_agent_turn_recall_fail_open + + recall = run_configured_agent_turn_recall_fail_open( + registry_path=registry_path, + goal_id=goal_id, + agent_id=agent_id, + quota_decision=decision, + turn_instance_id=turn_instance_id, + execute=True, + ) + return memory_context_result(payload, recall) + + return TurnContextHookRegistration( + "reward-memory.turn-context", + "reward-memory", + ("reward_memory_recall",), + produce, + ) + + +def memory_settlement_registration( + payload: Mapping[str, Any], + *, + registry_path: Path, + goal_id: str, + agent_id: str, +) -> PostSettlementHookRegistration | None: + if not reward_memory_automation_enabled(payload, operation="automatic_ingest"): + return None + + def ingest( + plan: Mapping[str, Any], result: Mapping[str, Any], evidence: Mapping[str, Any] + ) -> Mapping[str, Any]: + from . import run_configured_turn_outcome_ingest_fail_open + + transaction = plan.get("transaction") + transaction = transaction if isinstance(transaction, Mapping) else {} + return run_configured_turn_outcome_ingest_fail_open( + registry_path=registry_path, + goal_id=goal_id, + agent_id=agent_id, + turn_key=str(transaction.get("turn_key") or ""), + host_result=result, + settlement_evidence=evidence, + ) + + return PostSettlementHookRegistration( + "reward-memory.outcome-ingest", "reward-memory", ingest + ) + + +def run_staged_memory_settlement(**kwargs: Any) -> dict[str, Any] | None: + """Adapt the manual-host sidecar owner after its canonical settlement readback.""" + from .codex_app_outcome import run_staged_codex_app_turn_outcome_ingest_fail_open + + observed: dict[str, Any] | None = None + + def ingest( + _plan: Mapping[str, Any], + _result: Mapping[str, Any], + _evidence: Mapping[str, Any], + ) -> Mapping[str, Any]: + nonlocal observed + observed = run_staged_codex_app_turn_outcome_ingest_fail_open(**kwargs) + return observed or { + "status": "not_applicable", + "external_writes_performed": False, + } + + callback = registered_post_settlement( + ( + PostSettlementHookRegistration( + "reward-memory.staged-outcome-ingest", + "reward-memory", + ingest, + ), + ) + ) + assert callback is not None + readback = callback({}, {}, {})["results"]["reward-memory.staged-outcome-ingest"] + return ( + observed + if observed is not None + else (None if readback.get("status") == "not_applicable" else dict(readback)) + ) diff --git a/loopx/capabilities/turn_context.py b/loopx/capabilities/turn_context.py new file mode 100644 index 0000000000..8655bdf0d4 --- /dev/null +++ b/loopx/capabilities/turn_context.py @@ -0,0 +1,59 @@ +"""Compose configured capability registrations outside the control plane.""" + +from __future__ import annotations + +from collections.abc import Mapping +from pathlib import Path +from typing import Any + +from ..control_plane.capability_hooks import dispatch_turn_context_hooks +from ..control_plane.turn_driver.post_settlement import ( + PostSettlement, + registered_post_settlement, +) +from .reward_memory.turn_hooks import ( + memory_context_registration, + memory_settlement_registration, +) + + +def attach_turn_context( + payload: dict[str, Any], + *, + registry_path: Path, + goal_id: str, + agent_id: str, + decision: Mapping[str, Any], + turn_instance_id: str, +) -> None: + dispatch_turn_context_hooks( + payload, + ( + memory_context_registration( + payload, + registry_path=registry_path, + goal_id=goal_id, + agent_id=agent_id, + decision=decision, + turn_instance_id=turn_instance_id, + ), + ), + ) + + +def configured_post_settlement( + payload: Mapping[str, Any], + *, + registry_path: Path, + goal_id: str, + agent_id: str, +) -> PostSettlement | None: + registration = memory_settlement_registration( + payload, + registry_path=registry_path, + goal_id=goal_id, + agent_id=agent_id, + ) + return registered_post_settlement( + (registration,) if registration is not None else () + ) diff --git a/loopx/cli_commands/quota_reward_memory.py b/loopx/cli_commands/quota_reward_memory.py index ef3535b677..ba593af7d6 100644 --- a/loopx/cli_commands/quota_reward_memory.py +++ b/loopx/cli_commands/quota_reward_memory.py @@ -6,13 +6,11 @@ from pathlib import Path from typing import Any -from ..capabilities.agent_turn_recall import ( - run_configured_agent_turn_recall_fail_open, -) +from ..capabilities.turn_context import attach_turn_context from ..capabilities.reward_memory.codex_app_outcome import ( stage_codex_app_turn_outcome_candidate_fail_open, - run_staged_codex_app_turn_outcome_ingest_fail_open, ) +from ..capabilities.reward_memory.turn_hooks import run_staged_memory_settlement from ..control_plane.quota.settlement import read_heartbeat_settlement @@ -103,7 +101,7 @@ def attach_reward_memory_ingest_after_spend( or readback.writeback.failure is not None or readback.spend.failure is not None): return result = ( - run_staged_codex_app_turn_outcome_ingest_fail_open( + run_staged_memory_settlement( registry_path=registry_path, goal_id=identity.goal_id, agent_id=identity.agent_id, @@ -139,11 +137,11 @@ def attach_reward_memory_recall_after_should_run( or heartbeat_receipt.get("status") not in {"committed", "replayed"} ): return - payload["reward_memory_recall"] = run_configured_agent_turn_recall_fail_open( + attach_turn_context( + payload, registry_path=registry_path, goal_id=goal_id, agent_id=agent_id, - quota_decision=payload, + decision=payload, turn_instance_id=heartbeat_turn_id, - execute=True, ) diff --git a/loopx/cli_commands/turn.py b/loopx/cli_commands/turn.py index 1bf4b3019e..a923599b9c 100644 --- a/loopx/cli_commands/turn.py +++ b/loopx/cli_commands/turn.py @@ -173,7 +173,6 @@ def handle_turn_command( # Only an executing wake may sync inboxes or reserve a calendar window. turn_start_hook_dispatch = {} if args.turn_command == "run-once" and args.execute: - from ..capabilities.agent_turn_recall import run_configured_agent_turn_recall_fail_open from ..capabilities.periodic_report.cadence_runtime import extend_cadence_turn_start_dispatch from .lark_inbox import dispatch_goal_lark_turn_start_hooks @@ -277,15 +276,11 @@ def handle_turn_command( or transaction.get("turn_key") or "" ) - payload["reward_memory_recall"] = ( - run_configured_agent_turn_recall_fail_open( - registry_path=registry_path, - goal_id=args.goal_id, - agent_id=args.agent_id, - quota_decision=decision, - turn_instance_id=turn_instance_id, - execute=True, - ) + from ..capabilities.turn_context import attach_turn_context + + attach_turn_context( + payload, registry_path=registry_path, goal_id=args.goal_id, + agent_id=args.agent_id, decision=decision, turn_instance_id=turn_instance_id, ) capability_action = payload.get("capability_action") if isinstance(capability_action, dict): diff --git a/loopx/cli_commands/turn_run_once.py b/loopx/cli_commands/turn_run_once.py index 2a958d2fe4..85838d72a4 100644 --- a/loopx/cli_commands/turn_run_once.py +++ b/loopx/cli_commands/turn_run_once.py @@ -9,9 +9,7 @@ from ..capabilities.explore.composition_frontier import ( project_live_explore_composition_frontier, ) -from ..capabilities.reward_memory import ( - run_configured_turn_outcome_ingest_fail_open, -) +from ..capabilities.turn_context import configured_post_settlement from ..cli_rollout import append_cli_rollout_event from ..control_plane.agents.workspace_guard import capture_delivery_workspace from ..control_plane.goals.first_party_host_admission import ( @@ -854,22 +852,6 @@ def resolve_dsh_operation_session(envelope: Mapping[str, Any]) -> dict[str, str] args, workspace=project, environ=operator_environ, ) - def post_settlement_reward_memory( - plan: Mapping[str, Any], - result: Mapping[str, Any], - settlement_evidence: Mapping[str, Any], - ) -> Mapping[str, Any]: - transaction = plan.get("transaction") - transaction = transaction if isinstance(transaction, Mapping) else {} - return run_configured_turn_outcome_ingest_fail_open( - registry_path=registry_path, - goal_id=args.goal_id, - agent_id=args.agent_id, - turn_key=str(transaction.get("turn_key") or ""), - host_result=result, - settlement_evidence=settlement_evidence, - ) - def on_managed_start_admitted() -> None: replan_packet = stable_envelope.get("replan_action_packet") capability_guard = ( @@ -921,11 +903,9 @@ def on_managed_start_admitted() -> None: terminal_closeout_resolver if args.execute else None ), scheduler=scheduler if args.execute else None, - post_settlement=( - post_settlement_reward_memory - if args.execute and args.host == "codex-cli" - else None - ), + post_settlement=(configured_post_settlement( + payload, registry_path=registry_path, goal_id=args.goal_id, agent_id=args.agent_id, + ) if args.execute else None), admit_start=managed_cadence.admit if args.execute else None, confirm_start=managed_cadence.confirm if args.execute else None, goal_admission=goal_admission, diff --git a/loopx/control_plane/capability_hooks.py b/loopx/control_plane/capability_hooks.py index 81747f8730..784c8b796b 100644 --- a/loopx/control_plane/capability_hooks.py +++ b/loopx/control_plane/capability_hooks.py @@ -76,6 +76,28 @@ class PostWritebackSourceRejected(ValueError): """The typed runtime rejected deterministic source construction.""" +@dataclass(frozen=True, slots=True) +class TurnContextHookRegistration: + """Configured provider IO after governing selection, before host input. + + Registration bounds IO/results; it grants no provider or task authority. + """ + + hook_id: str + capability_id: str + readback_slots: tuple[str, ...] + producer: Callable[[], Mapping[str, Any]] + max_result_bytes: int = 64 * 1024 + + def contract(self) -> dict[str, Any]: + return { + "hook_id": self.hook_id, + "capability_id": self.capability_id, + "readback_slots": list(self.readback_slots), + "max_result_bytes": self.max_result_bytes, + } + + @dataclass(frozen=True, slots=True) class InteractionProjectionHookRegistration: """One read-only capability contribution to an interaction contract.""" @@ -170,6 +192,134 @@ def contract(self) -> dict[str, Any]: } +def _turn_context_source(payload: Mapping[str, Any]) -> Mapping[str, Any]: + envelope = payload.get("turn_envelope") + return ( + envelope + if isinstance(envelope, Mapping) + else { + "goal_id": payload.get("goal_id"), + "agent_id": payload.get("agent_id"), + } + ) + + +def dispatch_turn_context_hooks( + payload: dict[str, Any], + registrations: Sequence[TurnContextHookRegistration], +) -> None: + contributions: list[dict[str, Any]] = [] + failures: list[dict[str, str]] = [] + seen_hooks: set[str] = set() + seen_slots: set[str] = set() + for registration in registrations: + try: + if registration.hook_id in seen_hooks or seen_slots.intersection( + registration.readback_slots + ): + raise ValueError("duplicate context registration") + seen_hooks.add(registration.hook_id) + seen_slots.update(registration.readback_slots) + effect_runtime_result( + "capability_hook.turn_context.validate_registration", + { + "registration": registration.contract(), + }, + ) + if any(key in payload for key in registration.readback_slots): + raise ValueError("context readback slot already contributed") + result = registration.producer() + contribution = effect_runtime_result( + "capability_hook.turn_context.validate", + { + "registration": registration.contract(), + "result": result, + "source": _turn_context_source(payload), + }, + ) + readback = contribution.pop("readback") + payload.update(readback) + if ( + contribution["context_fields"] + or contribution["instructions"] + or contribution["result_fields"] + ): + contributions.append(contribution) + except Exception: # Optional context failures never gate business work. + failures.append( + {"hook_id": registration.hook_id, "error_code": "context_hook_failed"} + ) + if contributions: + payload["turn_context_contributions"] = contributions + if failures: + payload["turn_context_hook_failures"] = failures + + +def project_turn_context(payload: Mapping[str, Any]) -> dict[str, Any]: + context: dict[str, Any] = {} + instructions: list[str] = [] + result_fields: dict[str, Any] = {} + contributions = payload.get("turn_context_contributions") + if not isinstance(contributions, list): + return {} + for item in contributions: + try: + slots = item["readback_slots"] + contribution = effect_runtime_result( + "capability_hook.turn_context.project", + { + "contribution": item, + "source": _turn_context_source(payload), + "readback": {key: payload.get(key) for key in slots}, + }, + ) + if set(context) & set(contribution["context_fields"]) or set( + result_fields + ) & set(contribution["result_fields"]): + raise ValueError("conflicting capability contribution") + context.update(contribution["context_fields"]) + instructions.extend(contribution["instructions"]) + result_fields.update(contribution["result_fields"]) + except ( + Exception + ): # A stale/private contribution is omitted, never repaired from cached data. + continue + if not (context or instructions or result_fields): + return {} + return { + **context, + "capability_context_fields": list(context), + "capability_instructions": instructions, + "capability_result_fields": result_fields, + } + + +def host_context_projection(request: Mapping[str, Any]) -> dict[str, Any]: + fields = request.get("capability_context_fields") or () + projection = {key: request[key] for key in fields if key in request} + if ( + projection + or request.get("capability_instructions") + or request.get("capability_result_fields") + ): + projection["capability_instructions"] = list( + request.get("capability_instructions") or () + ) + projection["capability_result_fields"] = dict( + request.get("capability_result_fields") or {} + ) + return projection + + +def capability_result_fields(request: Mapping[str, Any] | None) -> dict[str, Any]: + value = ( + request.get("capability_result_fields") + if isinstance(request, Mapping) + else None + ) + return dict(value) if isinstance(value, Mapping) else {} + + def dispatch_interaction_projection_hooks( registrations: Sequence[InteractionProjectionHookRegistration] | None, ) -> dict[str, Any]: diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index dbe7fe386b..192b6f88ae 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -593,6 +593,12 @@ export function createEffectRuntimeHandlers( "capability_hook.turn_start.validate_registration", lazyHandler(() => import("./capability_hooks.ts"), ({validateTurnStartHookRegistration}) => (params) => validateTurnStartHookRegistration(params.registration)), ], + ["capability_hook.turn_context.validate_registration", lazyHandler(() => import("./turn_driver/capability_context.ts"), + ({validateTurnContextRegistration}) => (params) => validateTurnContextRegistration(params.registration))], + ["capability_hook.turn_context.validate", lazyHandler(() => import("./turn_driver/capability_context.ts"), + ({validateTurnContextContribution}) => validateTurnContextContribution)], + ["capability_hook.turn_context.project", lazyHandler(() => import("./turn_driver/capability_context.ts"), + ({projectTurnContextContribution}) => projectTurnContextContribution)], [ "capability_hook.turn_start.validate", lazyHandler(() => import("./capability_hooks.ts"), ({validateTurnStartHookInvocation}) => (params) => validateTurnStartHookInvocation({ diff --git a/loopx/control_plane/reward_memory.py b/loopx/control_plane/reward_memory.py index ad1e0a261e..f6a3061bdf 100644 --- a/loopx/control_plane/reward_memory.py +++ b/loopx/control_plane/reward_memory.py @@ -6,8 +6,50 @@ from .todos.contract import normalize_todo_claimed_by +def reward_memory_automation_enabled( + payload: Mapping[str, Any] | None, + *, + operation: str, +) -> bool: + """Project verified capability automation into the host IO contract. + + The experiment resolver owns enablement, config/receipt verification and + provider policy. A recall packet's presence or old context is not opt-in. + """ + if not isinstance(payload, Mapping): + return False + recall = payload.get("reward_memory_recall") + if not isinstance(recall, Mapping): + return False + experiment = recall.get("experiment") + envelope = payload.get("turn_envelope") + if not isinstance(experiment, Mapping) or not isinstance(envelope, Mapping): + return False + boundary = envelope.get("boundary") + capabilities = ( + boundary.get("capabilities") if isinstance(boundary, Mapping) else None + ) + memory = ( + capabilities.get("reward_memory") if isinstance(capabilities, Mapping) else None + ) + return ( + isinstance(memory, Mapping) + and memory.get(operation) is True + # Recall resolves the binding again after admission. A stale envelope + # must not resurrect a now unavailable or disabled binding. + and experiment.get("enabled") is True + and experiment.get("available") is True + and experiment.get("configured_for_agent") is True + and bool(envelope.get("goal_id")) + and experiment.get("goal_id") == envelope.get("goal_id") + and bool(envelope.get("agent_id")) + and experiment.get("agent_id") == envelope.get("agent_id") + and experiment.get(operation) is True + ) + + def reward_memory_host_coverage() -> list[dict[str, str]]: - """Project actual host lifecycle wiring without inferring parity.""" + """Project source lifecycle wiring, not Agent enablement or provider qualification.""" return [ { @@ -18,12 +60,12 @@ def reward_memory_host_coverage() -> list[dict[str, str]]: { "host_id": "generic_cli_turn", "automatic_recall": "connected", - "automatic_ingest": "uncovered", + "automatic_ingest": "connected_post_settlement", }, { "host_id": "dsh_turn", - "automatic_recall": "connected_task_body", - "automatic_ingest": "uncovered", + "automatic_recall": "connected", + "automatic_ingest": "connected_post_settlement", }, { "host_id": "codex_app_quota", @@ -128,6 +170,7 @@ def reward_memory_goal_policy( __all__ = [ + "reward_memory_automation_enabled", "reward_memory_goal_policy", "reward_memory_host_coverage", ] diff --git a/loopx/control_plane/turn_driver/capability_context.ts b/loopx/control_plane/turn_driver/capability_context.ts new file mode 100644 index 0000000000..3c4e4bf5e5 --- /dev/null +++ b/loopx/control_plane/turn_driver/capability_context.ts @@ -0,0 +1,87 @@ +import type {JsonObject} from "../effect_program.ts"; +import {canonicalAuthoritySha256} from "../coordination/authority_store_codec.ts"; +import {requireJsonObject, requireNonEmptyString, requireStringArray} from "../runtime_decode.ts"; +import {turnEnvelopeActionSignatureDocument} from "../quota/turn_envelope.ts"; + +// This is a local host-IO contract, not a new capability or authority vocabulary. +// Capability providers own enablement. These rules only bound their contributions. +const RESERVED = new Set([ + "schema_version", "turn_key", "turn_envelope", "session", "goal_ref", "route", + "result_contract", "capability_context_fields", "capability_instructions", + "capability_result_fields", "result_kind", "completed_phases", "classification", + "next_action", "recommended_action", "summary", "agent_vision_json", + "path_delta_mode", "delivery_batch_scale", "delivery_outcome", "vision_unchanged_reason", + "work_evidence", "validation_evidence", "child_execution_receipts", + "primary_action", "required_reads", "write_scope", "workspace_guard", "boundary", + "work_context", "selected_todo", "unavailable_context", "agent_context", + "replan_action_packet", "detail_ref", "action_signature", "execution_profile", + "vision_continuation_audit", "goal_id", "agent_id", +]); +const FIELD = /^[a-z][a-z0-9_]{2,95}$/; + +function contextSource(value: unknown): JsonObject { + const source = requireJsonObject(value, "context source"); + // Reuse the governing owner. Binding the original observation adds detail + // locations/measurements without changing the authorized decision. + return {goal_id: source.goal_id, agent_id: source.agent_id, + governing: turnEnvelopeActionSignatureDocument(source)}; +} + +function fields(value: unknown, label: string): JsonObject { + const result = requireJsonObject(value, label); + if (Object.keys(result).length > 8 || Object.keys(result).some(k => !FIELD.test(k) || RESERVED.has(k))) { + throw new Error(`${label} may not replace host authority or core result fields`); + } + return result; +} + +export function validateTurnContextRegistration(input: unknown): JsonObject { + const registration = requireJsonObject(input, "context registration"); + const hookId = requireNonEmptyString(registration.hook_id, "hook_id"); + const capabilityId = requireNonEmptyString(registration.capability_id, "capability_id"); + if (![hookId, capabilityId].every(s => /^[a-z][a-z0-9_.:-]{2,95}$/.test(s))) throw new Error("invalid context identity"); + const slots = requireStringArray(registration.readback_slots, "context readback slots"); + fields(Object.fromEntries(slots.map(s => [s, null])), "context readback slots"); + if (new Set(slots).size !== slots.length) throw new Error("duplicate context readback slots"); + const maximum = registration.max_result_bytes; + if (!Number.isInteger(maximum) || Number(maximum) < 1024 || Number(maximum) > 65536) throw new Error("context budget exceeded"); + return {hook_id: hookId, capability_id: capabilityId, readback_slots: slots, max_result_bytes: maximum}; +} + +export function validateTurnContextContribution(input: JsonObject): JsonObject { + const registration = validateTurnContextRegistration(input.registration); + const result = requireJsonObject(input.result, "context result"); + if (Buffer.byteLength(JSON.stringify(result)) > Number(registration.max_result_bytes)) throw new Error("context budget exceeded"); + const context = fields(result.context_fields, "context fields"); + const readback = fields(result.readback, "context readback"); + const slots = requireStringArray(registration.readback_slots, "context readback slots"); + if (Object.keys(readback).length !== slots.length || Object.keys(readback).some(k => !slots.includes(k))) throw new Error("undeclared context readback"); + const instructions = requireStringArray(result.instructions, "context instructions"); + if (instructions.length > 8 || instructions.some(s => s.length > 4096)) throw new Error("context instructions exceeded"); + const resultFields = fields(result.result_fields, "capability result fields"); + for (const descriptor of Object.values(resultFields)) { + const schema = requireJsonObject(descriptor, "capability result schema"); + if (Object.keys(schema).length !== 2 || schema.type !== "string" || + !Number.isInteger(schema.maxLength) || Number(schema.maxLength) < 1 || Number(schema.maxLength) > 16384) { + throw new Error("capability result must be bounded optional text"); + } + } + const normalized = {hook_id: registration.hook_id, capability_id: registration.capability_id, + readback_slots: slots, context_fields: context, instructions, result_fields: resultFields, readback}; + return {...normalized, basis_digest: canonicalAuthoritySha256({source: contextSource(input.source), ...normalized})}; +} + +export function projectTurnContextContribution(input: JsonObject): JsonObject { + const contribution = requireJsonObject(input.contribution, "context contribution"); + const readback = requireJsonObject(input.readback, "current context readback"); + // Revalidate serialized contributions on journal recovery and before host input. + const normalized = validateTurnContextContribution({source: input.source, + registration: {hook_id: contribution.hook_id, capability_id: contribution.capability_id, + readback_slots: contribution.readback_slots, max_result_bytes: 65536}, + result: {context_fields: contribution.context_fields, instructions: contribution.instructions, + result_fields: contribution.result_fields, readback}}); + if (contribution.basis_digest !== normalized.basis_digest) { + throw new Error("context source or contribution changed after registration"); + } + return normalized; +} diff --git a/loopx/control_plane/turn_driver/codex_cli.py b/loopx/control_plane/turn_driver/codex_cli.py index ff042f2f0c..b7ada80c26 100644 --- a/loopx/control_plane/turn_driver/codex_cli.py +++ b/loopx/control_plane/turn_driver/codex_cli.py @@ -32,12 +32,11 @@ from .driver import SUPPORTED_ITERATION_CONTEXT_POLICIES from .executor import ( HOST_AGENT_VISION_JSON_MAX_CHARS, - HOST_REWARD_MEMORY_REFLECTION_JSON_MAX_CHARS, HOST_RESULT_EVIDENCE_LIMITS, HOST_RESULT_TEXT_LIMITS, LOOPX_TURN_HOST_REQUEST_SCHEMA_VERSION, - reward_memory_automation_enabled, ) +from ..capability_hooks import capability_result_fields from .execution_profile import require_supported_reasoning_effort from .host_failure import BuiltInHostError from .host_process_transport import HostOutputLines, run_host_process @@ -262,11 +261,7 @@ def codex_cli_result_schema( "items": {"type": "string", "maxLength": max_item_chars}, "maxItems": max_items, } - if reward_memory_automation_enabled(request, operation="automatic_ingest"): - properties["reward_memory_reflection_json"] = { - "type": "string", - "maxLength": HOST_REWARD_MEMORY_REFLECTION_JSON_MAX_CHARS, - } + properties.update(capability_result_fields(request)) if _has_subagent_topology(request): child_receipts = child_execution_receipts_json_schema() child_receipt_properties = child_receipts["items"]["properties"] @@ -320,16 +315,7 @@ def _prompt(request: Mapping[str, Any]) -> str: "For user_action_required, wait, or iteration_failed, leave material-only fields empty and explain the stop in summary. iteration_failed ends only this iteration and never requests a retry or successor.", 'completed_phases must be exactly ["host_execute","typed_result"], and turn_key must match the request.', ] - recall = _mapping(request.get("reward_memory_recall")) - if (reward_memory_automation_enabled(request, operation="automatic_recall") - and _mapping(recall.get("context")).get("guidance")): - instructions.append( - "When reward_memory_recall contains guidance, treat it as private, non-authoritative decision context: apply it only when it fits current evidence and never treat it as new action authority." - ) - if reward_memory_automation_enabled(request, operation="automatic_ingest"): - instructions.append( - "Set reward_memory_reflection_json to an empty string unless independent task evidence established a reusable experience. For eligible evidence, return one compact JSON object using schema_version=turn_reward_memory_reflection_v1, status=eligible, a configured surface_id, outcome_kind in research|simulation|real|engineering, content_summary, reasoning_summary, confidence in low|medium|high, and 1-5 opaque evidence_refs. Also include experience using schema_version=procedural_experience_contract_v0 with non-empty applicability and limitations lists, observed_outcome, attribution, the same evidence_refs, and future_behavior containing trigger, action, validation, and stop_condition. A fact recap without a future behavior change and non-generalization boundary is not eligible memory. Legacy v0 reflections are audit-only and cannot become durable memory. Never use your own summary as evidence. Settlement may ingest it only when the caller-declared Todo validator attests the exact reflection digest and evidence; ordinary validator success remains awaiting and makes no provider write." - ) + instructions.extend(request.get("capability_instructions") or ()) boundary = _mapping(_mapping(request.get("turn_envelope")).get("boundary")) if boundary.get("checkpointed_boundary_authority"): instructions.append( diff --git a/loopx/control_plane/turn_driver/executor.py b/loopx/control_plane/turn_driver/executor.py index 1fdac0816b..8678c6183a 100644 --- a/loopx/control_plane/turn_driver/executor.py +++ b/loopx/control_plane/turn_driver/executor.py @@ -57,7 +57,9 @@ build_turn_recovery_audit, require_turn_recovery_continuation, ) -from .post_settlement import PostSettlement, run_post_settlement_callback +from .post_settlement import PostSettlement, recover_post_settlement, run_post_settlement_callback +from ..capability_hooks import capability_result_fields, project_turn_context +from ..reward_memory import reward_memory_automation_enabled from .session_recovery import ( SessionBindingResolver, build_host_recovery_record, @@ -142,41 +144,6 @@ JournalPersist = Callable[[Mapping[str, Any]], None] -def reward_memory_automation_enabled( - payload: Mapping[str, Any] | None, *, operation: str, -) -> bool: - """Project verified capability automation into the host IO contract. - - The experiment resolver owns enablement, config/receipt verification and - provider policy. A recall packet's presence or old context is not opt-in. - """ - if not isinstance(payload, Mapping): - return False - recall = payload.get("reward_memory_recall") - if not isinstance(recall, Mapping): - return False - experiment = recall.get("experiment") - envelope = payload.get("turn_envelope") - if not isinstance(experiment, Mapping) or not isinstance(envelope, Mapping): - return False - boundary = envelope.get("boundary") - capabilities = boundary.get("capabilities") if isinstance(boundary, Mapping) else None - memory = capabilities.get("reward_memory") if isinstance(capabilities, Mapping) else None - return ( - isinstance(memory, Mapping) - and memory.get(operation) is True - # Recall resolves the binding again after admission. A stale envelope - # must not resurrect a now unavailable or disabled binding. - and experiment.get("enabled") is True - and experiment.get("available") is True - and experiment.get("configured_for_agent") is True - and bool(envelope.get("goal_id")) - and experiment.get("goal_id") == envelope.get("goal_id") - and bool(envelope.get("agent_id")) - and experiment.get("agent_id") == envelope.get("agent_id") - and experiment.get(operation) is True - ) - def build_loopx_turn_host_request(plan: Mapping[str, Any]) -> dict[str, Any]: transaction = ( @@ -203,13 +170,7 @@ def build_loopx_turn_host_request(plan: Mapping[str, Any]) -> dict[str, Any]: goal_ref = plan.get("goal_ref") if isinstance(goal_ref, Mapping): request["goal_ref"] = dict(goal_ref) - reward_memory_recall = plan.get("reward_memory_recall") - recall_enabled = reward_memory_automation_enabled(plan, operation="automatic_recall") - ingest_enabled = reward_memory_automation_enabled(plan, operation="automatic_ingest") - if isinstance(reward_memory_recall, Mapping) and (recall_enabled or ingest_enabled): - request["reward_memory_recall"] = dict(reward_memory_recall) - if not recall_enabled: - request["reward_memory_recall"]["context"] = None + request.update(project_turn_context(plan)) request.update(subagent.subagent_host_request_projection(plan)) from ...extensions.codex_native_child import configured_native_child_limit @@ -403,7 +364,19 @@ def validate_loopx_turn_host_result( ) -> dict[str, Any]: result = dict(value) errors: list[str] = [] - allowed_fields = HOST_RESULT_FIELDS | subagent.subagent_host_result_fields(plan) + optional_fields = capability_result_fields(project_turn_context(plan)) + if ( + from_journal + and "turn_context_contributions" not in plan + and reward_memory_automation_enabled(plan, operation="automatic_ingest") + ): + # Pre-registration journals already admitted this bounded wire field. + # Preserve the cached result that independent validation attested; + # never reconstruct context, activate a Host, or change ingest policy. + optional_fields["reward_memory_reflection_json"] = { + "type": "string", "maxLength": HOST_REWARD_MEMORY_REFLECTION_JSON_MAX_CHARS, + } + allowed_fields = HOST_RESULT_FIELDS | subagent.subagent_host_result_fields(plan) | set(optional_fields) if from_journal: allowed_fields = allowed_fields | {"agent_vision"} unknown = sorted(set(result) - allowed_fields) @@ -458,24 +431,15 @@ def validate_loopx_turn_host_result( normalized[field] = text # Old/custom hosts may still return this field. Ignore it while disabled: # ordinary work must not gain a memory validation gate or retain the content. - reflection_json = ( - _bounded_public_text( - result, - "reward_memory_reflection_json", - limit=HOST_REWARD_MEMORY_REFLECTION_JSON_MAX_CHARS, - required=False, - errors=errors, + for field, descriptor in optional_fields.items(): + text = _bounded_public_text( + result, field, limit=descriptor["maxLength"], required=False, errors=errors, ) - if reward_memory_automation_enabled(plan, operation="automatic_ingest") - else None - ) - if reflection_json: - if material: - normalized["reward_memory_reflection_json"] = reflection_json - else: - errors.append( - "non-material host results cannot declare a reward memory reflection" - ) + if text: + if material: + normalized[field] = text + else: + errors.append("non-material host results cannot declare optional capability results") if material: try: normalized["delivery_batch_scale"] = require_delivery_batch_scale( @@ -1593,6 +1557,9 @@ def persist_source_journal(source_admission: Mapping[str, Any]) -> None: replayed=True, effects=empty_effects, ) + payload.update(recover_post_settlement( + plan=plan, post_settlement=post_settlement, journal=journal, + )) payload["recovery"] = build_turn_recovery_audit( recovery_decision, journal, diff --git a/loopx/control_plane/turn_driver/host_candidate.py b/loopx/control_plane/turn_driver/host_candidate.py index 8300562198..fbaf2c9c50 100644 --- a/loopx/control_plane/turn_driver/host_candidate.py +++ b/loopx/control_plane/turn_driver/host_candidate.py @@ -15,7 +15,10 @@ from ..work_items.delivery_batch_scale import DELIVERY_BATCH_SCALE_CHOICES from ..work_items.delivery_outcome import DELIVERY_OUTCOME_CHOICES from .driver import selected_turn_todo -from .executor import HOST_RESULT_EVIDENCE_LIMITS +from ..capability_hooks import capability_result_fields, host_context_projection +from .executor import ( + HOST_RESULT_EVIDENCE_LIMITS, +) LOOPX_TURN_HOST_REQUEST_SCHEMA = "loopx_turn_host_request_v0" LOOPX_TURN_RESULT_SCHEMA = "loopx_turn_result_v0" @@ -134,6 +137,7 @@ def extract_turn_authority(request: Mapping[str, Any]) -> dict[str, Any]: # its binding; this adapter neither reads it nor re-enters admission. authority["detail_ref"] = _mapping(envelope.get("detail_ref")) authority["action_signature"] = signature + authority.update(host_context_projection(request)) return authority @@ -409,6 +413,10 @@ def build_result( # Preserve the structured executor input verbatim. The executor # owns type, size, JSON and semantic validation. result[field] = candidate[field] + for field in capability_result_fields(request): + if field in candidate: + # Preserve optional values verbatim for the owning validator. + result[field] = candidate[field] for field in HOST_RESULT_EVIDENCE_LIMITS: if field in candidate: # Preserve the bounded evidence value for the executor's typed diff --git a/loopx/control_plane/turn_driver/post_settlement.py b/loopx/control_plane/turn_driver/post_settlement.py index 34458715c5..a60f538d8d 100644 --- a/loopx/control_plane/turn_driver/post_settlement.py +++ b/loopx/control_plane/turn_driver/post_settlement.py @@ -1,8 +1,9 @@ -"""Optional fail-open observers that run after accountable Turn settlement.""" +"""Optional capability effects that run after accountable Turn settlement.""" from __future__ import annotations from collections.abc import Callable, Mapping +from dataclasses import dataclass from typing import Any @@ -12,24 +13,102 @@ ] -def run_post_settlement_callback( +@dataclass(frozen=True, slots=True) +class PostSettlementHookRegistration: + """One optional capability IO adapter using the existing settlement owner.""" + + hook_id: str + capability_id: str + callback: PostSettlement + + +def registered_post_settlement( + registrations: tuple[PostSettlementHookRegistration, ...], +) -> PostSettlement | None: + if not registrations: + return None + if len({item.hook_id for item in registrations}) != len(registrations): + raise ValueError("duplicate post-settlement registration") + + def dispatch( + plan: Mapping[str, Any], result: Mapping[str, Any], evidence: Mapping[str, Any] + ) -> Mapping[str, Any]: + previous = evidence.get("previous_post_settlement") + previous = previous if isinstance(previous, Mapping) else {} + receipts = dict(previous.get("results") or {}) + identities = dict(previous.get("capability_ids") or {}) + pending = False + for registration in registrations: + prior = receipts.get(registration.hook_id) + if ( + identities.get(registration.hook_id) == registration.capability_id + and isinstance(prior, Mapping) + and prior.get("status") not in _RETRYABLE + ): + continue + try: + receipt = dict(registration.callback(plan, result, evidence)) + except ( + Exception + ): # The capability owner recovers ambiguous provider effects. + receipt = { + "status": "runtime_unavailable", + "fail_open": True, + "external_writes_performed": None, + } + receipts[registration.hook_id] = receipt + identities[registration.hook_id] = registration.capability_id + pending = pending or receipt.get("status") in _RETRYABLE + return { + "status": "runtime_unavailable" if pending else "observed", + "results": receipts, + "capability_ids": identities, + "fail_open": True, + "external_writes_performed": ( + True + if any( + item.get("external_writes_performed") is True + for item in receipts.values() + ) + else None + if any( + item.get("external_writes_performed") is None + for item in receipts.values() + ) + else False + ), + } + + return dispatch + + +_RETRYABLE = frozenset( + { + "provider_unavailable", + "committed_pending", + "readback_unverified", + "runtime_unavailable", + } +) + + +def post_settlement_receipt( *, plan: Mapping[str, Any], result: Mapping[str, Any], post_settlement: PostSettlement | None, - journal: dict[str, Any], - persist_journal: Callable[[Mapping[str, Any]], None], -) -> None: + journal: Mapping[str, Any], +) -> dict[str, Any] | None: + """Read/reconcile through capability owners without changing Turn state. + + Owners retain durable effect identity and retry policy. A terminal Turn's + original receipt remains immutable even when an owner recovers a lost reply. + """ if post_settlement is None: - return + return None existing = journal.get("post_settlement") - if isinstance(existing, Mapping) and existing.get("status") not in { - "provider_unavailable", - "committed_pending", - "readback_unverified", - "runtime_unavailable", - }: - return + if isinstance(existing, Mapping) and existing.get("status") not in _RETRYABLE: + return None try: receipt = post_settlement( plan, @@ -39,18 +118,61 @@ def run_post_settlement_callback( "task_validation": dict(journal.get("task_validation") or {}), "writeback": dict(journal.get("writeback") or {}), "quota_spend": dict(journal.get("quota_spend") or {}), + "previous_post_settlement": dict(existing or {}), }, ) if not isinstance(receipt, Mapping): raise TypeError("post-settlement callback must return an object") - journal["post_settlement"] = dict(receipt) + return dict(receipt) except Exception: # noqa: BLE001 - optional learning never gates the Turn - journal["post_settlement"] = { + return { "ok": True, "schema_version": "turn_post_settlement_observer_v0", "status": "runtime_unavailable", "reason_code": "post_settlement_observer_failed", "fail_open": True, - "external_writes_performed": False, + "external_writes_performed": None, } - persist_journal(journal) + + +def run_post_settlement_callback( + *, + plan: Mapping[str, Any], + result: Mapping[str, Any], + post_settlement: PostSettlement | None, + journal: dict[str, Any], + persist_journal: Callable[[Mapping[str, Any]], None], +) -> None: + receipt = post_settlement_receipt( + plan=plan, + result=result, + post_settlement=post_settlement, + journal=journal, + ) + if receipt is not None: + journal["post_settlement"] = receipt + persist_journal(journal) + + +def recover_post_settlement( + *, + plan: Mapping[str, Any], + post_settlement: PostSettlement | None, + journal: Mapping[str, Any], +) -> dict[str, Any]: + """Project current owner readback while preserving the terminal tombstone. + + This path has no journal writer or business-settlement adapter. Recovery + uses the original result and capability effect identity, never a new host. + """ + if journal.get("status") != "committed" or not isinstance( + journal.get("post_settlement"), Mapping + ): + return {} + receipt = post_settlement_receipt( + plan=plan, + result=dict(journal.get("host_result") or {}), + post_settlement=post_settlement, + journal=journal, + ) + return {"post_settlement_recovery": receipt} if receipt is not None else {} diff --git a/tests/control_plane/reward_memory_host_fixture.py b/tests/control_plane/reward_memory_host_fixture.py index 9cfe6c4bfe..68ff8ec954 100644 --- a/tests/control_plane/reward_memory_host_fixture.py +++ b/tests/control_plane/reward_memory_host_fixture.py @@ -17,9 +17,26 @@ def enable_plan_memory(plan: dict) -> None: envelope.setdefault("boundary", {})["capabilities"] = {"reward_memory": { "automatic_recall": True, "automatic_ingest": True, }} + register_plan_memory(plan) -def enable_live_memory(registry: Path, *, recall: bool = True, ingest: bool = True) -> Path: +def register_plan_memory(plan: dict) -> None: + """Exercise the actual registered context dispatcher over an explicit fixture.""" + from loopx.control_plane.capability_hooks import ( + TurnContextHookRegistration, + dispatch_turn_context_hooks, + ) + from loopx.capabilities.reward_memory.turn_hooks import memory_context_result + + recall = plan.pop("reward_memory_recall") + plan.pop("turn_context_contributions", None) + result = memory_context_result(plan, recall) + dispatch_turn_context_hooks(plan, (TurnContextHookRegistration( + "reward-memory.turn-context", "reward-memory", ("reward_memory_recall",), lambda: result, + ),)) + + +def enable_live_memory(registry: Path, *, recall: bool = True, ingest: bool = True, validated_ingest: bool = False) -> Path: # Synthetic pre-verified binding in a disposable project. Runtime resolution # still checks the exact digest, actor scope and receipt; no live promotion. from tests.capabilities.test_agent_turn_recall import raw_config @@ -29,6 +46,11 @@ def enable_live_memory(registry: Path, *, recall: bool = True, ingest: bool = Tr agent = goal["coordination"]["registered_agents"][0] config = raw_config(peer_ref=f"agent:{agent}", goal_id=goal["id"], agent_id=agent) config["automation"].update(automatic_recall=recall, automatic_ingest=ingest) + if validated_ingest: + entry = config["corpora"][0] + entry["standing_policy"]["allowed_actor_roles"] = ["validated_goal_agent"] + entry["standing_policy"]["allowed_source_kinds"] = ["engineering_review"] + entry["corpus"]["maintenance"]["writeback_triggers"] = ["engineering_review"] config["project_provider_binding"]["provider_binary"] = "fixture-unavailable-memory-provider" path = Path(goal["repo"]) / ".loopx/config/reward-memory/fixture.json" path.parent.mkdir(parents=True, exist_ok=True) diff --git a/tests/control_plane/test_host_bootstrap_lifecycle.py b/tests/control_plane/test_host_bootstrap_lifecycle.py index 119edd2800..d58127e0ad 100644 --- a/tests/control_plane/test_host_bootstrap_lifecycle.py +++ b/tests/control_plane/test_host_bootstrap_lifecycle.py @@ -188,7 +188,7 @@ def test_app_brief_with_registry_profile_keeps_budget_and_current_settlement(reg body = packet["task_body"] assert all(scope.rstrip(".!?") in body for scope in scopes) assert packet["agent_scope_source"] == "agent_profile_v1" - assert packet["interface_budget"]["max_chars"] == 4800 + assert packet["interface_budget"]["max_chars"] == 5400 assert packet["interface_budget"]["within_budget"], packet["interface_budget"] assert packet["cli_preflight"] in body assert "--codex-app" in body diff --git a/tests/control_plane/test_post_writeback_wait_cli.py b/tests/control_plane/test_post_writeback_wait_cli.py index 33123c91de..58733c1480 100644 --- a/tests/control_plane/test_post_writeback_wait_cli.py +++ b/tests/control_plane/test_post_writeback_wait_cli.py @@ -47,6 +47,12 @@ def test_committed_progress_before_wait_recovers_only_original_spend( cycle: bool, ) -> None: isolate_sqlite_runtime(tmp_path, monkeypatch) + # Compare complete Todo readbacks at one observation time. Elapsed wall + # time changes the derived wait projection without mutating the Todo. + monkeypatch.setattr( + "loopx.control_plane.todos.resume_condition.now_utc_iso", + lambda: "2026-09-20T00:00:00Z", + ) project, runtime, registry = _write_fixture(tmp_path) state = _configure_read_only_todo(project) _configure_selectable_alternative(project) diff --git a/tests/control_plane/test_reward_memory_host_isolation.py b/tests/control_plane/test_reward_memory_host_isolation.py index a93bf7e544..6878ab0b92 100644 --- a/tests/control_plane/test_reward_memory_host_isolation.py +++ b/tests/control_plane/test_reward_memory_host_isolation.py @@ -14,6 +14,10 @@ from loopx.control_plane.turn_driver import build_loopx_turn_host_request, build_loopx_turn_plan from loopx.control_plane.turn_driver.codex_cli import _prompt, codex_cli_result_schema from loopx.control_plane.turn_driver.executor import validate_loopx_turn_host_result +from loopx.control_plane.quota.turn_envelope import turn_envelope_action_signature_document +from loopx.control_plane.turn_driver.host_candidate import ( + _canonical_hash, build_result, extract_turn_authority, render_prompt, +) from test_loopx_turn_driver import _envelope @@ -38,6 +42,8 @@ def _plan(*, recall: bool = True, ingest: bool = True) -> dict: plan["turn_envelope"]["boundary"] = {"capabilities": {"reward_memory": { "automatic_recall": recall, "automatic_ingest": ingest, }}} + from tests.control_plane.reward_memory_host_fixture import register_plan_memory + register_plan_memory(plan) return plan @@ -70,10 +76,10 @@ def test_admitted_memory_does_not_resurrect_invalidated_runtime_binding(invalida experiment = plan["reward_memory_recall"]["experiment"] experiment[invalidated] = "another-actor" if invalidated.endswith("_id") else False request = build_loopx_turn_host_request(plan) - valid_binding = invalidated.startswith("automatic_") + valid_binding = False # Changed registered sources invalidate the entire optional contribution. assert ("reward_memory_recall" in request) is valid_binding - assert ("reward_memory_reflection_json" in codex_cli_result_schema(request)["properties"]) is (invalidated == "automatic_recall") - assert ("When reward_memory_recall contains guidance" in _prompt(request)) is (invalidated == "automatic_ingest") + assert ("reward_memory_reflection_json" in codex_cli_result_schema(request)["properties"]) is False + assert ("When reward_memory_recall contains guidance" in _prompt(request)) is False @pytest.mark.parametrize(("recall", "ingest"), [(True, False), (False, True), (True, True)]) @@ -93,6 +99,8 @@ def test_recall_and_reflection_are_independently_opted_in(recall: bool, ingest: def test_unavailable_provider_does_not_instruct_use_of_missing_recall_context() -> None: plan = _plan() plan["reward_memory_recall"].update(status="provider_unavailable", context=None) + from tests.control_plane.reward_memory_host_fixture import register_plan_memory + register_plan_memory(plan) request = build_loopx_turn_host_request(plan) assert "When reward_memory_recall contains guidance" not in _prompt(request) # Recall failure does not silently disable independently configured outcome ingest. @@ -100,7 +108,8 @@ def test_unavailable_provider_does_not_instruct_use_of_missing_recall_context() @pytest.mark.parametrize("ingest", [False, True]) -def test_old_host_reflection_is_ignored_when_ingest_is_off(ingest: bool) -> None: +@pytest.mark.parametrize("from_journal", [False, True]) +def test_old_host_reflection_is_ignored_when_ingest_is_off(ingest: bool, from_journal: bool) -> None: plan = _plan(recall=False, ingest=ingest) result = { "schema_version": "loopx_turn_result_v0", @@ -115,14 +124,207 @@ def test_old_host_reflection_is_ignored_when_ingest_is_off(ingest: bool) -> None "vision_unchanged_reason": "The original fixture remains open.", "reward_memory_reflection_json": '{"status":"no_evidence"}', } - validated = validate_loopx_turn_host_result(plan, result) + validated = validate_loopx_turn_host_result(plan, result, from_journal=from_journal) assert validated["ok"], validated assert ("reward_memory_reflection_json" in validated["result"]) is ingest +def test_pre_registration_journal_preserves_validated_reflection_on_recovery( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + from loopx.control_plane.turn_driver import executor + from loopx.control_plane.turn_driver.journal_store import ( + load_loopx_turn_plan_from_journal, turn_journal_path, + ) + from tests.capabilities.test_reward_memory_outcome_lifecycle import _reflection + + plan = _plan(recall=False) + reflection = _reflection() + attestation = { + "schema_version": "reward_memory_reflection_validation_v0", + "status": "validated", + "reflection_digest": executor.reward_memory_reflection_digest(reflection), + "evidence_refs": json.loads(reflection)["evidence_refs"], + } + result = { + "schema_version": "loopx_turn_result_v0", + "turn_key": plan["transaction"]["turn_key"], + "result_kind": "validated_progress", + "completed_phases": ["host_execute", "typed_result"], + "classification": "fixture_progress", "recommended_action": "Continue", + "next_action": "Continue", "delivery_batch_scale": "single_surface", + "delivery_outcome": "outcome_progress", + "vision_unchanged_reason": "The original objective remains open.", + "reward_memory_reflection_json": reflection, + } + counts = dict(host=0, validation=0, writeback=0, spend=0, post=0) + + def host(_request): + counts["host"] += 1 + return result + + def validate(_plan, candidate): + counts["validation"] += 1 + assert candidate["reward_memory_reflection_json"] == reflection + return {"status": "passed", "validator_kind": "fixture-independent-validator", + "summary": "Exact reflection and evidence verified.", + "reward_memory_reflection_validation": attestation} + + def writeback(candidate): + counts["writeback"] += 1 + assert candidate["reward_memory_reflection_json"] == reflection + return {"ok": True, "appended": True} + + def spend(): + counts["spend"] += 1 + return {"ok": True, "appended": True, "slots": 1} + + def post(_plan, candidate, evidence): + counts["post"] += 1 + assert candidate["reward_memory_reflection_json"] == reflection + assert evidence["task_validation"]["reward_memory_reflection_validation"] == attestation + return {"ok": True, "status": "fixture_observed", "external_writes_performed": False} + + class InterruptedAfterValidation(BaseException): + pass + + persist = executor._write_journal + + def interrupt(path, journal, **kwargs): + # Persist the producer's pre-registration shape from its first write; + # never rewrite the native journal owner's immutable saved plan. + legacy = copy.deepcopy(journal) + legacy["plan"].pop("turn_context_contributions", None) + persist(path, legacy, **kwargs) + if journal.get("completed_phases") == ["host_execute", "typed_result", "validation"]: + raise InterruptedAfterValidation + + options = dict(host_runner=host, task_validator=validate, writeback=writeback, + spend=spend, scheduler=lambda _: {"completed": True}, + post_settlement=post, project=tmp_path, runtime_root=tmp_path / "runtime", + goal_id="fixture-goal", timeout_seconds=10, execute=True) + with monkeypatch.context() as patch: + patch.setattr(executor, "_write_journal", interrupt) + with pytest.raises(InterruptedAfterValidation): + executor.run_loopx_turn_once(plan, **options) + + path = turn_journal_path(options["runtime_root"], goal_id="fixture-goal", + turn_key=result["turn_key"]) + journal = json.loads(path.read_text()) + # The pre-registration producer persisted this same result/attestation and + # verified participation, but no generic contribution annotation. This is a + # synthetic compatibility input, not a new activation or provider receipt. + assert "turn_context_contributions" not in journal["plan"] + original_plan = copy.deepcopy(journal["plan"]) + saved_plan = load_loopx_turn_plan_from_journal( + options["runtime_root"], goal_id="fixture-goal", turn_key=result["turn_key"], + ) + recovered = executor.run_loopx_turn_once(saved_plan, **options) + assert recovered["status"] == "committed", recovered + final = json.loads(path.read_text()) + assert final["host_result"]["reward_memory_reflection_json"] == reflection + assert final["task_validation"]["reward_memory_reflection_validation"] == attestation + assert final["plan"] == original_plan + before_replay = path.read_bytes() + replay = executor.run_loopx_turn_once(saved_plan, **options) + assert replay["replayed"] is True + assert path.read_bytes() == before_replay + assert counts == dict(host=1, validation=1, writeback=1, spend=1, post=1) + + +@pytest.mark.parametrize(("reflection", "kind", "valid"), [ + (None, "validated_progress", True), + ("x" * 2400, "validated_progress", True), + ("x" * 2401, "validated_progress", False), + ({"wrong": "type"}, "validated_progress", False), + ("bounded reflection", "wait", False), +]) +def test_legacy_cached_reflection_is_bounded_and_cannot_activate_a_host(reflection, kind, valid): + plan = _plan(recall=False) + plan.pop("turn_context_contributions") + before = copy.deepcopy(plan) + request = build_loopx_turn_host_request(plan) + assert "reward_memory_recall" not in request + assert "reward_memory_reflection_json" not in codex_cli_result_schema(request)["properties"] + assert "Set reward_memory_reflection_json" not in _prompt(request) + assert "When reward_memory_recall contains guidance" not in _prompt(request) + result = { + "schema_version": "loopx_turn_result_v0", "turn_key": plan["transaction"]["turn_key"], + "result_kind": kind, "completed_phases": ["host_execute", "typed_result"], + "classification": "fixture_progress", "recommended_action": "Continue", + "next_action": "Continue", "delivery_batch_scale": "single_surface", + "delivery_outcome": "outcome_progress", + "vision_unchanged_reason": "The original objective remains open.", + "reward_memory_reflection_json": reflection, + } + fresh = validate_loopx_turn_host_result(plan, result) + assert fresh["ok"], fresh + assert "reward_memory_reflection_json" not in fresh["result"] + cached = validate_loopx_turn_host_result(plan, result, from_journal=True) + assert cached["ok"] is valid, cached + if valid: + assert cached["result"].get("reward_memory_reflection_json") == reflection + # An invalidated registered contribution must not fall back to legacy input. + plan["turn_context_contributions"] = [] + current = validate_loopx_turn_host_result(plan, result, from_journal=True) + assert current["ok"], current + assert "reward_memory_reflection_json" not in current["result"] + plan.pop("turn_context_contributions") + assert plan == before + + +@pytest.mark.parametrize(("recall", "ingest"), [(False, False), (True, False), (False, True), (True, True)]) +@pytest.mark.parametrize("invalidated", [None, "enabled", "available", "configured_for_agent", "goal_id", "agent_id"]) +def test_shared_host_memory_keeps_binding_context_and_validation(recall, ingest, invalidated): + plan = _plan(recall=recall, ingest=ingest) + envelope = plan["turn_envelope"] + envelope["action"]["primary_action"] = "Advance one public fixture" + digest = _canonical_hash(turn_envelope_action_signature_document(envelope)) + envelope["action_signature"].update(source_hash=digest, envelope_hash=digest) + from tests.control_plane.reward_memory_host_fixture import register_plan_memory + register_plan_memory(plan) + if invalidated: + plan["reward_memory_recall"]["experiment"][invalidated] = "other" if invalidated.endswith("_id") else False + request = build_loopx_turn_host_request(plan) + before = copy.deepcopy(request) + authority = extract_turn_authority(request) + prompt = render_prompt(authority) + active = invalidated is None + assert ("When reward_memory_recall contains guidance" in prompt) is (active and recall) + assert ("Set reward_memory_reflection_json" in prompt) is (active and ingest) + if active and recall: + assert authority["reward_memory_recall"]["context"] == plan["reward_memory_recall"]["context"] + elif "reward_memory_recall" in authority: + assert authority["reward_memory_recall"]["context"] is None + if not active or not (recall or ingest): + ordinary = {k: v for k, v in request.items() if k != "reward_memory_recall"} + assert authority == extract_turn_authority(ordinary) + assert prompt == render_prompt(extract_turn_authority(ordinary)) + candidate = {"result_kind": "validated_progress", "next_action": "Continue the fixture", + "reward_memory_reflection_json": '{"status":"no_evidence"}'} + result = build_result(request, candidate) + assert ("reward_memory_reflection_json" in result) is (active and ingest) + assert validate_loopx_turn_host_result(plan, result)["ok"] + if active and ingest: + # Preserve the input for the owning validators, without coercion or + # truncation in this adapter. Oversize and stop-result rules still hold. + candidate["reward_memory_reflection_json"] = {"wrong": "type"} + assert build_result(request, candidate)["reward_memory_reflection_json"] == {"wrong": "type"} + candidate["reward_memory_reflection_json"] = "x" * 2401 + assert not validate_loopx_turn_host_result(plan, build_result(request, candidate))["ok"] + candidate.update(result_kind="wait", reward_memory_reflection_json='{"status":"no_evidence"}') + assert not validate_loopx_turn_host_result(plan, build_result(request, candidate))["ok"] + invalid = copy.deepcopy(request) + invalid["turn_envelope"]["boundary"]["capabilities"]["reward_memory"]["automatic_ingest"] = not ingest + with pytest.raises(ValueError, match="signature"): + extract_turn_authority(invalid) + assert request == before + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) -@pytest.mark.parametrize("mode", ["unconfigured", "disabled", "stale", "both_off", "recall_only", "ingest_only", "both_on"]) -def test_real_turn_cli_memory_projection_and_settlement(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, mode: str) -> None: +@pytest.mark.parametrize("host_kind", ["codex-cli", "dsh", "generic-cli"]) +@pytest.mark.parametrize("mode", ["unconfigured", "disabled", "stale", "both_off", "recall_only", "recall_context", "ingest_only", "both_on"]) +def test_real_turn_cli_memory_projection_and_settlement(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, host_kind: str, mode: str) -> None: from loopx.cli import main from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime from tests.control_plane.reward_memory_host_fixture import enable_live_memory @@ -132,13 +334,24 @@ def test_real_turn_cli_memory_projection_and_settlement(tmp_path: Path, monkeypa project, runtime, registry = _write_live_fixture(tmp_path, provider=provider) ingest = mode in {"ingest_only", "both_on"} if mode != "unconfigured": - config = enable_live_memory(registry, recall=mode in {"recall_only", "both_on"}, ingest=ingest) + config = enable_live_memory(registry, recall=mode in {"recall_only", "recall_context", "both_on"}, ingest=ingest) if mode == "disabled": data = json.loads(registry.read_text()) data["goals"][0]["control_plane"]["reward_memory"]["enabled"] = False registry.write_text(json.dumps(data)) elif mode == "stale": config.write_text(config.read_text() + "\n") + if mode == "recall_context": + from loopx.capabilities.agent_turn_recall.runtime import run_configured_agent_turn_recall_fail_open + from loopx.capabilities.reward_memory.experiment import resolve_reward_memory_experiment + from tests.capabilities.test_agent_turn_recall import RecallProvider, active_record + _, binding = resolve_reward_memory_experiment( + registry_path=registry, goal_id="loopx-turn-fixture", agent_id="codex-fixture") + provider_fixture = RecallProvider(active_record(binding, expires_at="2099-01-01T00:00:00+00:00")) + # Only provider IO is hermetic; native config, binding, recall receipt, + # admission, host, validator and canonical settlement still execute. + monkeypatch.setattr("loopx.capabilities.agent_turn_recall.run_configured_agent_turn_recall_fail_open", + lambda **kwargs: run_configured_agent_turn_recall_fail_open(**kwargs, provider=provider_fixture)) capture = tmp_path / "host-capture.json" host = tmp_path / "fixture-codex" host.write_text(f"#!{sys.executable}\n" + '''import json, os, pathlib, sys @@ -160,24 +373,102 @@ def test_real_turn_cli_memory_projection_and_settlement(tmp_path: Path, monkeypa print(json.dumps({'type': 'thread.started', 'thread_id': 'fixture-memory-session'})) ''', encoding="utf-8") host.chmod(0o755) + host_args = ["--host", "codex-cli", "--codex-bin", str(host)] + if host_kind == "dsh": + runner = tmp_path / "fixture_dsh_runner.py" + runner.write_text('''import json, os, pathlib +def run_dsh_turn(*, prompt, **kwargs): + authority = json.loads(prompt.split('Turn authority JSON:\\n', 1)[1].split('\\n\\nWhen finished', 1)[0]) + pathlib.Path(os.environ['FIXTURE_HOST_CAPTURE']).write_text(json.dumps({'prompt': prompt, 'authority': authority})) + return json.dumps({'result_kind': 'validated_progress', 'classification': 'fixture_progress', + 'next_action': 'Run the next bounded fixture', 'path_delta_mode': 'unchanged', + 'vision_unchanged_reason': 'The original fixture remains open.', + 'work_evidence': ['artifact:fixture-host-capture'], + 'validation_evidence': ['check:fixture-independent-validator'], + 'reward_memory_reflection_json': '{"status":"no_evidence"}'}) +''', encoding="utf-8") + host_args = ["--host", "dsh", "--dsh-runner", str(runner), "--dsh-home", str(tmp_path / "dsh-home")] + elif host_kind == "generic-cli": + runner = tmp_path / "generic_host.py" + runner.write_text('''import json, os, pathlib, sys +request = json.load(sys.stdin) +pathlib.Path(os.environ['FIXTURE_HOST_CAPTURE']).write_text(json.dumps({'request': request})) +json.dump({'schema_version': 'loopx_turn_result_v0', 'turn_key': request['turn_key'], + 'result_kind': 'validated_progress', 'completed_phases': ['host_execute', 'typed_result'], + 'classification': 'fixture_progress', 'next_action': 'Run the next bounded fixture', + 'recommended_action': 'Continue the original fixture', 'summary': 'Independent fixture validation passed.', + 'delivery_batch_scale': 'single_surface', 'delivery_outcome': 'outcome_progress', + 'vision_unchanged_reason': 'The original fixture remains open.', + 'reward_memory_reflection_json': '{"status":"no_evidence"}'}, sys.stdout) +''', encoding="utf-8") + host_args = ["--host", host_kind, "--host-adapter-command-json", json.dumps([sys.executable, str(runner)])] monkeypatch.setenv("FIXTURE_HOST_CAPTURE", str(capture)) output = io.StringIO() with contextlib.redirect_stdout(output): code = main(["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", "turn", "run-once", "--goal-id", "loopx-turn-fixture", "--agent-id", "codex-fixture", - "--host", "codex-cli", "--project", str(project), "--codex-bin", str(host), + *host_args, "--project", str(project), "--iteration-context", "fresh", "--validation-command-json", json.dumps([sys.executable, "-c", "pass"]), "--scan-root", str(project), "--no-global-sync", "--execute"]) result = json.loads(output.getvalue()) assert code == 0, result observed = json.loads(capture.read_text()) - assert ("reward_memory_reflection_json" in observed["schema"]["properties"]) is ingest - assert ("Set reward_memory_reflection_json" in observed["prompt"]) is ingest - assert "When reward_memory_recall contains guidance" not in observed["prompt"] + if host_kind == "codex-cli": + assert ("reward_memory_reflection_json" in observed["schema"]["properties"]) is ingest + if host_kind != "generic-cli": + assert ("Set reward_memory_reflection_json" in observed["prompt"]) is ingest + assert ("When reward_memory_recall contains guidance" in observed["prompt"]) is (mode == "recall_context") + if mode == "recall_context": + memory = observed["authority" if host_kind == "dsh" else "request"]["reward_memory_recall"] + assert memory["status"] == "applied" + assert "integration branch" in memory["context"]["guidance"][0]["content_summary"] if mode in {"unconfigured", "disabled", "stale", "both_off"}: - assert "reward_memory" not in observed["prompt"] + if host_kind == "generic-cli": + assert "reward_memory_recall" not in observed["request"] + else: + assert "reward_memory" not in observed["prompt"] assert result["effects"]["quota_spent"] is True - assert result["post_settlement"]["external_writes_performed"] is False + assert result.get("post_settlement", {}).get("external_writes_performed", False) is False + if ingest: + assert "post_settlement" in result journals = list((runtime / "goals/loopx-turn-fixture/turns").glob("*.json")) assert len(journals) == 1 assert ("reward_memory_reflection_json" in json.loads(journals[0].read_text())["host_result"]) is ingest + + # Read the real consumers after the native host and settlement. A connected + # lifecycle is source wiring, not an activation or live-provider receipt. + from loopx.capabilities.reward_memory.configuration import reward_memory_goal_configuration_summary + from loopx.capabilities.reward_memory.experiment import resolve_reward_memory_experiment + from loopx.presentation.renderers.reward_memory_markdown import append_agent_reward_memory_markdown + + experiment, _ = resolve_reward_memory_experiment( + registry_path=registry, goal_id="loopx-turn-fixture", agent_id="codex-fixture") + configuration = reward_memory_goal_configuration_summary(json.loads(registry.read_text())["goals"][0]) + assert configuration["effective_available"] is experiment["available"] + assert experiment["automatic_ingest"] is ingest + status_output = io.StringIO() + with contextlib.redirect_stdout(status_output): + status_code = main(["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", + "status", "--goal-id", "loopx-turn-fixture", "--agent-id", "codex-fixture", "--scan-root", str(project)]) + status = json.loads(status_output.getvalue()) + assert status_code == 0, status + item = next(item for item in status["attention_queue"]["items"] if item.get("goal_id") == "loopx-turn-fixture") + projection = item.get("project_asset", {}).get("agent_reward_memory") or item.get("agent_reward_memory") + if experiment["available"]: + assert projection is not None + consumers = [experiment, configuration] + if projection is not None: + assert projection["automatic_ingest"] is ingest + consumers.append(projection) + lines: list[str] = [] + append_agent_reward_memory_markdown(lines, {"agent_reward_memory": projection}, {}) + markdown = "\n".join(lines) + for host_id in ["codex_cli_turn", "dsh_turn", "generic_cli_turn"]: + assert f"{host_id}:recall=connected,ingest=connected_post_settlement" in markdown + assert "lark:recall=status_projection_only,ingest=uncovered" in markdown + for consumer in consumers: + coverage = {row["host_id"]: row for row in consumer["host_coverage"]} + for host_id in ["codex_cli_turn", "dsh_turn", "generic_cli_turn"]: + assert coverage[host_id]["automatic_recall"] == "connected" + assert coverage[host_id]["automatic_ingest"] == "connected_post_settlement" + assert coverage["lark"]["automatic_ingest"] == "uncovered" diff --git a/tests/control_plane/test_turn_capability_lifecycle.py b/tests/control_plane/test_turn_capability_lifecycle.py new file mode 100644 index 0000000000..dfb5e26c71 --- /dev/null +++ b/tests/control_plane/test_turn_capability_lifecycle.py @@ -0,0 +1,107 @@ +"""Capability IO remains optional and uses the original transaction's receipts.""" + +from __future__ import annotations + +import pytest + +from loopx.control_plane.capability_hooks import ( + TurnContextHookRegistration, + dispatch_turn_context_hooks, + project_turn_context, +) +from loopx.control_plane.turn_driver.post_settlement import ( + PostSettlementHookRegistration, + registered_post_settlement, + run_post_settlement_callback, +) + + +def test_invalid_or_duplicate_context_registration_never_reinvokes_provider() -> None: + calls = [] + + def produce(): + calls.append("read") + return { + "readback": {"fixture_readback": {}}, + "context_fields": {}, + "instructions": [], + "result_fields": {}, + } + + valid = TurnContextHookRegistration( + "fixture.context", "fixture", ("fixture_readback",), produce + ) + invalid = TurnContextHookRegistration( + "fixture.invalid", "fixture", ("primary_action",), produce + ) + payload = {"goal_id": "fixture", "agent_id": "fixture"} + dispatch_turn_context_hooks(payload, (invalid, valid, valid)) + assert calls == ["read"] + assert len(payload["turn_context_hook_failures"]) == 2 + + # Retrying composition on an already populated packet must not repeat IO. + dispatch_turn_context_hooks(payload, (valid,)) + assert calls == ["read"] + + +@pytest.mark.parametrize("serialized", [True, 3, "invalid", {"fixture": "invalid"}]) +def test_invalid_recovered_context_is_omitted_without_gating_work(serialized) -> None: + assert project_turn_context({"turn_context_contributions": serialized}) == {} + + +def test_post_settlement_retry_keeps_completed_peer_hook_and_original_evidence() -> ( + None +): + calls = [] + + def completed(_plan, _result, evidence): + assert evidence["writeback"]["appended"] is True + assert evidence["quota_spend"]["appended"] is True + calls.append("completed") + return {"status": "activated", "external_writes_performed": True} + + def uncertain(_plan, _result, _evidence): + calls.append("uncertain") + if calls.count("uncertain") == 1: + raise OSError("lost provider response") + return {"status": "activated", "external_writes_performed": False} + + callback = registered_post_settlement( + ( + PostSettlementHookRegistration( + "fixture.complete", "fixture-one", completed + ), + PostSettlementHookRegistration("fixture.pending", "fixture-two", uncertain), + ) + ) + journal = { + "task_validation": {"ok": True}, + "writeback": {"ok": True, "appended": True}, + "quota_spend": {"ok": True, "appended": True}, + } + saved = [] + for _ in range(3): + run_post_settlement_callback( + plan={}, + result={}, + post_settlement=callback, + journal=journal, + persist_journal=lambda value: saved.append(dict(value)), + ) + assert calls == ["completed", "uncertain", "uncertain"] + assert len(saved) == 2 + assert ( + saved[0]["post_settlement"]["results"]["fixture.pending"][ + "external_writes_performed" + ] + is None + ) + assert journal["post_settlement"]["status"] == "observed" + + +def test_duplicate_settlement_registration_is_rejected_before_io() -> None: + hook = PostSettlementHookRegistration( + "fixture.duplicate", "fixture", lambda *_: pytest.fail("no IO") + ) + with pytest.raises(ValueError, match="duplicate"): + registered_post_settlement((hook, hook)) diff --git a/tests/control_plane_ts/turn_capability_context.test.ts b/tests/control_plane_ts/turn_capability_context.test.ts new file mode 100644 index 0000000000..7750530b94 --- /dev/null +++ b/tests/control_plane_ts/turn_capability_context.test.ts @@ -0,0 +1,50 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import {validateTurnContextRegistration, validateTurnContextContribution, + projectTurnContextContribution} from "../../loopx/control_plane/turn_driver/capability_context.ts"; + +const registration = {hook_id: "fixture.context", capability_id: "fixture-capability", + readback_slots: ["fixture_readback"], max_result_bytes: 4096}; +const source = {goal_id: "fixture-goal", agent_id: "fixture-agent", boundary: {write_scope: []}}; +const result = {readback: {fixture_readback: {value: "private advice"}}, + context_fields: {fixture_context: {value: "private advice"}}, instructions: ["Advisory only."], + result_fields: {fixture_result: {type: "string", maxLength: 240}}}; + +test("registered context survives location-only capture binding", () => { + const contribution = validateTurnContextContribution({registration, source, result}); + assert.deepEqual(projectTurnContextContribution({contribution, readback: result.readback, + source: {...source, detail_ref: {full_decision: "fixture location"}, measurements: {bytes: 500}}}), contribution); +}); + +for (const change of ["goal", "agent", "authority", "readback", "instructions", "result_fields", "context_fields"]) { + test(`changed ${change} invalidates the optional contribution`, () => { + const contribution = validateTurnContextContribution({registration, source, result}); + const input = structuredClone({contribution, source, readback: result.readback}); + if (change === "goal") input.source.goal_id = "another-goal"; + else if (change === "agent") input.source.agent_id = "another-agent"; + else if (change === "authority") input.source.boundary.write_scope.push("forbidden" as never); + else if (change === "readback") input.readback.fixture_readback.value = "changed"; + else input.contribution[change] = change === "instructions" ? ["changed"] : {}; + assert.throws(() => projectTurnContextContribution(input)); + }); +} + +for (const field of ["write_scope", "primary_action", "turn_envelope", "result_contract", "work_evidence", "validation_evidence"]) { + test(`capability cannot replace ${field}`, () => { + assert.throws(() => validateTurnContextContribution({registration, source, + result: {...result, context_fields: {[field]: "replacement"}}})); + assert.throws(() => validateTurnContextContribution({registration, source, + result: {...result, result_fields: {[field]: {type: "string", maxLength: 240}}}})); + }); +} + +test("registration and result budgets fail closed", () => { + assert.throws(() => validateTurnContextRegistration({...registration, max_result_bytes: 65537})); + assert.throws(() => validateTurnContextRegistration({...registration, readback_slots: ["fixture", "fixture"]})); + assert.throws(() => validateTurnContextContribution({registration, source, + result: {...result, instructions: ["x".repeat(5000)]}})); + assert.throws(() => validateTurnContextContribution({registration, source, + result: {...result, result_fields: {fixture_result: {type: "object"}}}})); + assert.throws(() => validateTurnContextContribution({registration, source, + result: {...result, readback: {another_slot: {value: "private"}}}})); +}); diff --git a/tests/test_loopx_turn_driver.py b/tests/test_loopx_turn_driver.py index 8714aafbf3..0fb4aa4a15 100644 --- a/tests/test_loopx_turn_driver.py +++ b/tests/test_loopx_turn_driver.py @@ -927,6 +927,8 @@ def test_turn_host_request_carries_reward_memory_decision_context() -> None: "grants_new_action_authority": False, }) + from tests.control_plane.reward_memory_host_fixture import register_plan_memory + register_plan_memory(plan) request = build_loopx_turn_host_request(plan) assert request["reward_memory_recall"] == plan["reward_memory_recall"] @@ -3286,6 +3288,7 @@ def post_settlement( "task_validation": execution["validation"], "writeback": {"ok": True, "appended": True}, "quota_spend": {"ok": True, "appended": True, "slots": 1}, + "previous_post_settlement": {}, } ] assert loopx_turn_execution_committed(execution) is True @@ -3583,13 +3586,21 @@ def fake_codex_host(request: dict[str, object], **_kwargs: object) -> dict[str, assert "One%20public%20fixture%20advanced" not in state -def test_turn_run_once_codex_cli_wires_validated_reflection_post_settlement( +@pytest.mark.parametrize("memory_path", ["callback", "owned", "owned_lost_reply", "ordinary_validator"]) +@pytest.mark.parametrize("host_kind", ["codex-cli", "dsh", "generic-cli"]) +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_turn_run_once_host_wires_validated_reflection_post_settlement( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, + host_kind: str, + provider: str, + memory_path: str, ) -> None: - project, runtime, registry = _write_live_fixture(tmp_path) + from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_live_fixture(tmp_path, provider=provider) from tests.control_plane.reward_memory_host_fixture import enable_live_memory - enable_live_memory(registry) + enable_live_memory(registry, validated_ingest=memory_path != "callback") reflection = json.dumps( { "schema_version": "turn_reward_memory_reflection_v1", @@ -3654,13 +3665,25 @@ def fake_codex_host( "delivery_outcome": "outcome_progress", "vision_unchanged_reason": "The fixture objective remains open.", "summary": "One Codex CLI fixture step passed validation.", + "work_evidence": ["artifact:fixture-host-result"], + "validation_evidence": ["check:fixture-independent-validator"], "reward_memory_reflection_json": reflection, } + from loopx.capabilities import reward_memory + from tests.capabilities.test_reward_memory_ingestion import FakeProvider + actual_ingest = reward_memory.run_configured_turn_outcome_ingest_fail_open + memory_provider = FakeProvider() observed: list[dict[str, Any]] = [] def fake_ingest(**kwargs: object) -> dict[str, object]: observed.append(dict(kwargs)) + if memory_path != "callback": + receipt = actual_ingest(**kwargs, provider=memory_provider) + if memory_path == "owned_lost_reply" and len(observed) == 1: + # The memory owner committed, but its caller lost the response. + raise OSError("provider owner response lost before Turn receipt") + return receipt return { "ok": True, "schema_version": "turn_reward_memory_ingest_v0", @@ -3676,43 +3699,79 @@ def fake_ingest(**kwargs: object) -> dict[str, object]: ) monkeypatch.setattr( ( - "loopx.cli_commands.turn_run_once." + "loopx.capabilities.reward_memory." "run_configured_turn_outcome_ingest_fail_open" ), fake_ingest, ) + host_args = ["--host", host_kind] + if host_kind == "dsh": + runner = tmp_path / "reflection_runner.py" + candidate = fake_codex_host({"turn_key": "unused"}) + runner.write_text( + "import json\n" + "def run_dsh_turn(**kwargs):\n" + f" return {json.dumps(candidate)!r}\n", + encoding="utf-8", + ) + host_args += ["--dsh-runner", str(runner), "--dsh-home", str(tmp_path / "dsh-home")] + elif host_kind == "generic-cli": + runner = tmp_path / "generic_reflection_runner.py" + candidate = fake_codex_host({"turn_key": "unused"}) + runner.write_text( + "import json, sys\n" + "request = json.load(sys.stdin)\n" + f"result = json.loads({json.dumps(candidate)!r})\n" + "result['turn_key'] = request['turn_key']\n" + "json.dump(result, sys.stdout)\n", + encoding="utf-8", + ) + host_args += ["--host-adapter-command-json", json.dumps([sys.executable, str(runner)])] + args = [ + "--registry", + str(registry), + "--runtime-root", + str(runtime), + "--format", + "json", + "turn", + "run-once", + "--goal-id", + "loopx-turn-fixture", + "--agent-id", + "codex-fixture", + *host_args, + "--project", + str(project), + "--validation-command-json", + json.dumps([sys.executable, "-c", "pass"] if memory_path == "ordinary_validator" else [sys.executable, str(validator)]), + "--scan-root", + str(project), + "--no-global-sync", + "--execute", + ] output = io.StringIO() with contextlib.redirect_stdout(output): - exit_code = cli_main( - [ - "--registry", - str(registry), - "--runtime-root", - str(runtime), - "--format", - "json", - "turn", - "run-once", - "--goal-id", - "loopx-turn-fixture", - "--agent-id", - "codex-fixture", - "--host", - "codex-cli", - "--project", - str(project), - "--validation-command-json", - json.dumps([sys.executable, str(validator)]), - "--scan-root", - str(project), - "--no-global-sync", - "--execute", - ] - ) + exit_code = cli_main(args) payload = json.loads(output.getvalue()) assert exit_code == 0, payload - assert payload["post_settlement"] == { + receipt = payload["post_settlement"]["results"]["reward-memory.outcome-ingest"] + if memory_path == "owned_lost_reply": + assert receipt["status"] == "runtime_unavailable", receipt + assert receipt["external_writes_performed"] is None + assert memory_provider.sync_calls == memory_provider.retrieve_calls == 1 + elif memory_path == "owned": + assert receipt["status"] == "activated", receipt + assert receipt["exact_readback_verified"] is True + assert memory_provider.sync_calls == 1 + assert memory_provider.retrieve_calls == 1 + elif memory_path == "ordinary_validator": + assert receipt["status"] == "evidence_validation_required", receipt + assert receipt["external_writes_performed"] is False + assert memory_provider.sync_calls == memory_provider.retrieve_calls == 0 + else: + assert receipt == { "ok": True, "schema_version": "turn_reward_memory_ingest_v0", "status": "activated", @@ -3723,12 +3782,40 @@ def fake_ingest(**kwargs: object) -> dict[str, object]: assert len(observed) == 1 assert observed[0]["host_result"]["reward_memory_reflection_json"] == reflection evidence = observed[0]["settlement_evidence"] - assert evidence["task_validation"]["reward_memory_reflection_validation"][ - "status" - ] == "validated" + if memory_path != "ordinary_validator": + assert evidence["task_validation"]["reward_memory_reflection_validation"]["status"] == "validated" + else: + assert "reward_memory_reflection_validation" not in evidence["task_validation"] assert evidence["writeback"]["appended"] is True assert evidence["quota_spend"]["appended"] is True + # The terminal Turn remains immutable. Each resume can read the capability + # owner's durable receipt, without repeating provider IO or settlement. + journal_path = runtime / "goals" / "loopx-turn-fixture" / "turns" / ( + payload["resume_turn_key"].removeprefix("sha256:") + ".json" + ) + terminal_journal = journal_path.read_bytes() + for _ in range(2): + replay_output = io.StringIO() + with contextlib.redirect_stdout(replay_output): + replay_code = cli_main([*args, "--resume-turn-key", payload["resume_turn_key"]]) + replay = json.loads(replay_output.getvalue()) + assert replay_code == 0, replay + assert replay["replayed"] is True + assert not any(replay["effects"].values()) + assert journal_path.read_bytes() == terminal_journal + assert replay["post_settlement"] == payload["post_settlement"] + if memory_path == "owned_lost_reply": + recovered = replay["post_settlement_recovery"]["results"]["reward-memory.outcome-ingest"] + assert recovered["status"] == "activated", recovered + assert recovered["deduplicated"] is True + assert recovered["sidecar_receipt_reused"] is True + assert recovered["external_writes_performed"] is False + else: + assert "post_settlement_recovery" not in replay + assert memory_provider.sync_calls == memory_provider.retrieve_calls == (0 if memory_path in {"callback", "ordinary_validator"} else 1) + assert len(observed) == (3 if memory_path == "owned_lost_reply" else 1) + def test_turn_run_once_cli_resumes_session_from_recoverable_failed_turn( tmp_path: Path,