diff --git a/loopx/capabilities/agent_turn_recall/core.py b/loopx/capabilities/agent_turn_recall/core.py index 4f7954981d..bce98d2a15 100644 --- a/loopx/capabilities/agent_turn_recall/core.py +++ b/loopx/capabilities/agent_turn_recall/core.py @@ -45,7 +45,9 @@ def _recent_outcomes(quota_decision: Mapping[str, Any]) -> list[str]: ] ) handoff = _mapping(quota_decision.get("handoff_readiness")) - candidates.append(_mapping(handoff.get("post_handoff_latest_run")).get("classification")) + candidates.append( + _mapping(handoff.get("post_handoff_latest_run")).get("classification") + ) result: list[str] = [] for value in candidates: text = _compact(value, limit=180) @@ -121,9 +123,7 @@ def build_agent_turn_situation( "decision": _compact(quota_decision.get("decision"), limit=80), "selected_todo": selected, "recent_outcomes": _recent_outcomes(quota_decision), - "next_intent": _compact( - quota_decision.get("recommended_action"), limit=320 - ), + "next_intent": _compact(quota_decision.get("recommended_action"), limit=320), "conflict_state": conflict_state, "status_health_ok": status_health_ok, "user_prompt_included": False, @@ -231,6 +231,9 @@ def apply_guidance( "candidate_ref": item.candidate_ref, "target_class": item.target_class, "content_summary": item.content_summary, + "experience": ( + dict(item.experience) if item.experience is not None else None + ), } for item in items ] diff --git a/loopx/capabilities/content_ops/README.md b/loopx/capabilities/content_ops/README.md index a68a2c4af6..7b50211e67 100644 --- a/loopx/capabilities/content_ops/README.md +++ b/loopx/capabilities/content_ops/README.md @@ -145,7 +145,10 @@ user's provider, and another Agent's private corpus is not a shared seed store. credentials and scope references in its owner-local configuration. 3. Prepare one `scoped_feedback_reward_memory_event_v0` using the seed's `content_summary`, `target_class=procedural_experience`, and source kind - `reviewed_learning_card`. Use the seed id/version/digest as a stable source + `reviewed_learning_card`. Map the seed's applicability, observations, + procedure and limits into the required `procedural_experience_contract_v0`; + bind its evidence refs to the reviewed seed revision. Use the seed + id/version/digest as a stable source reference; fill workspace/project/user/peer and surface from the configured corpus. Set `requested_action_scopes=[]` and `raw_content_captured=false`. The configured standing policy still reviews the event. Set diff --git a/loopx/capabilities/content_ops/experiences/x-composer-preflight-v1.json b/loopx/capabilities/content_ops/experiences/x-composer-preflight-v1.json index fb7614d7ab..da80477ce1 100644 --- a/loopx/capabilities/content_ops/experiences/x-composer-preflight-v1.json +++ b/loopx/capabilities/content_ops/experiences/x-composer-preflight-v1.json @@ -41,7 +41,7 @@ "query_hint": "X browser draft preparation: multi-link card selection, rich-text replacement and final readback", "instructions": [ "Use the actor's explicitly configured corpus and provider; do not discover or reuse another Agent's private memory.", - "Review this seed against current artifacts and higher-priority instructions. Fill the existing scoped_feedback event with the configured scope, this content_summary and a seed/version/digest source reference.", + "Review this seed against current artifacts and higher-priority instructions. Fill the existing scoped_feedback event with the configured scope, this content_summary and a seed/version/digest source reference. Map applicability, an observed outcome and attribution from observations, future_behavior from the procedure, limits, and the reviewed seed ref into procedural_experience_contract_v0.", "Do not mark current_artifact_verified merely because the seed exists. Import as procedural_experience only after current verification through the existing candidate/review and ingest-event path.", "Require the existing provider write and exact readback receipt before claiming initialization. Deduplicate the same scoped seed revision; review updates rather than overwriting newer local experience.", "For automatic recall, configure the existing agent_workflow.turn_admission surface and include the actual X preparation work in the selected Todo. A configured provider alone does not connect a recall surface.", diff --git a/loopx/capabilities/decision_context/outcome_feedback.py b/loopx/capabilities/decision_context/outcome_feedback.py index 3181cc4f61..b96e9d1090 100644 --- a/loopx/capabilities/decision_context/outcome_feedback.py +++ b/loopx/capabilities/decision_context/outcome_feedback.py @@ -260,6 +260,41 @@ def build_decision_outcome_feedback( f"Decision: {accepted['summary']} " f"Verified outcome: {verified_outcome['summary']}" ), + "experience": { + "schema_version": "procedural_experience_contract_v0", + "applicability": [ + f"A future decision matches: {accepted['summary']}" + ], + "observed_outcome": verified_outcome["summary"], + "attribution": ( + "The accepted decision and verified outcome are linked to " + "the same exact-read evidence packet; rejected or stale " + "claims were excluded from the attribution." + ), + "future_behavior": { + "trigger": ( + "The same decision pattern recurs with a current source " + "revision and no unresolved authority conflict." + ), + "action": ( + f"Re-evaluate the evidence before applying this decision: " + f"{accepted['summary']}" + ), + "validation": ( + "Require an exact-read promoted claim and a verified outcome " + "linked to the same evidence packet." + ), + "stop_condition": ( + "Do not reuse the experience when the source revision is " + "stale, the outcome is unverified, or a conflict is unresolved." + ), + }, + "limitations": [ + "One verified outcome does not establish universal causality.", + "Reuse requires current evidence and the original scope boundaries.", + ], + "evidence_refs": [evidence["packet_ref"], outcome["packet_ref"]], + }, "source": { "source_kind": "verified_decision_outcome", "source_ref": outcome["packet_ref"], diff --git a/loopx/capabilities/reward_memory/README.md b/loopx/capabilities/reward_memory/README.md index 02636441e1..0ea33c280b 100644 --- a/loopx/capabilities/reward_memory/README.md +++ b/loopx/capabilities/reward_memory/README.md @@ -179,10 +179,14 @@ path rather than a compatibility fallback. The production Codex CLI Turn performs recall after quota/Todo admission and accepts outcome ingestion only after independent validation, durable writeback, -and quota settlement. A reflection must use `turn_reward_memory_reflection_v0` +and quota settlement. A reflection must use `turn_reward_memory_reflection_v1` and include an exact configured surface, a distinct research/simulation/real/ engineering source kind, and opaque evidence refs; an ordinary Turn summary is -not evidence. Ambiguous provider commits and unverified readbacks remain in a +not evidence. It also carries `procedural_experience_contract_v0` with +applicability, observed outcome, attribution, complete future behavior, +limitations, and the same evidence refs. Legacy v0 reflections remain +audit-only and cannot become durable memory. Ambiguous provider commits and +unverified readbacks remain in a mode-0600 Goal+Agent+event sidecar. The next executing Turn retries the same deterministic event before recall, so the provider can deduplicate it and LoopX can require exact readback. Explicit disable suppresses reconciliation and all @@ -191,7 +195,7 @@ provider calls. The Codex App uses the same settlement boundary without copying the raw reflection into run indexes, rollout events, or public projections. For a Todo-bound accountable refresh, the caller may add -`--reward-memory-reflection-json `. +`--reward-memory-reflection-json `. LoopX stores that candidate only in a mode-0600 Goal+Agent+candidate sidecar and runs the exact completion-validation command already declared by that Todo. The validator must return `reward_memory_reflection_validation_v0` with the exact @@ -261,7 +265,12 @@ The agent reviews the conversation before choosing what, if anything, to learn: `schema_version=scoped_feedback_reward_memory_event_v0`, a stable `feedback_ref`, actual `source`, `reasoning`, `guard_context`, compact `content_summary`, `target_class`, and exact identity/surface/revision/action - scope. Advisory classes require empty `requested_action_scopes`; allowed + scope. `procedural_experience` additionally requires a typed + `procedural_experience_contract_v0`: applicability, observed outcome, + attribution, future behavior (`trigger`, `action`, `validation`, and + `stop_condition`), limitations, and opaque evidence refs. A fact recap does + not satisfy this contract. Advisory classes require empty + `requested_action_scopes`; allowed policy scopes do not grant advisory memory action authority. Do not copy the fixture's actor or verified-guard assertions. 4. Replace the hint's input placeholder and preview `ingest-event` without @@ -286,7 +295,7 @@ the agent (or its applicable route); no extra store, queue or scheduler exists. | `run_bound_reward` | Explicit human judgment attached to one exact goal/run. | Evidence about that outcome only. Future influence requires compact candidate derivation and an activation policy; the overlay itself is not a standing instruction. | Append-only overlay; corrections and revocations append references instead of rewriting the judged run. | | `hard_policy` | Explicit user/repository/operator authority, or policy content inferred from verified owner/core-contributor evidence and bound to an existing project/action authority scope. | Constraint or veto inside the verified scope. Reasoning may infer policy meaning from rewards, preferences, current-artifact-verified experience, selected options, accepted/rejected outcomes, and maintainer corrections; it may not infer credentials, new publish/production scope, or cross-user/repository authority. | Active records retain actor, evidence, scope, and derivation provenance until superseded, revoked, or expired; temporary or weakly reinforced inference should expire or return to review. | | `soft_preference` | Explicit feedback, selected options, or later reviewed candidates scoped to a workspace/project and module-owned surface. | Advisory ranking or rewrite only. It cannot grant publish, merge, write, credential, or production authority. | Durable only after explicit review; editable, rejectable, supersedable, revocable, and retireable. | -| `procedural_experience` | Revision-stamped trajectories, distilled experiences, maintainer corrections, accepted/rejected changes, and reviewed architectural learning, with repository/module/revision/applicability scope. | Advisory diagnosis, scope, routing, or validation guidance only after current-artifact verification. A training/evaluation case is evidence, not an executable instruction. Retrieval alone has zero patch authority. | Trajectories may be add-only; distilled or architectural experiences are supersedable. New source truth can stale, quarantine, refute, or retire them. | +| `procedural_experience` | Revision-stamped trajectories, distilled experiences, maintainer corrections, accepted/rejected changes, and reviewed architectural learning. The typed experience contract preserves applicability, observed outcome, attribution, future behavior, limitations, and evidence refs. | Advisory diagnosis, scope, routing, or validation guidance only after current-artifact verification and experience-quality qualification. A fact-only summary or successful provider write is not experience evidence. Retrieval alone has zero patch authority. | New source truth or application evidence can mark an experience stale, harmful, refuted, superseded, or retired. Initial qualification means structurally reusable and evidence-bound; value remains unproven until application/outcome attribution. | | `working_context` | Either fresh execution state (`fresh_execution_context`) or a revisioned session-continuation summary (`session_working_memory`). | Supports only the current execution/session continuation. Neither subtype becomes reusable policy or grants action authority. Fresh source-of-truth reads outrank recalled material. | `fresh_execution_context` already exists in LoopX registry/state/todo/quota/checkout observations and is reused, not rebuilt. Session context remains bound to its session/archive revision. | Every durable record must name `source`, `scope`, `authority`, `confidence`, @@ -459,13 +468,32 @@ repository write authority, publish/production scope, or cross-project authority. Out-of-scope, conflicted, stale, raw, or unmodelled input is `guard_blocked` before any provider call. +For `procedural_experience`, deterministic guards first require the complete +typed experience contract. Missing or malformed future behavior, limitations, +or evidence binding is rejected before the provider is called. The complete +contract participates in `candidate_ref`, survives the active envelope, and is +injected into recall guidance. Legacy fact-only procedural records are not +recallable. This is a qualification gate, not a claim of proven utility: +application receipts and verified outcome attribution still decide whether an +experience was helpful, neutral, harmful, or should be retired. + The command then composes deterministic `candidate_ref` deduplication, standing policy acceptance, active-envelope construction, declared-provider `sync`, and -one exact-corpus/surface function-boundary recall. A +one exact-corpus/surface function-boundary readback. For +`procedural_experience`, it also performs a destination-surface +`business_recall` using the applicability and future-behavior fields, without +the candidate id in the query. The recalled record must match both the +candidate and experience digests. This prevents an exact-id storage check from +being mistaken for evidence that a lesson can be found where decisions use it. A `reward_memory_ingest_receipt_v0` reports `activated` and `memory_available_for_recall=true` only when resource ref, candidate ref, and -canonical content digest all match. Provider unavailability, pending commit, -or readback mismatch fails open and does not block the caller's normal work. +canonical content digest all match, the class-specific experience-quality gate +passes, and any required destination recall succeeds. A destination miss stays +`recall_unverified` for bounded reconciliation rather than becoming active. +`exact_readback_verified=true` alone proves persistence, not experience quality, +discoverability, or utility. Provider unavailability, pending commit, readback +mismatch, or destination-recall miss fails open and does not block the caller's +normal work. `observed_at` is the immutable first-observed event timestamp and must be reused on retries. The provider target binds both standing-policy and candidate digests so a policy revision cannot silently reuse an older activation. diff --git a/loopx/capabilities/reward_memory/README.zh-CN.md b/loopx/capabilities/reward_memory/README.zh-CN.md index 6ead50523d..c1289bbbf1 100644 --- a/loopx/capabilities/reward_memory/README.zh-CN.md +++ b/loopx/capabilities/reward_memory/README.zh-CN.md @@ -147,15 +147,18 @@ receipt。它不采集聊天、不解析 tool log、不保存 raw content，也 生产 Codex CLI Turn 在 quota/Todo admission 后执行 recall；只有独立验证、持久 writeback 与 quota settlement 均完成后，才接受 outcome ingest。Reflection 必须使用 -`turn_reward_memory_reflection_v0`，携带精确配置的 surface、相互区分的 -research/simulation/real/engineering 来源类型和 opaque evidence refs；普通 Turn summary 不算 -证据。Provider commit 含糊或精确读回失败时，会保留一份权限为 0600、按 Goal+Agent+事件隔离的 +`turn_reward_memory_reflection_v1`，携带精确配置的 surface、相互区分的 +research/simulation/real/engineering 来源类型和 opaque evidence refs；还必须携带 +`procedural_experience_contract_v0`，分别写出适用情境、已观察结果、归因、未来行为（触发、动作、 +验证、停止条件）、失效边界，并绑定相同 evidence refs。只有事实复述、没有未来行为改变的普通 +Turn summary 不算经验。旧的 v0 reflection 只保留审计兼容，不会写成可持久召回的记忆。 +Provider commit 含糊或精确读回失败时，会保留一份权限为 0600、按 Goal+Agent+事件隔离的 sidecar。下一个执行 Turn 会在 recall 前以同一 deterministic event 重试，使 Provider 可以去重， LoopX 再要求精确读回。显式关闭会停止 reconciliation，并保持零 Provider 调用。 Codex App 复用同一结算边界，但不会把原始 reflection 放入 run index、rollout event 或公共 projection。Todo-bound 的 accountable refresh 可以追加 -`--reward-memory-reflection-json `；LoopX 只把候选存入 +`--reward-memory-reflection-json `；LoopX 只把候选存入 权限为 0600、按 Goal+Agent+candidate 隔离的 sidecar，并运行该 Todo 已声明的精确 completion validation 命令。Validator 必须返回 `reward_memory_reflection_validation_v0`，且其中的 reflection digest 与 evidence refs 必须完全匹配；普通验收命令仅仅退出 0 并不足够。随后只有同一 identity @@ -194,7 +197,7 @@ recall 调用点：`reviewer_artifact.summary` 应用简短 reviewer-facing 摘 | `run_bound_reward` | 人对某个精确 goal/run 给出的显式评价。 | 只描述该次结果。若要影响后续行为，必须先形成紧凑候选并经过 activation policy；reward overlay 本身不是长期指令。 | Overlay 只追加；修正和撤销通过引用追加，不回写被评价的原始 run。 | | `hard_policy` | 显式的用户、仓库或 operator authority，或者从已验证的 owner/核心贡献者证据中推导、并绑定到既有 project/action authority scope 的策略内容。 | 在已验证作用域内形成约束或否决。模型可以从 reward、preference、经过当前 artifact 验证的 experience、选项选择、接受/拒绝结果和 maintainer correction 中推导策略含义；不能推导 credential、新的 publish/production scope 或跨用户、跨仓库 authority。 | Active record 保留 actor、evidence、scope 和 derivation provenance，直到被 supersede、revoke 或 expire。临时或证据较弱的推导应当过期或回到 review。 | | `soft_preference` | 显式反馈、用户选择，或者后续经过 review 的候选；作用域绑定到 workspace/project 和模块自有 surface。 | 只用于 advisory ranking 或 rewrite，不能授予 publish、merge、write、credential 或 production authority。 | 只有经过显式 review 才能持久化；支持 edit、reject、supersede、revoke 和 retire。 | -| `procedural_experience` | 带 revision 的 trajectory、distilled experience、maintainer correction、接受/拒绝变更和经过 review 的架构经验；同时带 repository/module/revision/applicability scope。 | 只有经过当前 artifact 验证，才能作为诊断、范围判断、路由或验证建议。训练/评测 case 是证据，不是可执行指令；单次 retrieval 对 patch 没有 authority。 | Trajectory 可以只追加；distilled 或 architectural experience 支持 supersede。新的 source truth 可以把旧经验标记为 stale、quarantine、refute 或 retire。 | +| `procedural_experience` | 带 revision 的 trajectory、distilled experience、maintainer correction、接受/拒绝变更和经过 review 的架构经验。结构化经验契约保留适用情境、已观察结果、归因、未来行为、失效边界和证据引用。 | 只有经过当前 artifact 验证和经验质量门禁，才能作为诊断、范围判断、路由或验证建议。只有事实摘要或 provider 写入成功都不构成经验；单次 retrieval 也不授予动作 authority。 | 新 source truth 或后续应用证据可以把经验标记为 stale、harmful、refuted、superseded 或 retired。初始门禁只证明结构可复用且证据绑定；真实价值仍需应用与结果归因。 | | `working_context` | 包含 fresh execution state（`fresh_execution_context`）和带 revision 的 session continuation 摘要（`session_working_memory`）。 | 只服务当前执行或 session 延续，不能自动升级成可复用 policy，也不能授予动作 authority。当前 source-of-truth 读取始终高于 recall 内容。 | `fresh_execution_context` 已存在于 LoopX 的 registry/state/todo/quota/checkout observation 中，本设计直接复用。Session context 继续绑定对应 session/archive revision。 | 每条持久记录除了类别，还必须包含 `source`、`scope`、`authority`、`confidence`、 @@ -329,7 +332,9 @@ surface。两者都只把严格、紧凑的字段映射到同一个 `reward_memo lifecycle、store、scheduler、recall path 或 semantic router。LoopX 不读取或保存原始 feedback body，也不会根据关键词自动判断“哪条反馈值得记忆”。模型或调用模块先把材料压缩成只包含 source ref、已验证 actor/role、精确 workspace/project/surface/revision、内容摘要、reasoning -和当前 artifact 证据的 event。 +和当前 artifact 证据的 event。若目标类别是 `procedural_experience`，event 还必须提供完整的 +`procedural_experience_contract_v0`；缺少适用情境、已观察结果、归因、未来行为任一子字段、 +失效边界或 evidence refs 时，会在 provider 调用前拒绝。 一个 `reward_memory_standing_policy_v0` 预先声明 corpus owner、reviewer、authority source、 精确 project/surface、唯一 memory class、允许的 source kind、已验证 actor role 和 action @@ -337,12 +342,21 @@ scope。它把“每条 comment 重复审批”收敛成“一次批准精确边 仓库写权限、publish/production scope 或跨项目 authority。任何越界、冲突、非 current source 或 raw/unmodelled 字段都会在 provider 调用前 `guard_blocked`。 +完整经验契约参与 `candidate_ref`，写入 active envelope，并在召回时一并注入。旧的“只有摘要” +程序性记录不会被召回。该门禁证明的是结构与证据资格，不会提前宣称经验真的 helpful；后续仍由 +application receipt 和已验证 outcome attribution 判断 helpful、neutral、harmful 或 retire。 + 同一命令按顺序执行：确定性 `candidate_ref` 去重、standing-policy accept、active envelope、 -声明 provider 的 `sync`、同一 exact corpus/surface 的 function-boundary recall，以及 -resource ref、candidate ref、canonical content digest 三重读回校验。只有三项都一致时， -`reward_memory_ingest_receipt_v0` 才返回 `activated` 和 -`memory_available_for_recall=true`。Provider 不可用、commit pending 或读回不一致都 fail open， -不阻塞调用方的正常工作。`observed_at` 是事件第一次被观察到的不可变时间，retry 必须复用它； +声明 provider 的 `sync`、同一 exact corpus/surface 的 function-boundary 精确读回，以及 +resource ref、candidate ref、canonical content digest 三重校验。对于 +`procedural_experience`，还会在真实目标 surface 用适用情境与未来行为执行一次不带 candidate id +的 `business_recall`，召回记录必须同时匹配 candidate digest 与 experience digest。这样可避免把 +“按精确 id 能读回来”误当成“做决策时能找到这条经验”。只有精确读回、经验质量门禁和必要的 +业务召回全部通过，`reward_memory_ingest_receipt_v0` 才返回 `activated` 和 +`memory_available_for_recall=true`。业务召回未命中时保持 `recall_unverified`，交给有界 reconciliation， +不会提前激活。`exact_readback_verified=true` 只证明存储完整，不证明经验质量、可发现性或真实效用。 +Provider 不可用、commit pending、读回不一致或业务召回未命中都 fail open，不阻塞调用方的正常工作。 +`observed_at` 是事件第一次被观察到的不可变时间，retry 必须复用它； provider target 同时绑定 standing-policy 与 candidate digest，避免策略换版误复用旧激活。 `--execute` 缺省关闭，dry-run 只返回 `planned`。Execute 还必须同时提供 goal id 和 allowlist 内 agent id，调用方不能通过直接传 provider binding 绕过默认关闭的实验策略。 diff --git a/loopx/capabilities/reward_memory/application.py b/loopx/capabilities/reward_memory/application.py index 0733e23852..e863fb2c47 100644 --- a/loopx/capabilities/reward_memory/application.py +++ b/loopx/capabilities/reward_memory/application.py @@ -18,6 +18,11 @@ ) from ..context_providers.openviking import classify_openviking_scope from .candidate_review import REWARD_MEMORY_REVIEW_SCHEMA_VERSION +from .experience_quality import ( + normalize_procedural_experience, + procedural_experience_digest, + procedural_experience_quality, +) from .registry import IDENTITY_SCOPE_FIELDS, normalize_reward_memory_corpus @@ -59,6 +64,8 @@ class RewardMemoryRecallItem: candidate_ref: str target_class: str content_summary: str + experience: Mapping[str, Any] | None = None + experience_digest: str = "" content_digest: str = "" @@ -182,7 +189,12 @@ def build_active_reward_memory_record( expires_at = record["lifecycle"].get("expires_at") if expires_at: active_lifecycle["expires_at"] = expires_at - return { + experience = record.get("experience") + quality = procedural_experience_quality( + target_class=str(record.get("target_class") or ""), + experience=experience, + ) + active = { "schema_version": REWARD_MEMORY_ACTIVE_RECORD_SCHEMA_VERSION, "activation_ref": activation_ref, "activated_at": activated, @@ -192,6 +204,7 @@ def build_active_reward_memory_record( "content_summary": _compact( record.get("content_summary"), "content_summary", limit=500 ), + "experience_quality": quality, "scope": dict(record["scope"]), "source": dict(record["source"]), "review": dict(reviewed_candidate["review"]), @@ -205,6 +218,9 @@ def build_active_reward_memory_record( "provider_write_performed": False, "external_writes_performed": False, } + if experience is not None: + active["experience"] = normalize_procedural_experience(experience) + return active def _authority_checkpoint( @@ -534,6 +550,16 @@ def _active_item( return None if expires.tzinfo is None or observed.tzinfo is None or observed >= expires: return None + experience = envelope.get("experience") + quality = procedural_experience_quality( + target_class=str(envelope.get("target_class") or ""), + experience=experience, + ) + if quality["passed"] is not True: + return None + normalized_experience = ( + normalize_procedural_experience(experience) if experience is not None else None + ) return RewardMemoryRecallItem( memory_ref=item.resource_ref, candidate_ref=_token(envelope.get("candidate_ref"), "candidate_ref"), @@ -541,6 +567,12 @@ def _active_item( content_summary=_compact( envelope.get("content_summary"), "content_summary", limit=500 ), + experience=normalized_experience, + experience_digest=( + procedural_experience_digest(normalized_experience) + if normalized_experience is not None + else "" + ), content_digest=hashlib.sha256( canonical_context_text(item.content).encode("utf-8") ).hexdigest(), @@ -674,6 +706,12 @@ def execute_reward_memory_recall( "candidate_ref": item.candidate_ref, "target_class": item.target_class, "content_summary": item.content_summary if expose_summary else None, + "experience_quality": ( + procedural_experience_quality( + target_class=item.target_class, + experience=item.experience, + ) + ), "content_exposed": expose_summary, } for item in results diff --git a/loopx/capabilities/reward_memory/candidate_review.py b/loopx/capabilities/reward_memory/candidate_review.py index 91faeaea28..0848dbb753 100644 --- a/loopx/capabilities/reward_memory/candidate_review.py +++ b/loopx/capabilities/reward_memory/candidate_review.py @@ -8,6 +8,10 @@ from typing import Any from ...control_plane.runtime.public_safety import public_safe_compact_text +from .experience_quality import ( + normalize_procedural_experience, + procedural_experience_quality, +) from .registry import IDENTITY_SCOPE_FIELDS @@ -186,13 +190,14 @@ def _authority_checkpoint(raw: object) -> dict[str, Any]: def _candidate_ref(candidate: Mapping[str, Any]) -> str: identity = { - key: candidate[key] + key: candidate.get(key) for key in ( "target_class", "content_summary", "source", "scope", "guard_context", + "experience", "requested_action_scopes", ) } @@ -209,6 +214,11 @@ def _guard( requested = set(candidate["requested_action_scopes"]) target_class = candidate["target_class"] guard_context = candidate["guard_context"] + experience_quality = procedural_experience_quality( + target_class=target_class, + experience=candidate.get("experience"), + ) + reasons.extend(experience_quality["reason_codes"]) if guard_context["source_freshness"] != "current": reasons.append("source_freshness_not_current") if guard_context["conflict_state"] != "clear": @@ -246,6 +256,7 @@ def _guard( "and_no_authority_expansion" ), "semantic_reasoning_preserved": True, + "experience_quality": experience_quality, } @@ -283,6 +294,10 @@ def build_reward_memory_candidate( "lifecycle": lifecycle, "privacy": {"raw_content_captured": False}, } + if proposal.get("experience") is not None: + candidate["experience"] = normalize_procedural_experience( + proposal.get("experience") + ) candidate["candidate_ref"] = _candidate_ref(candidate) checkpoint = _authority_checkpoint(authority_checkpoint) guard = _guard(candidate, checkpoint) @@ -339,9 +354,7 @@ def review_reward_memory_candidate( if isinstance(lifecycle, Mapping) else [] ) - expires_at = ( - lifecycle.get("expires_at") if isinstance(lifecycle, Mapping) else None - ) + expires_at = lifecycle.get("expires_at") if isinstance(lifecycle, Mapping) else None if decision == "retire" and state != "active": raise ValueError("retire requires an active reviewed record") if decision != "retire" and state != "candidate": diff --git a/loopx/capabilities/reward_memory/codex_app_outcome.py b/loopx/capabilities/reward_memory/codex_app_outcome.py index 730acd4d9d..3098b54475 100644 --- a/loopx/capabilities/reward_memory/codex_app_outcome.py +++ b/loopx/capabilities/reward_memory/codex_app_outcome.py @@ -291,12 +291,22 @@ def stage_codex_app_turn_outcome_candidate( ): raise ValueError("Codex App reward memory candidate identity is incomplete") reflection = _reflection(reflection_json) - if reflection is None or reflection.get("status") == "no_evidence": + if reflection is None or reflection.get("status") in { + "no_evidence", + "legacy_no_write", + }: + legacy = ( + reflection is not None and reflection.get("status") == "legacy_no_write" + ) return _candidate_base( goal_id=goal_id, agent_id=agent_id, status="no_eligible_evidence", - reason_code="app_refresh_declared_no_reward_evidence", + reason_code=( + "legacy_reflection_requires_transferable_experience" + if legacy + else "app_refresh_declared_no_reward_evidence" + ), ) digest = _reflection_digest(reflection_json) candidate_id = "app:" + hashlib.sha256( diff --git a/loopx/capabilities/reward_memory/experience_quality.py b/loopx/capabilities/reward_memory/experience_quality.py new file mode 100644 index 0000000000..0d51c3f7c4 --- /dev/null +++ b/loopx/capabilities/reward_memory/experience_quality.py @@ -0,0 +1,173 @@ +from __future__ import annotations + +import hashlib +import json +import re +from collections.abc import Mapping, Sequence +from typing import Any + +from ...control_plane.runtime.public_safety import public_safe_compact_text + + +PROCEDURAL_EXPERIENCE_SCHEMA_VERSION = "procedural_experience_contract_v0" +PROCEDURAL_EXPERIENCE_FIELDS = { + "schema_version", + "applicability", + "observed_outcome", + "attribution", + "future_behavior", + "limitations", + "evidence_refs", +} +FUTURE_BEHAVIOR_FIELDS = {"trigger", "action", "validation", "stop_condition"} +OPAQUE_REF_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/#-]{0,199}$") +MAX_CONTEXT_ITEMS = 5 +MAX_EVIDENCE_REFS = 8 + + +def _strict_object( + value: object, + *, + label: str, + fields: set[str], +) -> Mapping[str, Any]: + if not isinstance(value, Mapping): + raise ValueError(f"{label} must be an object") + missing = sorted(fields - set(value)) + unknown = sorted(set(value) - fields) + if missing or unknown: + raise ValueError( + f"{label} has invalid fields: missing={missing}, unknown={unknown}" + ) + return value + + +def _compact(value: object, label: str, *, limit: int = 500) -> str: + result = public_safe_compact_text(value, limit=limit) + if not result: + raise ValueError(f"{label} must be compact and public-safe") + return result + + +def _compact_list(value: object, label: str) -> list[str]: + if ( + not isinstance(value, Sequence) + or isinstance(value, (str, bytes)) + or not 1 <= len(value) <= MAX_CONTEXT_ITEMS + ): + raise ValueError( + f"{label} must contain between 1 and {MAX_CONTEXT_ITEMS} items" + ) + result = [_compact(item, label) for item in value] + if len(set(result)) != len(result): + raise ValueError(f"{label} must not contain duplicates") + return result + + +def _evidence_refs(value: object) -> list[str]: + if ( + not isinstance(value, Sequence) + or isinstance(value, (str, bytes)) + or not 1 <= len(value) <= MAX_EVIDENCE_REFS + ): + raise ValueError( + "experience.evidence_refs must contain between 1 and " + f"{MAX_EVIDENCE_REFS} opaque references" + ) + result = [str(item or "").strip() for item in value] + if any(not OPAQUE_REF_RE.fullmatch(item) for item in result): + raise ValueError( + "experience.evidence_refs must contain compact opaque references" + ) + if len(set(result)) != len(result): + raise ValueError("experience.evidence_refs must not contain duplicates") + return result + + +def normalize_procedural_experience(value: object) -> dict[str, Any]: + """Normalize the transferable lesson required by procedural memory. + + The contract is structural rather than keyword based. It separates the + situation, independently observed result, attribution, future behavior, + and non-generalization boundary so a fact-only summary cannot become an + active procedural experience merely because persistence succeeded. + """ + + raw = _strict_object( + value, + label="experience", + fields=PROCEDURAL_EXPERIENCE_FIELDS, + ) + if raw.get("schema_version") != PROCEDURAL_EXPERIENCE_SCHEMA_VERSION: + raise ValueError(f"experience must use {PROCEDURAL_EXPERIENCE_SCHEMA_VERSION}") + future = _strict_object( + raw.get("future_behavior"), + label="experience.future_behavior", + fields=FUTURE_BEHAVIOR_FIELDS, + ) + return { + "schema_version": PROCEDURAL_EXPERIENCE_SCHEMA_VERSION, + "applicability": _compact_list( + raw.get("applicability"), "experience.applicability" + ), + "observed_outcome": _compact( + raw.get("observed_outcome"), "experience.observed_outcome" + ), + "attribution": _compact(raw.get("attribution"), "experience.attribution"), + "future_behavior": { + key: _compact(future.get(key), f"experience.future_behavior.{key}") + for key in ("trigger", "action", "validation", "stop_condition") + }, + "limitations": _compact_list(raw.get("limitations"), "experience.limitations"), + "evidence_refs": _evidence_refs(raw.get("evidence_refs")), + } + + +def procedural_experience_quality( + *, + target_class: str, + experience: object, +) -> dict[str, Any]: + """Return a deterministic qualification receipt for one memory class.""" + + required = target_class == "procedural_experience" + if experience is None: + return { + "required": required, + "passed": not required, + "status": "missing" if required else "not_applicable", + "contract_version": PROCEDURAL_EXPERIENCE_SCHEMA_VERSION, + "reason_codes": ( + ["procedural_experience_quality_contract_missing"] if required else [] + ), + "value_status": "unproven", + } + normalized = normalize_procedural_experience(experience) + return { + "required": required, + "passed": True, + "status": "qualified_for_activation" if required else "supporting_context", + "contract_version": PROCEDURAL_EXPERIENCE_SCHEMA_VERSION, + "experience_digest": procedural_experience_digest(normalized), + "reason_codes": [], + "value_status": "unproven_until_application_evidence", + } + + +def procedural_experience_digest(value: object) -> str: + normalized = normalize_procedural_experience(value) + canonical = json.dumps( + normalized, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + return "sha256:" + hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +__all__ = [ + "PROCEDURAL_EXPERIENCE_SCHEMA_VERSION", + "normalize_procedural_experience", + "procedural_experience_digest", + "procedural_experience_quality", +] diff --git a/loopx/capabilities/reward_memory/ingestion.py b/loopx/capabilities/reward_memory/ingestion.py index 788930549d..b8c3b5da31 100644 --- a/loopx/capabilities/reward_memory/ingestion.py +++ b/loopx/capabilities/reward_memory/ingestion.py @@ -25,6 +25,7 @@ TARGET_CLASS_IDS, review_reward_memory_candidate, ) +from .experience_quality import procedural_experience_digest from .registry import IDENTITY_SCOPE_FIELDS, normalize_reward_memory_corpus @@ -276,6 +277,7 @@ def _policy_guard( return { "passed": not reasons, "reason_codes": sorted(set(reasons)), + "experience_quality": dict(guard.get("experience_quality") or {}), "semantic_reasoning_preserved": True, "rule": ( "standing_policy_checks_exact_owner_class_source_actor_scope_and_" @@ -337,6 +339,7 @@ def ingest_reward_memory_candidate( "policy_id": policy["policy_id"], "surface_ids": surfaces, "guard": guard, + "experience_quality": dict(guard.get("experience_quality") or {}), "deduplicated": False, "exact_readback_verified": False, "memory_available_for_recall": False, @@ -496,10 +499,103 @@ def ingest_reward_memory_candidate( and item.content_digest == expected_digest for item in recall.items ) + qualified = (guard.get("experience_quality") or {}).get("passed") is True + destination_required = candidate.get("target_class") == "procedural_experience" + destination_verified = not destination_required + destination_recall = None + experience_digest = None + if exact and destination_required: + experience = candidate.get("experience") + assert isinstance(experience, Mapping) + experience_digest = procedural_experience_digest(experience) + applicability = list(experience.get("applicability") or []) + future_behavior = experience.get("future_behavior") + assert isinstance(future_behavior, Mapping) + destination_request = build_reward_memory_recall_request( + normalized_corpus, + { + "workspace_ref": scope["workspace_ref"], + "project_ref": scope["project_ref"], + **identity_scope, + "surface_id": surface_id, + "revision_ref": scope.get("revision_ref"), + "mode": "function_boundary", + "query_kind": "business_recall", + "queries": [ + { + "query": " ".join( + [ + str(applicability[0]), + str(future_behavior["trigger"]), + str(future_behavior["action"]), + ] + ), + "query_summary": ( + "Verify semantic recall at the configured destination " + "surface." + ), + } + ], + "limit": 3, + "observed_at": observed_at, + "freshness_context": _freshness_context( + normalized_corpus, + scope.get("revision_ref"), + ), + "conflict_state": "clear", + "raw_content_captured": False, + }, + read_authority_checkpoint={ + "verified": True, + "corpus_id": normalized_corpus["corpus_id"], + "workspace_ref": scope["workspace_ref"], + "project_ref": scope["project_ref"], + **identity_scope, + "surface_id": surface_id, + "read_authority": normalized_corpus["read_authority"], + "source_ref": policy["authority_source_ref"], + }, + ) + destination_recall = execute_reward_memory_recall( + destination_request, + provider_binding=binding, + provider=configured_provider, + ) + destination_verified = any( + item.candidate_ref == candidate_ref + and item.experience_digest == experience_digest + for item in destination_recall.items + ) + available = exact and qualified and destination_verified + status = ( + "activated" + if available + else "readback_unverified" + if not exact + else "recall_unverified" + ) + reason_codes = ( + [] + if available + else ["exact_provider_readback_unverified"] + if not exact + else ["destination_business_recall_unverified"] + ) return synced | { - "status": "activated" if exact else "readback_unverified", + "status": status, "readback": recall.public_packet, "exact_readback_verified": exact, - "memory_available_for_recall": exact, - "reason_codes": [] if exact else ["exact_provider_readback_unverified"], + "destination_recall": { + "required": destination_required, + "verified": destination_verified, + "query_kind": "business_recall" if destination_required else None, + "experience_digest": experience_digest, + "receipt": ( + destination_recall.public_packet + if destination_recall is not None + else None + ), + }, + "memory_available_for_recall": available, + "reason_codes": reason_codes, } diff --git a/loopx/capabilities/reward_memory/outcome_lifecycle.py b/loopx/capabilities/reward_memory/outcome_lifecycle.py index ae4026df8c..a5e363165b 100644 --- a/loopx/capabilities/reward_memory/outcome_lifecycle.py +++ b/loopx/capabilities/reward_memory/outcome_lifecycle.py @@ -18,6 +18,7 @@ resolve_reward_memory_experiment, resolve_reward_memory_surface_config, ) +from .experience_quality import normalize_procedural_experience from .runtime_hooks import run_reward_memory_automatic_ingest_hook from .scoped_feedback import ( SCOPED_FEEDBACK_ADAPTER, @@ -26,17 +27,22 @@ ) -TURN_REWARD_MEMORY_REFLECTION_SCHEMA_VERSION = "turn_reward_memory_reflection_v0" +TURN_REWARD_MEMORY_REFLECTION_SCHEMA_VERSION = "turn_reward_memory_reflection_v1" +LEGACY_TURN_REWARD_MEMORY_REFLECTION_SCHEMA_VERSION = "turn_reward_memory_reflection_v0" TURN_REWARD_MEMORY_INGEST_SCHEMA_VERSION = "turn_reward_memory_ingest_v0" TURN_REWARD_MEMORY_RECONCILIATION_SCHEMA_VERSION = ( "turn_reward_memory_reconciliation_v0" ) TURN_REWARD_MEMORY_SIDECAR_SCHEMA_VERSION = "turn_reward_memory_sidecar_v0" +LEGACY_PENDING_MIGRATION_SCHEMA_VERSION = ( + "turn_reward_memory_legacy_pending_migration_v0" +) _SAFE_PATH_TOKEN = re.compile(r"[^A-Za-z0-9._-]+") _PENDING_PROVIDER_STATUSES = { "committed_pending", "provider_unavailable", "readback_unverified", + "recall_unverified", } _OUTCOME_SOURCE_KINDS = { "research": "research_review", @@ -44,7 +50,7 @@ "real": "real_outcome_review", "engineering": "engineering_review", } -_REFLECTION_FIELDS = { +_LEGACY_REFLECTION_FIELDS = { "schema_version", "status", "surface_id", @@ -54,6 +60,7 @@ "confidence", "evidence_refs", } +_REFLECTION_FIELDS = _LEGACY_REFLECTION_FIELDS | {"experience"} def _base( @@ -129,8 +136,7 @@ def _load_sidecar(path: Path) -> dict[str, Any] | None: raise ValueError("reward memory outcome sidecar is unreadable") from exc if ( not isinstance(payload, dict) - or payload.get("schema_version") - != TURN_REWARD_MEMORY_SIDECAR_SCHEMA_VERSION + or payload.get("schema_version") != TURN_REWARD_MEMORY_SIDECAR_SCHEMA_VERSION ): raise ValueError("reward memory outcome sidecar is invalid") return payload @@ -167,6 +173,59 @@ def _replayed_completed_receipt(value: Mapping[str, Any]) -> dict[str, Any]: } +def _terminalize_legacy_pending_sidecar( + path: Path, + value: Mapping[str, Any], + *, + goal_id: str, + agent_id: str, + migrated_at: str, +) -> dict[str, Any] | None: + reflection = value.get("reflection") + if not isinstance(reflection, Mapping): + return None + normalized = _reflection( + json.dumps( + reflection, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + ) + if normalized is None or normalized.get("status") != "legacy_no_write": + return None + migration_receipt = { + "schema_version": LEGACY_PENDING_MIGRATION_SCHEMA_VERSION, + "status": "terminal_rejected", + "reason_code": "legacy_reflection_requires_transferable_experience", + "provider_write_state": "unknown_may_have_committed", + "provider_cleanup_performed": False, + "migrated_at": migrated_at, + } + public_receipt = _base( + goal_id=goal_id, + agent_id=agent_id, + status="no_eligible_evidence", + reason_code="legacy_reflection_requires_transferable_experience", + ) | { + "automatic_ingest": True, + "source_event_id": str(value.get("source_event_id") or "") or None, + "surface_id": str(value.get("surface_id") or "") or None, + "reconciliation_state": "rejected", + "legacy_pending_migration": migration_receipt, + } + _write_sidecar( + path, + dict(value) + | { + "status": "rejected", + "public_receipt": public_receipt, + "legacy_pending_migration": migration_receipt, + }, + ) + return public_receipt + + def _reflection(value: object) -> dict[str, Any] | None: text = str(value or "").strip() if not text: @@ -177,19 +236,33 @@ def _reflection(value: object) -> dict[str, Any] | None: raw = json.loads(text) except json.JSONDecodeError as exc: raise ValueError("reward memory reflection must be JSON") from exc - if not isinstance(raw, dict) or set(raw) - _REFLECTION_FIELDS: + if not isinstance(raw, dict): + raise ValueError("reward memory reflection must decode to an object") + schema_version = raw.get("schema_version") + allowed_fields = ( + _LEGACY_REFLECTION_FIELDS + if schema_version == LEGACY_TURN_REWARD_MEMORY_REFLECTION_SCHEMA_VERSION + else _REFLECTION_FIELDS + ) + if set(raw) - allowed_fields: raise ValueError("reward memory reflection contains unsupported fields") - if raw.get("schema_version") != TURN_REWARD_MEMORY_REFLECTION_SCHEMA_VERSION: + if schema_version not in { + LEGACY_TURN_REWARD_MEMORY_REFLECTION_SCHEMA_VERSION, + TURN_REWARD_MEMORY_REFLECTION_SCHEMA_VERSION, + }: raise ValueError("reward memory reflection schema is unsupported") status = str(raw.get("status") or "") if status == "no_evidence": return {"schema_version": raw["schema_version"], "status": status} if status != "eligible": raise ValueError("reward memory reflection status is unsupported") + if schema_version == LEGACY_TURN_REWARD_MEMORY_REFLECTION_SCHEMA_VERSION: + return { + "schema_version": schema_version, + "status": "legacy_no_write", + } surface_id = public_safe_compact_text(raw.get("surface_id"), limit=160) - content_summary = public_safe_compact_text( - raw.get("content_summary"), limit=500 - ) + content_summary = public_safe_compact_text(raw.get("content_summary"), limit=500) reasoning_summary = public_safe_compact_text( raw.get("reasoning_summary"), limit=500 ) @@ -214,6 +287,11 @@ def _reflection(value: object) -> dict[str, Any] | None: compact_refs.append(ref) if len(set(compact_refs)) != len(compact_refs): raise ValueError("reward memory evidence refs must be unique") + experience = normalize_procedural_experience(raw.get("experience")) + if experience["evidence_refs"] != compact_refs: + raise ValueError( + "experience.evidence_refs must match the validated reflection evidence_refs" + ) return { "schema_version": raw["schema_version"], "status": status, @@ -223,6 +301,7 @@ def _reflection(value: object) -> dict[str, Any] | None: "reasoning_summary": reasoning_summary, "confidence": confidence, "evidence_refs": compact_refs, + "experience": experience, } @@ -273,8 +352,7 @@ def _validated_reflection_evidence( if not isinstance(validation, Mapping): return None if ( - validation.get("schema_version") - != "reward_memory_reflection_validation_v0" + validation.get("schema_version") != "reward_memory_reflection_validation_v0" or validation.get("status") != "validated" or validation.get("reflection_digest") != _reflection_digest(reflection_json) or validation.get("evidence_refs") != reflection.get("evidence_refs") @@ -294,7 +372,7 @@ def _event_scope( *, surface_id: str, agent_id: str, -) -> tuple[dict[str, Any], str]: +) -> tuple[dict[str, Any], Mapping[str, Any]]: route = resolve_reward_memory_surface_config( config, surface_id, @@ -311,7 +389,7 @@ def _event_scope( "user_ref": scope.get("user_ref"), "peer_ref": peer_ref, "session_ref": scope.get("session_ref"), - }, str(route["standing_policy"]["policy_id"]) + }, route def run_configured_turn_outcome_ingest( @@ -347,19 +425,28 @@ def run_configured_turn_outcome_ingest( reason_code="automatic_ingest_explicitly_disabled", ) reflection = _reflection(host_result.get("reward_memory_reflection_json")) - if reflection is None or reflection["status"] == "no_evidence": + if reflection is None or reflection["status"] in { + "no_evidence", + "legacy_no_write", + }: + legacy = reflection is not None and reflection["status"] == "legacy_no_write" return _base( goal_id=goal_id, agent_id=agent_id, status="no_eligible_evidence", - reason_code="validated_turn_declared_no_reward_evidence", + reason_code=( + "legacy_reflection_requires_transferable_experience" + if legacy + else "validated_turn_declared_no_reward_evidence" + ), ) | {"automatic_ingest": True} surface_id = str(reflection["surface_id"]) - scope, policy_id = _event_scope( + scope, route = _event_scope( config, surface_id=surface_id, agent_id=agent_id, ) + policy_id = str(route["standing_policy"]["policy_id"]) evidence_digest = hashlib.sha256( json.dumps( reflection["evidence_refs"], @@ -380,8 +467,9 @@ def run_configured_turn_outcome_ingest( **scope, "surface_id": surface_id, "revision_ref": turn_key, - "target_class": "soft_preference", + "target_class": str(route["corpus"]["class_id"]), "content_summary": reflection["content_summary"], + "experience": reflection["experience"], "source": { "source_kind": _OUTCOME_SOURCE_KINDS[reflection["outcome_kind"]], "source_ref": source_ref, @@ -486,9 +574,7 @@ def run_configured_turn_outcome_ingest( else (previous or {}).get("settlement_validation") ) settlement_validation = ( - settlement_validation - if isinstance(settlement_validation, Mapping) - else {} + settlement_validation if isinstance(settlement_validation, Mapping) else {} ) attempt_count = int((previous or {}).get("attempt_count") or 0) + 1 pending_sidecar = { @@ -536,6 +622,10 @@ def run_configured_turn_outcome_ingest( guard = receipt.get("guard") guard = guard if isinstance(guard, Mapping) else {} guard_reason_codes = guard.get("reason_codes") + destination_recall = receipt.get("destination_recall") + destination_recall = ( + destination_recall if isinstance(destination_recall, Mapping) else {} + ) public_receipt = _base( goal_id=goal_id, agent_id=agent_id, @@ -552,11 +642,16 @@ def run_configured_turn_outcome_ingest( "surface_id": surface_id, "source_event_id": source_event_id, "provider_sync_count": int(telemetry.get("provider_sync_count") or 0), - "exact_readback_verified": bool( - telemetry.get("exact_readback_verified") - ), + "exact_readback_verified": bool(telemetry.get("exact_readback_verified")), "deduplicated": bool(telemetry.get("deduplicated")), "external_writes_performed": bool(result.get("external_writes_performed")), + "experience_quality": dict(receipt.get("experience_quality") or {}), + "destination_recall": { + "required": bool(destination_recall.get("required")), + "verified": bool(destination_recall.get("verified")), + "query_kind": destination_recall.get("query_kind"), + "experience_digest": destination_recall.get("experience_digest"), + }, } reconciliation_state = ( "completed" @@ -631,7 +726,7 @@ def reconcile_pending_turn_outcome_ingests( ).parent if not directory.is_dir(): return base | {"status": "empty"} - pending: list[dict[str, Any]] = [] + pending: list[tuple[Path, dict[str, Any]]] = [] for path in sorted(directory.glob("*.json")): value = _load_sidecar(path) if ( @@ -640,9 +735,20 @@ def reconcile_pending_turn_outcome_ingests( and value.get("agent_id") == agent_id and value.get("status") == "pending" ): - pending.append(value) + pending.append((path, value)) receipts: list[dict[str, Any]] = [] - for value in pending[:limit]: + effective_observed_at = observed_at or datetime.now(timezone.utc).isoformat() + for path, value in pending[:limit]: + legacy_receipt = _terminalize_legacy_pending_sidecar( + path, + value, + goal_id=goal_id, + agent_id=agent_id, + migrated_at=effective_observed_at, + ) + if legacy_receipt is not None: + receipts.append(legacy_receipt) + continue reflection = value.get("reflection") if not isinstance(reflection, Mapping): continue @@ -665,9 +771,7 @@ def reconcile_pending_turn_outcome_ingests( "task_validation": { "ok": True, "validator_kind": str( - (value.get("reflection_validation") or {}).get( - "validator_kind" - ) + (value.get("reflection_validation") or {}).get("validator_kind") or "reconciliation" ), "reward_memory_reflection_validation": dict( @@ -675,12 +779,10 @@ def reconcile_pending_turn_outcome_ingests( ), }, "writeback": dict( - (value.get("settlement_validation") or {}).get("writeback") - or {} + (value.get("settlement_validation") or {}).get("writeback") or {} ), "quota_spend": dict( - (value.get("settlement_validation") or {}).get("quota_spend") - or {} + (value.get("settlement_validation") or {}).get("quota_spend") or {} ), }, provider=provider, @@ -691,8 +793,7 @@ def reconcile_pending_turn_outcome_ingests( "pending_count": len(pending), "attempted_count": len(receipts), "completed_count": sum( - receipt.get("reconciliation_state") == "completed" - for receipt in receipts + receipt.get("reconciliation_state") == "completed" for receipt in receipts ), "provider_sync_count": sum( int(receipt.get("provider_sync_count") or 0) for receipt in receipts diff --git a/loopx/capabilities/reward_memory/scoped_feedback.py b/loopx/capabilities/reward_memory/scoped_feedback.py index 6c7b050f07..58d69760e8 100644 --- a/loopx/capabilities/reward_memory/scoped_feedback.py +++ b/loopx/capabilities/reward_memory/scoped_feedback.py @@ -31,6 +31,7 @@ "revision_ref", "target_class", "content_summary", + "experience", "source", "reasoning", "guard_context", @@ -92,6 +93,7 @@ def build_scoped_feedback_reward_memory_candidate( { "target_class": raw.get("target_class"), "content_summary": raw.get("content_summary"), + "experience": raw.get("experience"), "source": raw.get("source"), "scope": { "workspace_ref": raw.get("workspace_ref"), diff --git a/loopx/cli_commands/project_lifecycle_refresh_state.py b/loopx/cli_commands/project_lifecycle_refresh_state.py index 484cf9bb4c..3a6c4ff635 100644 --- a/loopx/cli_commands/project_lifecycle_refresh_state.py +++ b/loopx/cli_commands/project_lifecycle_refresh_state.py @@ -198,7 +198,7 @@ def register_refresh_state_command( refresh_state_parser.add_argument( "--reward-memory-reflection-json", help=( - "Optional compact turn_reward_memory_reflection_v0 JSON for a " + "Optional compact turn_reward_memory_reflection_v1 JSON for a " "Todo-bound accountable Codex App refresh. LoopX stages it privately, " "requires the Todo's caller-declared validator to attest the exact " "reflection and evidence, and performs no provider write until the " diff --git a/loopx/control_plane/heartbeat/rules.py b/loopx/control_plane/heartbeat/rules.py index a33945c7ce..3332f722ab 100644 --- a/loopx/control_plane/heartbeat/rules.py +++ b/loopx/control_plane/heartbeat/rules.py @@ -28,7 +28,7 @@ ) REWARD_MEMORY_OUTCOME_RULE = ( "`reward_memory_recall.experiment.automatic_ingest=true`: reusable Todo outcomes " - "add `--reward-memory-reflection-json ` " + "add `--reward-memory-reflection-json ` " "to refresh. LoopX stages privately; provider ingest needs caller-declared Todo " "validator to attest exact reflection digest/evidence, then exact writeback/spend " "readback. Missing attestation stays awaiting; zero provider calls. Never include " diff --git a/loopx/control_plane/turn_driver/codex_cli.py b/loopx/control_plane/turn_driver/codex_cli.py index f4db8b7e86..4b95984412 100644 --- a/loopx/control_plane/turn_driver/codex_cli.py +++ b/loopx/control_plane/turn_driver/codex_cli.py @@ -400,7 +400,7 @@ def _prompt(request: Mapping[str, Any]) -> str: "Execute exactly one bounded LoopX Turn in the current workspace.", "Use the TurnEnvelope as the source of truth. Perform work only when its contract allows it.", "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.", - "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_v0, 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. 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.", + "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.", "Do not write LoopX state, spend quota, or apply scheduler changes; the adapter owns those effects.", "Return only the schema-constrained result. For validated_progress, repair_required, or replan_required, fill every material field with public-safe evidence.", "For those material results, set path_delta_mode=material_replan only when this Turn changes a prior assumption, route, scope, acceptance rule, or stops prior work; then provide a complete bounded agent vision packet with goal_path_delta_v0 in agent_vision_json and leave vision_unchanged_reason empty.", @@ -499,10 +499,7 @@ def _diagnostic_failure_category(line: str) -> str | None: ) ): return "quota_exhausted" - if any( - marker in text - for marker in ("rate limit", "too many requests") - ): + if any(marker in text for marker in ("rate limit", "too many requests")): return "rate_limited" if "session" in text and "not found" in text: return "session_missing" @@ -624,9 +621,7 @@ def _event_failure_categories( candidate = container.get(field) if not _meaningful_structured_value(candidate): continue - code_categories.append( - _structured_failure_category(candidate) or "unknown" - ) + code_categories.append(_structured_failure_category(candidate) or "unknown") if code_categories: return _select_failure_category(code_categories), None @@ -850,7 +845,9 @@ def run_codex_cli_host( stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, - text=True, encoding="utf-8", errors="replace", + text=True, + encoding="utf-8", + errors="replace", start_new_session=True, ) observed_session: list[str] = [] @@ -927,8 +924,7 @@ def discard_stderr() -> None: if returncode != 0 and category in SESSION_INVALIDATING_FAILURE_CATEGORIES: _discard_codex_cli_session(runtime_root, lineage=lineage) if observed_session and ( - returncode == 0 - or category not in SESSION_INVALIDATING_FAILURE_CATEGORIES + returncode == 0 or category not in SESSION_INVALIDATING_FAILURE_CATEGORIES ): _store_codex_cli_session( runtime_root, diff --git a/tests/capabilities/test_reward_memory_codex_app_outcome.py b/tests/capabilities/test_reward_memory_codex_app_outcome.py index 0ee39ec51d..ad25673619 100644 --- a/tests/capabilities/test_reward_memory_codex_app_outcome.py +++ b/tests/capabilities/test_reward_memory_codex_app_outcome.py @@ -16,16 +16,44 @@ def _reflection() -> str: + evidence_refs = ["artifact:app-route", "receipt:app-validation"] return json.dumps( { - "schema_version": "turn_reward_memory_reflection_v0", + "schema_version": "turn_reward_memory_reflection_v1", "status": "eligible", "surface_id": "agent_workflow.turn_admission", "outcome_kind": "engineering", "content_summary": "Run the exact admission test before changing routing.", "reasoning_summary": "The independently checked route avoided stale state.", "confidence": "high", - "evidence_refs": ["artifact:app-route", "receipt:app-validation"], + "evidence_refs": evidence_refs, + "experience": { + "schema_version": "procedural_experience_contract_v0", + "applicability": ["Changing a managed Turn admission route"], + "observed_outcome": ( + "The exact admission test avoided a route based on stale state." + ), + "attribution": ( + "Independent validation bound the route result to current " + "admission evidence." + ), + "future_behavior": { + "trigger": "A managed Turn admission route may change.", + "action": "Run the exact admission test before changing routing.", + "validation": ( + "Bind the result to the declared validator and current route " + "receipt." + ), + "stop_condition": ( + "Do not change routing when the validator or current receipt " + "is missing." + ), + }, + "limitations": [ + "This result applies only to the validated route and revision." + ], + "evidence_refs": evidence_refs, + }, } ) diff --git a/tests/capabilities/test_reward_memory_experience_quality.py b/tests/capabilities/test_reward_memory_experience_quality.py new file mode 100644 index 0000000000..77fc46c936 --- /dev/null +++ b/tests/capabilities/test_reward_memory_experience_quality.py @@ -0,0 +1,352 @@ +from __future__ import annotations + +import hashlib +import json +from typing import Any + +import pytest + +from loopx.capabilities.context_providers.base import ( + ContextProviderItem, + ContextProviderRetrieval, +) +from loopx.capabilities.reward_memory.application import _active_item +from loopx.capabilities.reward_memory.scoped_feedback import ( + build_scoped_feedback_reward_memory_candidate, + ingest_scoped_feedback_reward_memory_event, +) +from tests.capabilities.test_reward_memory_ingestion import FakeProvider + + +OBSERVED_AT = "2026-09-20T17:00:00+00:00" +WORKSPACE = "workspace:finance" +PROJECT = "project:finance-research" +SURFACE = "finance.trade.precheck" +SCOPE_REF = "viking://resources/reward-memory/finance-trade" + + +class DestinationMissProvider(FakeProvider): + def retrieve(self, **kwargs: Any) -> ContextProviderRetrieval: + if self.retrieve_calls == 0: + return super().retrieve(**kwargs) + self.retrieve_calls += 1 + return ContextProviderRetrieval( + provider=self.provider_id, + namespace=str(kwargs["namespace"]), + status="completed", + query_summary=str(kwargs["query_summary"]), + observed_at=str(kwargs["observed_at"]), + search_performed=True, + read_performed=True, + items=(), + requested_limit=int(kwargs["max_results"]), + ) + + +def experience() -> dict[str, Any]: + return { + "schema_version": "procedural_experience_contract_v0", + "applicability": [ + "Considering a short after price and open interest accelerate together" + ], + "observed_outcome": ( + "A NEAR short planned near 3.80 would have reached its 4.15 stop before " + "costs." + ), + "attribution": ( + "The thesis relied on unrealized value capture while price and open " + "interest were still accelerating without exhaustion confirmation." + ), + "future_behavior": { + "trigger": ( + "A short thesis conflicts with accelerating price and open interest." + ), + "action": ( + "Wait for exhaustion or a failed breakout before proposing entry." + ), + "validation": ( + "Recheck price, open interest, funding, liquidity, and invalidation at " + "the executable quote." + ), + "stop_condition": ( + "Abandon the entry when momentum remains intact or the quote exceeds " + "the precomputed invalidation." + ), + }, + "limitations": [ + "One counterfactual loss does not prove all momentum shorts are invalid.", + "Execution delay is not evidence that the original thesis was correct.", + ], + "evidence_refs": ["artifact:near-postmortem", "receipt:market-readback"], + } + + +def event() -> dict[str, Any]: + return { + "schema_version": "scoped_feedback_reward_memory_event_v0", + "feedback_ref": "trade-review:near-3p8-short", + "workspace_ref": WORKSPACE, + "project_ref": PROJECT, + "peer_ref": "agent:finance-research-explorer", + "surface_id": SURFACE, + "revision_ref": "revision:near-postmortem-v1", + "target_class": "procedural_experience", + "content_summary": ( + "Do not short accelerating price and open interest without exhaustion " + "confirmation." + ), + "experience": experience(), + "source": { + "source_kind": "real_outcome_review", + "source_ref": "trade-review:near-3p8-short", + "actor_ref": "agent:finance-research-explorer", + "actor_role": "validated_goal_agent", + }, + "reasoning": { + "summary": "A counterfactual stop was verified against market readback.", + "confidence": "high", + }, + "guard_context": { + "source_freshness": "current", + "conflict_state": "clear", + "current_artifact_verified": True, + }, + "requested_action_scopes": [], + "raw_content_captured": False, + } + + +def corpus() -> dict[str, Any]: + return { + "corpus_id": "finance_trade_outcome", + "class_id": "procedural_experience", + "provider_id": "fake_provider", + "owner_ref": "finance_reward_memory_owner", + "source_of_truth": "verified_trade_outcome", + "read_authority": "module_scoped", + "write_authority": "provider_managed", + "scope": { + "workspace_ref": WORKSPACE, + "project_ref": PROJECT, + "peer_ref": "agent:finance-research-explorer", + "surface_ids": [SURFACE], + }, + "freshness": {"mode": "source_truth_bound"}, + "lifecycle": {"state": "active", "supersedes": []}, + "retrieval": { + "index_required": True, + "readback_required": True, + "application_receipt_required": True, + }, + "maintenance": { + "writeback_triggers": ["real_outcome_review"], + "closure_policy": "write_exact_readback_then_recall", + "retirement_authority": "finance_reward_memory_owner", + }, + "privacy": {"visibility": "private", "raw_content_in_registry": False}, + "provider_scope_ref_digest": hashlib.sha256( + SCOPE_REF.encode("utf-8") + ).hexdigest()[:16], + } + + +def policy() -> dict[str, Any]: + return { + "schema_version": "reward_memory_standing_policy_v0", + "policy_id": "policy:finance:trade-outcome", + "enabled": True, + "auto_activate": True, + "owner_ref": "finance_reward_memory_owner", + "reviewer_ref": "agent:finance-research-explorer", + "authority_source_ref": "policy:finance:reward-memory", + "scope": { + "workspace_ref": WORKSPACE, + "project_ref": PROJECT, + "peer_ref": "agent:finance-research-explorer", + "surface_ids": [SURFACE], + }, + "allowed_target_classes": ["procedural_experience"], + "allowed_source_kinds": ["real_outcome_review"], + "allowed_actor_roles": ["validated_goal_agent"], + "allowed_action_scopes": [], + "raw_content_captured": False, + } + + +def binding() -> dict[str, Any]: + return { + "corpus_id": "finance_trade_outcome", + "provider_id": "fake_provider", + "namespace": "reward_memory", + "scope_ref": SCOPE_REF, + "timeout_seconds": 5, + "actor_peer_id": "finance-research-explorer", + } + + +def test_fact_only_summary_is_blocked_before_provider_write() -> None: + fact_only = event() + fact_only.pop("experience") + candidate = build_scoped_feedback_reward_memory_candidate(fact_only)[ + "shared_candidate" + ] + provider = FakeProvider() + + receipt = ingest_scoped_feedback_reward_memory_event( + fact_only, + corpus=corpus(), + standing_policy=policy(), + provider_binding=binding(), + observed_at=OBSERVED_AT, + execute=True, + provider=provider, + ) + + assert candidate["status"] == "guard_blocked" + assert candidate["guard"]["reason_codes"] == [ + "procedural_experience_quality_contract_missing" + ] + assert receipt["status"] == "guard_blocked" + assert receipt["experience_quality"]["passed"] is False + assert provider.sync_calls == 0 + assert provider.retrieve_calls == 0 + + +def test_missing_future_behavior_field_is_invalid_before_provider_write() -> None: + incomplete = event() + del incomplete["experience"]["future_behavior"]["action"] + provider = FakeProvider() + + with pytest.raises( + ValueError, match="experience.future_behavior has invalid fields" + ): + ingest_scoped_feedback_reward_memory_event( + incomplete, + corpus=corpus(), + standing_policy=policy(), + provider_binding=binding(), + observed_at=OBSERVED_AT, + execute=True, + provider=provider, + ) + + assert provider.sync_calls == 0 + assert provider.retrieve_calls == 0 + + +def test_finance_experience_cannot_cross_to_a_different_surface() -> None: + wrong_surface = event() | {"surface_id": "finance.research.review"} + provider = FakeProvider() + + receipt = ingest_scoped_feedback_reward_memory_event( + wrong_surface, + corpus=corpus(), + standing_policy=policy(), + provider_binding=binding(), + observed_at=OBSERVED_AT, + execute=True, + provider=provider, + ) + + assert receipt["status"] == "guard_blocked" + assert "candidate_surface_not_allowed" in receipt["guard"]["reason_codes"] + assert "candidate_surface_corpus_mismatch" in receipt["guard"]["reason_codes"] + assert provider.sync_calls == 0 + assert provider.retrieve_calls == 0 + + +def test_qualified_experience_is_preserved_and_exactly_read_back() -> None: + provider = FakeProvider() + receipt = ingest_scoped_feedback_reward_memory_event( + event(), + corpus=corpus(), + standing_policy=policy(), + provider_binding=binding(), + observed_at=OBSERVED_AT, + execute=True, + provider=provider, + ) + + assert receipt["status"] == "activated" + assert receipt["exact_readback_verified"] is True + assert receipt["memory_available_for_recall"] is True + assert receipt["experience_quality"]["required"] is True + assert receipt["experience_quality"]["passed"] is True + assert receipt["experience_quality"]["status"] == "qualified_for_activation" + assert receipt["experience_quality"]["experience_digest"].startswith("sha256:") + assert receipt["experience_quality"]["value_status"] == ( + "unproven_until_application_evidence" + ) + assert receipt["destination_recall"]["verified"] is True + assert receipt["destination_recall"]["query_kind"] == "business_recall" + assert provider.retrieve_calls == 2 + stored = json.loads(next(iter(provider.resources.values()))) + assert stored["experience"] == experience() + assert stored["experience_quality"] == receipt["experience_quality"] + + +def test_exact_readback_without_destination_recall_is_not_available() -> None: + provider = DestinationMissProvider() + receipt = ingest_scoped_feedback_reward_memory_event( + event(), + corpus=corpus(), + standing_policy=policy(), + provider_binding=binding(), + observed_at=OBSERVED_AT, + execute=True, + provider=provider, + ) + + assert receipt["exact_readback_verified"] is True + assert receipt["destination_recall"]["verified"] is False + assert receipt["status"] == "recall_unverified" + assert receipt["memory_available_for_recall"] is False + assert receipt["reason_codes"] == ["destination_business_recall_unverified"] + assert provider.sync_calls == 1 + assert provider.retrieve_calls == 2 + + +def test_legacy_fact_only_procedural_record_is_not_recalled() -> None: + provider = FakeProvider() + receipt = ingest_scoped_feedback_reward_memory_event( + event(), + corpus=corpus(), + standing_policy=policy(), + provider_binding=binding(), + observed_at=OBSERVED_AT, + execute=True, + provider=provider, + ) + target, serialized = next(iter(provider.resources.items())) + legacy = json.loads(serialized) + legacy.pop("experience") + legacy.pop("experience_quality") + + item = ContextProviderItem( + resource_ref=target, + summary="Legacy fact-only record", + content=json.dumps(legacy), + score=0.99, + ) + + assert receipt["status"] == "activated" + assert ( + _active_item( + item, + corpus(), + surface_id=SURFACE, + observed_at=OBSERVED_AT, + ) + is None + ) + + +def test_future_behavior_changes_candidate_identity() -> None: + first = build_scoped_feedback_reward_memory_candidate(event())["shared_candidate"] + changed = event() + changed["experience"]["future_behavior"]["action"] = ( + "Require a failed breakout and declining open interest before entry." + ) + second = build_scoped_feedback_reward_memory_candidate(changed)["shared_candidate"] + + assert first["candidate"]["candidate_ref"] != second["candidate"]["candidate_ref"] diff --git a/tests/capabilities/test_reward_memory_outcome_lifecycle.py b/tests/capabilities/test_reward_memory_outcome_lifecycle.py index f271740348..f883daab18 100644 --- a/tests/capabilities/test_reward_memory_outcome_lifecycle.py +++ b/tests/capabilities/test_reward_memory_outcome_lifecycle.py @@ -61,13 +61,14 @@ def _reflection(*, status: str = "eligible") -> str: if status == "no_evidence": return json.dumps( { - "schema_version": "turn_reward_memory_reflection_v0", + "schema_version": "turn_reward_memory_reflection_v1", "status": "no_evidence", } ) + evidence_refs = ["artifact:flow-probe", "receipt:validation"] return json.dumps( { - "schema_version": "turn_reward_memory_reflection_v0", + "schema_version": "turn_reward_memory_reflection_v1", "status": "eligible", "surface_id": "agent_workflow.turn_admission", "outcome_kind": "research", @@ -78,7 +79,53 @@ def _reflection(*, status: str = "eligible") -> str: "Independent validation showed stale snapshots reverse the conclusion." ), "confidence": "high", - "evidence_refs": ["artifact:flow-probe", "receipt:validation"], + "evidence_refs": evidence_refs, + "experience": { + "schema_version": "procedural_experience_contract_v0", + "applicability": [ + "Comparing provider flow snapshots across observation times" + ], + "observed_outcome": ( + "A stale snapshot reversed the comparison relative to the exact " + "timestamped provider read." + ), + "attribution": ( + "Independent validation isolated snapshot freshness as the cause " + "of the reversed conclusion." + ), + "future_behavior": { + "trigger": "A decision compares provider flows across snapshots.", + "action": ( + "Obtain an exact timestamped provider read before comparison." + ), + "validation": ( + "Bind the comparison to the provider read receipt and timestamp." + ), + "stop_condition": ( + "Do not use the comparison when read time or provider receipt is " + "missing." + ), + }, + "limitations": [ + "The lesson does not establish that every older snapshot is wrong." + ], + "evidence_refs": evidence_refs, + }, + } + ) + + +def _legacy_reflection() -> str: + return json.dumps( + { + "schema_version": "turn_reward_memory_reflection_v0", + "status": "eligible", + "surface_id": "agent_workflow.turn_admission", + "outcome_kind": "research", + "content_summary": "A fact-only summary from the legacy contract.", + "reasoning_summary": "The legacy contract has no transferable rule.", + "confidence": "high", + "evidence_refs": ["artifact:legacy-reflection"], } ) @@ -177,6 +224,9 @@ def test_validated_turn_reflection_writes_reads_back_and_replays( assert guidance[0]["content_summary"] == ( "Require an exact timestamped provider read before comparing flows." ) + assert guidance[0]["experience"]["future_behavior"]["action"].startswith( + "Obtain an exact timestamped provider read" + ) assert str(guidance[0]["candidate_ref"]).startswith("candidate:") assert next_turn["outcome_ingest_reconciliation"]["status"] == "empty" @@ -198,7 +248,9 @@ def test_no_evidence_and_explicit_disable_make_zero_provider_calls( goal_id="goal", agent_id="pilot", turn_key="sha256:no-evidence", - host_result={"reward_memory_reflection_json": _reflection(status="no_evidence")}, + host_result={ + "reward_memory_reflection_json": _reflection(status="no_evidence") + }, provider=provider, ) @@ -224,6 +276,115 @@ def test_no_evidence_and_explicit_disable_make_zero_provider_calls( assert provider.retrieve_calls == 0 +def test_legacy_fact_only_reflection_is_audit_only_and_never_calls_provider( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + provider = OutcomeProvider() + config = _config(tmp_path) + monkeypatch.setattr( + outcome_lifecycle, + "resolve_reward_memory_experiment", + lambda **_kwargs: (_status(), config), + ) + + receipt = run_configured_turn_outcome_ingest( + registry_path=tmp_path / "registry.json", + goal_id="goal", + agent_id="pilot", + turn_key="sha256:legacy-reflection", + host_result={"reward_memory_reflection_json": _legacy_reflection()}, + provider=provider, + ) + + assert receipt["status"] == "no_eligible_evidence" + assert receipt["reason_code"] == ( + "legacy_reflection_requires_transferable_experience" + ) + assert receipt["automatic_ingest"] is True + assert provider.sync_calls == 0 + assert provider.retrieve_calls == 0 + + +def test_reconciliation_terminalizes_preupgrade_pending_legacy_reflection( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + provider = OutcomeProvider() + config = _config(tmp_path) + monkeypatch.setattr( + outcome_lifecycle, + "resolve_reward_memory_experiment", + lambda **_kwargs: (_status(), config), + ) + monkeypatch.setattr(outcome_lifecycle, "_goal_repo", lambda *_args: tmp_path) + sidecar_path = turn_outcome_ingest_sidecar_path( + tmp_path, + goal_id="goal", + agent_id="pilot", + source_event_id="legacy-pending", + ) + outcome_lifecycle._write_sidecar( + sidecar_path, + { + "schema_version": "turn_reward_memory_sidecar_v0", + "status": "pending", + "goal_id": "goal", + "agent_id": "pilot", + "turn_key": "sha256:legacy-pending-turn", + "source_event_id": "legacy-pending", + "surface_id": "agent_workflow.turn_admission", + "reflection": json.loads(_legacy_reflection()), + "reflection_validation": {}, + "settlement_validation": {}, + "observed_at": "2026-08-02T10:00:00+00:00", + "attempt_count": 1, + "raw_content_captured": False, + "public_receipt": { + "status": "committed_pending", + "reconciliation_state": "pending", + }, + }, + ) + + first = reconcile_pending_turn_outcome_ingests( + registry_path=tmp_path / "registry.json", + goal_id="goal", + agent_id="pilot", + observed_at="2026-08-03T10:00:00+00:00", + provider=provider, + ) + second = reconcile_pending_turn_outcome_ingests( + registry_path=tmp_path / "registry.json", + goal_id="goal", + agent_id="pilot", + observed_at="2026-08-03T11:00:00+00:00", + provider=provider, + ) + + assert first["pending_count"] == 1 + assert first["attempted_count"] == 1 + assert first["completed_count"] == 0 + assert first["receipts"][0]["reconciliation_state"] == "rejected" + assert first["receipts"][0]["legacy_pending_migration"] == { + "schema_version": "turn_reward_memory_legacy_pending_migration_v0", + "status": "terminal_rejected", + "reason_code": "legacy_reflection_requires_transferable_experience", + "provider_write_state": "unknown_may_have_committed", + "provider_cleanup_performed": False, + "migrated_at": "2026-08-03T10:00:00+00:00", + } + migrated = json.loads(sidecar_path.read_text(encoding="utf-8")) + assert migrated["status"] == "rejected" + assert migrated["legacy_pending_migration"]["provider_write_state"] == ( + "unknown_may_have_committed" + ) + assert second["pending_count"] == 0 + assert second["attempted_count"] == 0 + assert provider.sync_calls == 0 + assert provider.retrieve_calls == 0 + + def test_sibling_agent_scope_is_rejected_before_provider( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, @@ -414,3 +575,65 @@ def test_reflection_validation_without_durable_settlement_is_not_written( assert receipt["candidate_state"] == "awaiting_evidence_validation" assert provider.sync_calls == 0 assert provider.retrieve_calls == 0 + + +def test_reflection_rejects_experience_bound_to_different_evidence( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + config = _config(tmp_path) + provider = OutcomeProvider() + monkeypatch.setattr( + outcome_lifecycle, + "resolve_reward_memory_experiment", + lambda **_kwargs: (_status(), config), + ) + payload = json.loads(_reflection()) + payload["experience"]["evidence_refs"] = ["artifact:different"] + + with pytest.raises(ValueError, match="must match the validated reflection"): + run_configured_turn_outcome_ingest( + registry_path=tmp_path / "registry.json", + goal_id="goal", + agent_id="pilot", + turn_key="sha256:mismatched-evidence", + host_result={"reward_memory_reflection_json": json.dumps(payload)}, + provider=provider, + ) + + assert provider.sync_calls == 0 + assert provider.retrieve_calls == 0 + + +def test_automatic_ingest_uses_the_routed_procedural_class( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + config = _config(tmp_path) + entry = config["corpora"]["agent_turn_preferences"] + entry["corpus"]["class_id"] = "procedural_experience" + entry["standing_policy"]["allowed_target_classes"] = ["procedural_experience"] + provider = OutcomeProvider() + monkeypatch.setattr( + outcome_lifecycle, + "resolve_reward_memory_experiment", + lambda **_kwargs: (_status(), config), + ) + monkeypatch.setattr(outcome_lifecycle, "_goal_repo", lambda *_args: tmp_path) + + receipt = run_configured_turn_outcome_ingest( + registry_path=tmp_path / "registry.json", + goal_id="goal", + agent_id="pilot", + turn_key="sha256:procedural-route", + host_result={"reward_memory_reflection_json": _reflection()}, + settlement_evidence=_settlement_evidence(), + observed_at="2026-08-02T10:00:00+00:00", + provider=provider, + ) + + stored = json.loads(next(iter(provider.resources.values()))) + assert receipt["status"] == "activated" + assert receipt["experience_quality"]["passed"] is True + assert stored["target_class"] == "procedural_experience" + assert stored["experience"]["future_behavior"]["stop_condition"] diff --git a/tests/capabilities/test_value_connector_social_profile.py b/tests/capabilities/test_value_connector_social_profile.py index bebd1adebb..67439a165b 100644 --- a/tests/capabilities/test_value_connector_social_profile.py +++ b/tests/capabilities/test_value_connector_social_profile.py @@ -144,6 +144,20 @@ def test_seed_reuses_scoped_memory_guard_not_an_import_permission() -> None: "revision_ref": "revision:synthetic-verification", "target_class": seed["target_class"], "content_summary": seed["content_summary"], + "experience": { + "schema_version": "procedural_experience_contract_v0", + "applicability": seed["applicability"], + "observed_outcome": seed["observations"][1], + "attribution": seed["observations"][2], + "future_behavior": { + "trigger": seed["applicability"][0], + "action": seed["procedure"][2], + "validation": seed["procedure"][5], + "stop_condition": seed["procedure"][4], + }, + "limitations": seed["limits"], + "evidence_refs": [source_ref], + }, "source": { "source_kind": "reviewed_learning_card", "source_ref": source_ref, @@ -168,6 +182,7 @@ def test_seed_reuses_scoped_memory_guard_not_an_import_permission() -> None: candidate = build_scoped_feedback_reward_memory_candidate(event)["shared_candidate"] assert candidate["status"] == "review_ready" assert candidate["candidate"]["content_summary"] == seed["content_summary"] + assert candidate["guard"]["experience_quality"]["passed"] is True assert candidate["provider_write_performed"] is False assert candidate["external_writes_performed"] is False assert ( diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py index 4d115b95bf..e9d351d505 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -707,7 +707,7 @@ def test_codex_app_refresh_stages_validated_memory_and_spend_finalizes_hook( ) reflection = json.dumps( { - "schema_version": "turn_reward_memory_reflection_v0", + "schema_version": "turn_reward_memory_reflection_v1", "status": "eligible", "surface_id": "agent_workflow.turn_admission", "outcome_kind": "engineering", @@ -715,6 +715,28 @@ def test_codex_app_refresh_stages_validated_memory_and_spend_finalizes_hook( "reasoning_summary": "The declared validator covered the bound outcome.", "confidence": "high", "evidence_refs": ["artifact:app-settlement", "receipt:validator"], + "experience": { + "schema_version": "procedural_experience_contract_v0", + "applicability": ["Retrying a settled managed Turn outcome"], + "observed_outcome": ( + "The exact settlement identity preserved idempotent retry behavior." + ), + "attribution": ( + "The declared validator covered the outcome bound to that identity." + ), + "future_behavior": { + "trigger": "A settled outcome requires retry or reconciliation.", + "action": "Reuse the exact settlement identity on every retry.", + "validation": "Verify the identity against the settlement receipt.", + "stop_condition": ( + "Stop when the receipt is missing or the identity has drifted." + ), + }, + "limitations": [ + "The identity rule does not validate the business outcome itself." + ], + "evidence_refs": ["artifact:app-settlement", "receipt:validator"], + }, }, separators=(",", ":"), ) diff --git a/tests/test_loopx_turn_driver.py b/tests/test_loopx_turn_driver.py index 61e5400335..555baed935 100644 --- a/tests/test_loopx_turn_driver.py +++ b/tests/test_loopx_turn_driver.py @@ -2763,7 +2763,7 @@ def host_runner(request: dict[str, object]) -> dict[str, object]: "summary": "One intermediate fixture step passed validation.", "reward_memory_reflection_json": json.dumps( { - "schema_version": "turn_reward_memory_reflection_v0", + "schema_version": "turn_reward_memory_reflection_v1", "status": "eligible", "surface_id": "agent_workflow.turn_admission", "outcome_kind": "engineering", @@ -2771,6 +2771,28 @@ def host_runner(request: dict[str, object]) -> dict[str, object]: "reasoning_summary": "The independent validator passed.", "confidence": "high", "evidence_refs": ["artifact:fixture-validation"], + "experience": { + "schema_version": "procedural_experience_contract_v0", + "applicability": ["Repeating the validated fixture sequence"], + "observed_outcome": ( + "The independent validator accepted the fixture sequence." + ), + "attribution": ( + "The result is bound to the exact fixture validation artifact." + ), + "future_behavior": { + "trigger": "The same fixture sequence must be repeated.", + "action": "Reuse the verified fixture sequence.", + "validation": "Run the independent fixture validator.", + "stop_condition": ( + "Stop reuse when the fixture revision or validator changes." + ), + }, + "limitations": [ + "The sequence applies only to the validated fixture revision." + ], + "evidence_refs": ["artifact:fixture-validation"], + }, } ), } @@ -3145,7 +3167,7 @@ def test_turn_run_once_codex_cli_wires_validated_reflection_post_settlement( project, runtime, registry = _write_live_fixture(tmp_path) reflection = json.dumps( { - "schema_version": "turn_reward_memory_reflection_v0", + "schema_version": "turn_reward_memory_reflection_v1", "status": "eligible", "surface_id": "agent_workflow.turn_admission", "outcome_kind": "engineering", @@ -3153,6 +3175,26 @@ def test_turn_run_once_codex_cli_wires_validated_reflection_post_settlement( "reasoning_summary": "The independent validator passed.", "confidence": "high", "evidence_refs": ["receipt:codex-cli-validator"], + "experience": { + "schema_version": "procedural_experience_contract_v0", + "applicability": ["Running the validated Codex CLI sequence"], + "observed_outcome": ( + "The independent validator accepted the Codex CLI sequence." + ), + "attribution": "The outcome is bound to the exact validator receipt.", + "future_behavior": { + "trigger": "The same Codex CLI sequence is considered again.", + "action": "Reuse the validated Codex CLI sequence.", + "validation": "Require the exact Codex CLI validator receipt.", + "stop_condition": ( + "Stop reuse when the sequence or validator revision changes." + ), + }, + "limitations": [ + "The result applies only to the validated CLI sequence." + ], + "evidence_refs": ["receipt:codex-cli-validator"], + }, }, separators=(",", ":"), )