From c5d9aa51d6bd13637ab5a2bea991946f0a6022e6 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 11 Nov 2026 03:21:19 +0800 Subject: [PATCH] Validate recorded Turn versions before journal recovery Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/reference/protocols/turn-envelope-v0.md | 12 +++ loopx/control_plane/quota/turn_envelope.py | 7 ++ loopx/control_plane/quota/turn_envelope.ts | 35 ++++++-- .../turn_driver/journal_store.py | 16 ++-- .../references/repair-patterns.md | 1 + .../test_complete_turn_journal.py | 79 +++++++++++++++++-- 6 files changed, 125 insertions(+), 25 deletions(-) diff --git a/docs/reference/protocols/turn-envelope-v0.md b/docs/reference/protocols/turn-envelope-v0.md index fabbe72185..fa23f0a0e3 100644 --- a/docs/reference/protocols/turn-envelope-v0.md +++ b/docs/reference/protocols/turn-envelope-v0.md @@ -464,6 +464,18 @@ permitted. Downgrading code must preserve journals and captures; deleting them is not rollback. This transport qualification does not establish live model, packaged App/Lark or overall convergence acceptance. +Recovery previews and journal interpretation use the same typed plan reader. +Supported historical records keep their original versions and bytes. An +unsupported declared plan, envelope or action-signature version or coverage +blocks resume. Journal inspection can report the invalid work contract without +performing recovery or a debit. Retain the historical reader until the last +supported journal and its recovery window are reconciled. + +恢复预览和 journal 解释共用同一个 typed plan reader。受支持的历史记录保留原版本和 +字节。显式声明不支持的 plan、envelope 或动作签名版本及覆盖范围时，恢复会停止。 +检查入口仍可报告契约无效，不执行恢复或扣额。保留历史读取器，直到最后一份受支持的 +journal 及其恢复窗口完成核对。 + Repair and replan diagnostics remain in the journal's typed `host_result`, including its summary and follow-up action. They do not implicitly replace Todo notes or evidence, even when settlement succeeds. Semantic refusal and cached diff --git a/loopx/control_plane/quota/turn_envelope.py b/loopx/control_plane/quota/turn_envelope.py index 6f7bfa82df..eadd76862c 100644 --- a/loopx/control_plane/quota/turn_envelope.py +++ b/loopx/control_plane/quota/turn_envelope.py @@ -226,6 +226,13 @@ def read_turn_plan_work(plan: Mapping[str, Any]) -> dict[str, Any]: return _wire_mapping(plan.get("turn_envelope")) +def read_recorded_turn_plan_lineage(plan: Mapping[str, Any]) -> dict[str, Any]: + """Check a recorded plan with the same source reader as journal inspection.""" + tokens = source_number_tokens(plan.get("work_contract")) + return _turn_envelope_result("recorded_plan_lineage", plan=plan, + **({"source_number_tokens": tokens} if tokens else {})) + + def turn_host_lineage(request: Mapping[str, Any]) -> dict[str, str]: """Read identity from the complete owner, with historical request decoding.""" checked = complete_turn_host_source(turn_plan_host_source(request), operation="complete_host_lineage") diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts index bbc341d374..5aaabd90a6 100644 --- a/loopx/control_plane/quota/turn_envelope.ts +++ b/loopx/control_plane/quota/turn_envelope.ts @@ -1578,9 +1578,10 @@ function signatureCoverage(envelope: JsonObject, responsePlanValue: unknown): st * with its existing owner. Explicit unknown/null declarations are never v0. * This does not make the bounded envelope a complete future work contract. */ -export function assertTurnEnvelopeConsumerVersion(envelope: JsonObject): void { +export function assertTurnEnvelopeConsumerVersion(envelope: JsonObject, allowUnversionedHistory = false): void { assertSupportedQuotaPacket(envelope); - if (envelope.schema_version !== TURN_ENVELOPE_SCHEMA_VERSION) { + if (envelope.schema_version !== TURN_ENVELOPE_SCHEMA_VERSION && + !(allowUnversionedHistory && !Object.hasOwn(envelope, "schema_version"))) { throw new EffectRuntimeRequestError("unsupported TurnEnvelope schema_version; use a compatible consumer"); } const signature = object(envelope.action_signature); @@ -1804,8 +1805,8 @@ export function completeHostSource(request: JsonObject): {decision: JsonObject; /** A current private plan stores its one admitted source and transport header. * The old plan codec is read only for recorded journals or explicit old input. */ -export function turnPlanWorkSource(plan: JsonObject, numberTokens?: unknown): { - work: JsonObject; decision: JsonObject | null; scopedIdentity: JsonObject | null | undefined; +function checkedTurnPlanSource(plan: JsonObject, numberTokens?: unknown): { + work: JsonObject; decision: JsonObject | null; } { if (plan.schema_version === "loopx_turn_plan_v1") { if (!Object.hasOwn(plan, "work_contract") || Object.hasOwn(plan, "turn_envelope") @@ -1817,8 +1818,7 @@ export function turnPlanWorkSource(plan: JsonObject, numberTokens?: unknown): { turn_key: object(plan.transaction).turn_key, source_decision_hash: plan.source_decision_hash, ...(Object.hasOwn(plan, "detail_ref") ? {detail_ref: plan.detail_ref} : {})}}); const work = {goal_id: checked.decision.goal_id ?? null, ...checked.projection}; - return {work, decision: checked.decision, scopedIdentity: scopedTurnSettlementIdentity(work, - object(plan.transaction).turn_instance_id)}; + return {work, decision: checked.decision}; } if (Object.hasOwn(plan, "schema_version") && plan.schema_version !== "loopx_turn_plan_v0") { throw new EffectRuntimeRequestError("unsupported Turn plan version"); @@ -1827,9 +1827,22 @@ export function turnPlanWorkSource(plan: JsonObject, numberTokens?: unknown): { if (Object.hasOwn(plan, "work_contract")) { const checked = completeHostSource({payload: plan.work_contract, envelope, source_number_tokens: numberTokens}); - return {work: {...envelope, ...checked.projection}, decision: checked.decision, scopedIdentity: undefined}; + return {work: {...envelope, ...checked.projection}, decision: checked.decision}; } - return {work: envelope, decision: null, scopedIdentity: undefined}; + // Older journals can omit the envelope header. A declared version or + // signature format must pass the same compatibility check as Host input. + assertTurnEnvelopeConsumerVersion(envelope, true); + return {work: envelope, decision: null}; +} + +export function turnPlanWorkSource(plan: JsonObject, numberTokens?: unknown): { + work: JsonObject; decision: JsonObject | null; scopedIdentity: JsonObject | null | undefined; +} { + const source = checkedTurnPlanSource(plan, numberTokens); + const scopedIdentity = plan.schema_version === "loopx_turn_plan_v1" + ? scopedTurnSettlementIdentity(source.work, object(plan.transaction).turn_instance_id) + : undefined; + return {...source, scopedIdentity}; } function admittedDriverLineage(work: JsonObject, decision: JsonObject | null, @@ -2403,6 +2416,12 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject { const request = requireJsonObject(value, "turn envelope request"); const operation = request.operation; if (operation === "planning_facts") return completePlanningFacts(request); + if (operation === "recorded_plan_lineage") { + const {work, decision} = checkedTurnPlanSource( + requireJsonObject(request.plan, "recorded Turn plan"), request.source_number_tokens); + const identity = decision === null ? work : completeSessionLineage(decision); + return {goal_id: identity.goal_id ?? null, agent_id: identity.agent_id ?? null}; + } if (operation === "plan_work") { const {decision, projection} = completeHostSource(request); return {goal_id: decision.goal_id ?? null, ...projection}; diff --git a/loopx/control_plane/turn_driver/journal_store.py b/loopx/control_plane/turn_driver/journal_store.py index aae4bcfcde..7f6ded60c0 100644 --- a/loopx/control_plane/turn_driver/journal_store.py +++ b/loopx/control_plane/turn_driver/journal_store.py @@ -14,7 +14,7 @@ runtime_artifact_lock_target, shadow_maintenance_lock_target, ) -from ..effect_runtime import EffectRuntimeConflict, effect_runtime_result +from ..effect_runtime import EffectRuntimeConflict, EffectRuntimeRejected, effect_runtime_result from .turn_journal_runtime import ( write_turn_journal, ) @@ -129,14 +129,12 @@ def load_loopx_turn_plan_from_journal( ) if transaction.get("turn_key") != turn_key or journal.get("turn_key") != turn_key: raise ValueError("LoopX Turn resume journal has mismatched turn lineage") - from ..quota.turn_envelope import COMPLETE_TURN_PLAN_SCHEMA_VERSION, turn_host_lineage - - if plan.get("schema_version") == COMPLETE_TURN_PLAN_SCHEMA_VERSION or "work_contract" in plan: - # Recorded transitional plans must check their supplied complete source, - # just like current plans. Only source-absent history uses raw identity. - identity = turn_host_lineage(plan) - else: - identity = cast(dict[str, Any], plan.get("turn_envelope")) if isinstance(plan.get("turn_envelope"), dict) else {} + from ..quota.turn_envelope import read_recorded_turn_plan_lineage + + try: + identity = read_recorded_turn_plan_lineage(plan) + except EffectRuntimeRejected as exc: + raise ValueError(str(exc)) from exc if identity.get("goal_id") != goal_id or journal.get("goal_id") != goal_id: raise ValueError("LoopX Turn resume journal belongs to another goal") if agent_id is not None and identity.get("agent_id") != agent_id: diff --git a/skills/loopx-self-repair/references/repair-patterns.md b/skills/loopx-self-repair/references/repair-patterns.md index 9d1389fdec..2a21221211 100644 --- a/skills/loopx-self-repair/references/repair-patterns.md +++ b/skills/loopx-self-repair/references/repair-patterns.md @@ -5,6 +5,7 @@ teaches a reusable control-plane lesson. | Pattern | Symptoms | Evidence To Read | Likely Root | Durable Repair | | --- | --- | --- | --- | --- | +| `recorded_turn_version_drift` | An unsupported historical plan, envelope or action signature appears consistent or resumes as a committed replay. | Compare preview, inspection and explicit resume using a journal from the old writer. Change one version declaration at a time and verify capture bytes, Host invocation and quota receipts. | A historical reader extracts identity or replays the receipt without applying the version check used by the Host. | Use the typed plan source reader for resume and journal interpretation. Preserve valid historical records, and refuse unsupported declarations before recovery. Prove refusal and original replay through File and SQLite. | | `complete_source_retry_callback_drift` | A valid complete-source Host request reaches normal execution, but its failed Session retry gives the binding callback an empty envelope and rejects the original actor. | Declared current and historical request versions, a provider callback that reads its argument, real File/SQLite sources, saved journal/capture bytes, retry actions and settlement observations. | The writer removed the current envelope while a recovery reader still selects that field; callbacks that ignore their argument hide the migration gap. | Validate supplied complete material with the existing source owner, then deliver the callback body by the recorded wire version. Current requests use the complete decision; v0 keeps its original envelope even with a complete sidecar. Refuse malformed/mixed current input before provider IO. Test original-Session resume, actor/work drift refusal, immutable source/journal recovery and once-only results/effects. Recorded callback counts do not establish real model adoption or canonical settlement. | | `marked_todo_archive_pointer_reader_drift` | An installed `quota should-run` reports `quota_unexpected_collection_error`; read-only `loopx --format json quota status --goal-id --verbose` confirms `ValueError: non-generated content inside marked Todo region`, while current source reads succeed. Thin Codex App output may expose only the error code. | Use `python3 skills/loopx-self-repair/scripts/find_pattern.py --query 'quota_unexpected_collection_error non-generated content inside marked Todo region'`, then read this exact pattern. Inspect the launcher, Python/import route, component hashes, exact emitted line and installed `machine_region.py`; compare synthetic File/SQLite source/wheel fixtures. Keep verbose diagnostics and snapshots private. | The observed reader matching `8d314af49` rejects an emitted archive-summary comment without recognized Todo metadata. It already accepts the emitted per-Todo active-field omission row through valid `todo_id` metadata. Their shared `loopx:todo-archive-pointer` prefix does not establish the cause; attribute an omission-row failure only when the exact installed reader actually rejects that exact generated line. Version or declared commit alone cannot establish dirty installed bytes. | Verify projection ownership and compare archive-summary, per-Todo omission and foreign-narrative controls under the same interpreter and real backend; preserve foreign-content rejection, original-Turn recovery/replay, unchanged canonical work and zero quota debit. If main already includes the archive-summary reader repair, as in #7834, do not duplicate it, strip active metadata or relax the grammar. Qualify the affected source/wheel composition, then follow existing upgrade authority and original-binding readback. Source recovery alone does not qualify the installed runtime or prove long-horizon benefit. | | `outcome_review_same_batch_link_invalidation` | A successful review-plus-result refresh immediately reports its Agent review basis stale; an unchanged standalone review is then repeated. | Exact refinement and result-delivery receipts, original CAS revision, retained conclusions and canonical Todo links before/after; distinguish the authored effect from other new evidence. | Refinement hashes pre-link work, then post-writeback delivery adds a work-basis input. | Validate the complete intent and same-Goal ownership before determining relevance in the typed owner. Bind only the newly authored complete review whose outcome includes that work to the exact normalized same-batch additive link using the shared append rule and original CAS snapshot. Keep it stale until actual commit; a valid separate-work result leaves the reviewed outcome unchanged. Preserve independent verification, all retained requirements, separate/legacy reviews, initial CAS refusal, failed delivery, exact receipt replay and concurrent work/evidence invalidation through real File/SQLite CLI. Never exclude all evidence links, refresh old review digests against an arbitrary later head, reorder effects before primary admission or claim model/score benefit from this repair. | diff --git a/tests/control_plane/test_complete_turn_journal.py b/tests/control_plane/test_complete_turn_journal.py index bce1041340..5d2d675ed9 100644 --- a/tests/control_plane/test_complete_turn_journal.py +++ b/tests/control_plane/test_complete_turn_journal.py @@ -1,6 +1,7 @@ """Complete-source integrity must survive every journal read/write boundary.""" from copy import deepcopy import json +import sys import pytest @@ -132,7 +133,9 @@ def test_complete_journal_readers_and_writer_refuse_source_drift(complete_journa def test_complete_journal_keeps_numeric_source_and_historical_decoder(complete_journal): - from loopx.control_plane.quota.turn_envelope import compose_turn_decision + from loopx.control_plane.quota.turn_envelope import build_turn_envelope, compose_turn_decision + from loopx.control_plane.turn_driver.decision_capture import prepare_turn_decision_capture + from loopx.control_plane.turn_driver.driver import build_loopx_turn_plan project, runtime, registry, journal = complete_journal plan = journal["plan"] @@ -146,6 +149,8 @@ def test_complete_journal_keeps_numeric_source_and_historical_decoder(complete_j identity = plan["transaction"]["settlement_plan"]["identity"] path = turn_journal_path(runtime, goal_id=GOAL_ID, turn_key=key) path.parent.mkdir(parents=True, exist_ok=True) + journal["plan"] = plan = prepare_turn_decision_capture(plan, journal_path=path, + decision=contract, existing_journal=False) written = write_turn_journal(str(path), journal, expected_effect_id=identity["effect_id"]) assert written["appended"] is True assert json.loads(path.read_text())["plan"]["work_contract"]["future_measurements"] == [1.0, -0.0, 2**60 + 1] @@ -174,14 +179,26 @@ def test_complete_journal_keeps_numeric_source_and_historical_decoder(complete_j with pytest.raises(ValueError, match="complete Turn plan requires one source"): load_loopx_turn_plan_from_journal(runtime, goal_id=GOAL_ID, agent_id=AGENT_ID, turn_key=key) - historical["plan"] = _journal_plan(contract, legacy=True, goal_ref=plan.get("goal_ref")) - historical["plan"].pop("work_contract") - path.write_text(json.dumps(historical)) - original_bytes = path.read_bytes() + path.write_bytes(before) + + # A separate, genuine raw-source capture exercises the old decoder. Keep + # the original complete observation bound to its own Turn. + historical_plan = build_loopx_turn_plan(build_turn_envelope(source), + host="generic-cli", execution_mode="isolated-headless", turn_instance_id="historical-journal-source", + goal_ref=plan.get("goal_ref")) + historical_key = historical_plan["transaction"]["turn_key"] + historical_path = turn_journal_path(runtime, goal_id=GOAL_ID, turn_key=historical_key) + historical["turn_key"] = historical_key + historical["plan"] = prepare_turn_decision_capture(historical_plan, journal_path=historical_path, + decision=source, existing_journal=False) + assert "work_contract" not in historical["plan"] + historical_path.write_text(json.dumps(historical)) + original_bytes = historical_path.read_bytes() assert load_loopx_turn_plan_from_journal(runtime, goal_id=GOAL_ID, - agent_id=AGENT_ID, turn_key=key) == historical["plan"] - assert _inspect(project, runtime, registry, key)["journal_consistent"] is True - assert path.read_bytes() == original_bytes + agent_id=AGENT_ID, turn_key=historical_key) == historical["plan"] + assert _inspect(project, runtime, registry, historical_key)["journal_consistent"] is True + assert historical_path.read_bytes() == original_bytes + assert path.read_bytes() == before def test_cli_resume_refuses_changed_complete_source_without_rewriting_history(complete_journal): @@ -203,6 +220,52 @@ def test_cli_resume_refuses_changed_complete_source_without_rewriting_history(co assert path.read_bytes() == original_bytes +@pytest.mark.parametrize("complete_journal", [("file", True), ("sqlite", True)], indirect=True) +def test_legacy_journal_refuses_unknown_recorded_versions(complete_journal): + from loopx.control_plane.turn_driver.transaction import TRANSACTION_PHASES + + project, runtime, registry, journal = complete_journal + journal["plan"].pop("work_contract") + journal.update(status="committed", completed_phases=list(TRANSACTION_PHASES)) + key = journal["turn_key"] + path = turn_journal_path(runtime, goal_id=GOAL_ID, turn_key=key) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(journal)) + resume = ("turn", "run-once", "--resume-turn-key", key, + "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--host", "generic-cli", + "--execution-mode", "isolated-headless", "--project", str(project), + "--host-command-json", json.dumps([sys.executable, "-c", "raise SystemExit(99)"])) + + code, result = _run_cli(registry, runtime, *resume, cwd=project) + assert code == 0 and result["status"] == "preview", result + assert _inspect(project, runtime, registry, key)["journal_consistent"] is True + + for fields in ( + ("schema_version",), + ("turn_envelope", "schema_version"), + ("turn_envelope", "action_signature", "schema_version"), + ("turn_envelope", "action_signature", "coverage"), + ): + invalid = deepcopy(journal) + carrier = invalid["plan"] + for field in fields[:-1]: + carrier = carrier[field] + carrier[fields[-1]] = "future_unsupported" + path.write_text(json.dumps(invalid)) + saved = path.read_bytes() + + observed = _inspect(project, runtime, registry, key) + assert observed["journal_consistent"] is False, fields + assert "work_contract_invalid" in observed["violations"] + assert observed["recovery_decision"]["action"] == "blocked" + assert observed["effects"] == [] + for flags in ((), ("--execute",)): + code, rejected = _run_cli(registry, runtime, *resume, *flags, cwd=project) + assert code == 1 and rejected["ok"] is False, (fields, rejected) + assert "unsupported Turn" in rejected["error"], rejected + assert path.read_bytes() == saved + + def test_managed_retry_reads_complete_source_with_original_numbers(complete_journal): from loopx.control_plane.quota.turn_envelope import build_turn_envelope, compose_turn_decision from loopx.control_plane.turn_driver.host_failure import build_host_failure_record