Validate recorded Turn versions before journal recovery

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

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")

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};

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:

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 <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. |

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