fix(turn): preserve managed host material replan packets

docs/reference/protocols/loopx-turn-v0.md

@@ -657,6 +657,32 @@ Every attempted tick returns one result kind:
 | `validation_failed` | Host output exists but task validation failed or is inconclusive. | Preserve failure evidence and route to repair/replan. |
 | `writeback_failed` | Validated work could not be durably recorded. | Do not spend; retry idempotent writeback before more delivery. |
 
+Managed host adapters carry the existing path-delta fields through the typed
+result unchanged: `path_delta_mode`, `agent_vision_json`, and
+`vision_unchanged_reason`. A material path change uses
+`path_delta_mode: "material_replan"`, `result_kind: "replan_required"`, and a
+JSON-encoded `goal_vision_replan_contract_v0` packet whose
+`goal_path_delta_v0.outcome` is `replan`; it must not also claim an unchanged
+reason. An unchanged path uses `path_delta_mode: "unchanged"`, no vision
+packet, and a bounded `vision_unchanged_reason`. The Turn executor remains the
+owner of packet parsing, bounds, and semantic validation. These declarations do
+not grant write or Goal authority, and legacy hosts that omit both path-delta
+fields retain their prior compatibility behavior. For material results, an
+explicit null or other non-string value for either path-delta field is invalid;
+omitting the field remains distinct from supplying null. Stop results
+(`wait`, `user_action_required`, and `iteration_failed`) do not declare a path
+delta. Managed dsh and optional-cloud prompts may omit both fields; the Codex
+CLI output schema requires both strings, so empty strings express the same
+no-declaration result. The executor accepts those empty strings and rejects
+non-empty stop-result declarations. The shared material validation applies to
+generic CLI, Codex CLI, dsh, and optional-cloud host results.
+
+The existing TypeScript Vision normalizer owns packet interpretation and emits
+the canonical `goal_vision_replan_contract_v0` shape. Its current compatibility
+behavior does not verify the submitted packet's top-level `schema_version`;
+the nested `path_delta.schema_version` remains optional, and when present the
+existing owner requires `goal_path_delta_v0`.
+
 ### Settlement identity decoding / 结算身份解码
 
 Executable settlement decodes exactly one Todo or autonomous-replan binding in

loopx/control_plane/turn_driver/executor.py

@@ -240,10 +240,20 @@ def _normalize_host_path_delta(
     unchanged_reason: str,
     errors: list[str],
 ) -> tuple[str, dict[str, Any] | None]:
-    path_delta_mode = str(result.get("path_delta_mode") or "").strip()
+    raw_path_delta_mode = result.get("path_delta_mode")
+    path_delta_mode = ""
+    if raw_path_delta_mode is None and "path_delta_mode" in result:
+        errors.append("path_delta_mode must be a string")
+    elif raw_path_delta_mode is not None and not isinstance(raw_path_delta_mode, str):
+        path_delta_mode = ""
+        errors.append("path_delta_mode must be a string")
+    else:
+        path_delta_mode = str(raw_path_delta_mode or "").strip()
     raw_agent_vision = result.get("agent_vision_json")
     if raw_agent_vision is None:
         agent_vision_json = ""
+        if "agent_vision_json" in result:
+            errors.append("agent_vision_json must be a JSON string")
     elif isinstance(raw_agent_vision, str):
         agent_vision_json = raw_agent_vision.strip()
     else:
@@ -279,7 +289,7 @@ def _normalize_host_path_delta(
                     goal_id=str(envelope.get("goal_id") or ""),
                     agent_id=str(envelope.get("agent_id") or "") or None,
                 )
-            except (json.JSONDecodeError, ValueError) as exc:
+            except (json.JSONDecodeError, TypeError, ValueError) as exc:
                 errors.append(f"invalid agent_vision_json: {exc}")
 
     if path_delta_mode == "material_replan":
@@ -420,7 +430,7 @@ def validate_loopx_turn_host_result(
         or str(result.get("agent_vision_json") or "").strip()
     ):
         errors.append(
-            "wait and user_action_required results cannot declare a path delta"
+            "non-material host results cannot declare path_delta_mode or agent_vision_json"
         )
     if subagent.subagent_execution_topology(plan) is not None:
         subagent.observe_subagent_host_result(plan, result, normalized, errors)

loopx/control_plane/turn_driver/host_candidate.py

@@ -147,7 +147,19 @@ def render_prompt(authority: Mapping[str, Any]) -> str:
         "- summary: what changed or why stopped (<=400 chars)\n"
         "- recommended_action: the bounded follow-up recommendation (<=1200 chars)\n"
         "- next_action: the concrete next step (<=1200 chars)\n"
-        "- vision_unchanged_reason: why the goal path is unchanged (<=240 chars)\n"
+        "- vision_unchanged_reason: why a material result's goal path is unchanged "
+        "(<=240 chars)\n"
+        "For material result kinds (validated_progress, repair_required, and "
+        "replan_required), describe the path outcome. For a material goal-path "
+        "change, set path_delta_mode to material_replan, "
+        "result_kind to replan_required, and agent_vision_json to a JSON-encoded "
+        "bounded goal_vision_replan_contract_v0 packet whose path_delta is "
+        "goal_path_delta_v0 with outcome replan; leave vision_unchanged_reason "
+        "empty. For unchanged paths, set path_delta_mode to unchanged, omit "
+        "agent_vision_json, and provide vision_unchanged_reason. These fields "
+        "describe the result only; they grant no authority. For stop results "
+        "(wait, user_action_required, iteration_failed), omit both "
+        "path_delta_mode and agent_vision_json.\n"
         "Use repair_required when the task is sound but a recoverable defect "
         "blocks it, replan_required when this route is exhausted, and "
         "wait/user_action_required when no material write is safe, and "
@@ -280,15 +292,24 @@ def build_result(
 
     if kind in MATERIAL_KINDS:
         _complete_material_fields(result, kind)
-    # This adapter has no goal-vision packet, so the executor treats the path
-    # delta as unchanged and requires a bounded reason for material results.
-    result["vision_unchanged_reason"] = _bounded(
-        candidate.get("vision_unchanged_reason")
-        or (
+    for field in ("path_delta_mode", "agent_vision_json"):
+        if field in candidate:
+            # Preserve the structured executor input verbatim. The executor
+            # owns type, size, JSON and semantic validation.
+            result[field] = candidate[field]
+
+    has_path_delta = "path_delta_mode" in candidate or "agent_vision_json" in candidate
+    unchanged_reason = candidate.get("vision_unchanged_reason")
+    if unchanged_reason:
+        result["vision_unchanged_reason"] = _bounded(
+            unchanged_reason, limit=TEXT_LIMITS["vision_unchanged_reason"]
+        )
+    elif not has_path_delta:
+        # Keep the legacy material-without-packet behavior for older hosts.
+        result["vision_unchanged_reason"] = _bounded(
             "host reported material work without a goal vision replan packet"
             if kind in MATERIAL_KINDS
-            else "host reported no material change"
-        ),
-        limit=TEXT_LIMITS["vision_unchanged_reason"],
-    )
+            else "host reported no material change",
+            limit=TEXT_LIMITS["vision_unchanged_reason"],
+        )
     return result

tests/test_dsh_goal_mode.py

@@ -16,6 +16,7 @@
     turn_envelope_action_signature_document,
 )
 from loopx.control_plane.turn_driver.executor import validate_loopx_turn_host_result
+from loopx.control_plane.turn_driver.codex_cli import codex_cli_result_schema
 from loopx.control_plane.turn_driver.host_failure import (
     BuiltInHostError,
     build_host_failure_record,
@@ -362,6 +363,9 @@ def test_prompt_requests_one_typed_public_safe_json_result() -> None:
     # Boundary discipline stays in the prompt text.
     assert "write_scope" in prompt
     assert "credentials" in prompt
+    assert "material_replan" in prompt and "goal_path_delta_v0" in prompt
+    assert "These fields describe the result only; they grant no authority." in prompt
+    assert "For stop results (wait, user_action_required, iteration_failed), omit both path_delta_mode and agent_vision_json." in prompt
 
 
 def test_prompt_carries_the_signed_authority_without_prose_reconstruction() -> None:
@@ -448,6 +452,135 @@ def test_build_result_shapes_material_results_with_required_fields() -> None:
     # Sparse material blocks get bounded fallbacks, never empty authority text.
     assert result["recommended_action"]
     assert result["vision_unchanged_reason"]
+    assert "path_delta_mode" not in result
+    assert "agent_vision_json" not in result
+
+
+def _material_replan_candidate() -> tuple[dict[str, object], str]:
+    vision = {
+        "schema_version": "goal_vision_replan_contract_v0",
+        "state": "active",
+        "vision_patch": {
+            "vision_summary": "Continue the next bounded fixture check.",
+            "acceptance_summary": "Validate the next fixture outcome independently.",
+        },
+        "path_delta": {
+            "schema_version": "goal_path_delta_v0",
+            "outcome": "replan",
+            "prior_assumption": "The prior fixture stage remained open. " + "a" * 250,
+            "observed_reality": "The prior stage passed its local validation. " + "b" * 250,
+            "evidence_refs": [f"fixture:{index}-" + "e" * 120 for index in range(4)],
+            "changed": [f"Advance bounded fixture check {index}-" + "c" * 70 for index in range(3)],
+        },
+    }
+    vision_json = json.dumps(vision, indent=2)
+    candidate: dict[str, object] = {
+        "result_kind": "replan_required",
+        "classification": "fixture replan",
+        "recommended_action": "Continue the validated successor stage.",
+        "next_action": "Validate the next bounded fixture check.",
+        "path_delta_mode": "material_replan",
+        "agent_vision_json": vision_json,
+    }
+    return candidate, vision_json
+
+
+def test_build_result_preserves_material_replan_for_the_executor() -> None:
+    candidate, vision_json = _material_replan_candidate()
+    result = turn_host_adapter.build_result(_signed_request(), candidate)
+
+    assert result["path_delta_mode"] == "material_replan"
+    assert result["agent_vision_json"] == vision_json
+    assert len(vision_json) > 400
+    assert "vision_unchanged_reason" not in result
+    verdict = validate_loopx_turn_host_result(
+        _turn_plan(_signed_request()), result
+    )
+    assert verdict["ok"], verdict["errors"]
+
+
+@pytest.mark.parametrize(
+    ("change", "expected_error"),
+    [
+        ({"agent_vision_json": ["not a string"]}, "agent_vision_json must be a JSON string"),
+        ({"agent_vision_json": None}, "agent_vision_json must be a JSON string"),
+        ({"agent_vision_json": "{"}, "invalid agent_vision_json"),
+        ({"agent_vision_json": "[]"}, "agent_vision_json must decode to a JSON object"),
+        ({"agent_vision_json": "x" * 3_201}, "agent_vision_json exceeds 3200 characters"),
+        ({"path_delta_mode": "unexpected"}, "path_delta_mode must be unchanged or material_replan"),
+        ({"path_delta_mode": False}, "path_delta_mode must be a string"),
+        ({"path_delta_mode": None}, "path_delta_mode must be a string"),
+        ({"result_kind": "validated_progress"}, "material_replan path_delta_mode requires result_kind replan_required"),
+        ({"vision_unchanged_reason": "The goal path is unchanged."}, "material_replan cannot also declare vision_unchanged_reason"),
+    ],
+    ids=["wrong-json-type", "null-json", "malformed-json", "nonobject-json", "oversized-json", "wrong-mode", "wrong-mode-type", "null-mode", "wrong-kind", "conflicting-reason"],
+)
+def test_build_result_preserves_invalid_replan_for_executor_rejection(
+    change: dict[str, object], expected_error: str,
+) -> None:
+    candidate, _vision_json = _material_replan_candidate()
+    candidate.update(change)
+    result = turn_host_adapter.build_result(_signed_request(), candidate)
+    for field in ("agent_vision_json", "path_delta_mode"):
+        if field in change:
+            assert result[field] == change[field]
+
+    verdict = validate_loopx_turn_host_result(_turn_plan(_signed_request()), result)
+    assert not verdict["ok"]
+    assert expected_error in " ".join(verdict["errors"])
+
+
+@pytest.mark.parametrize("kind", ["wait", "user_action_required", "iteration_failed"])
+def test_stop_candidates_omit_path_delta_fields_and_pass_executor(kind: str) -> None:
+    candidate = {
+        "result_kind": kind,
+        "classification": "bounded stop",
+        "next_action": "Await the next authorized iteration.",
+    }
+    result = turn_host_adapter.build_result(_signed_request(), candidate)
+
+    assert "path_delta_mode" not in result
+    assert "agent_vision_json" not in result
+    verdict = validate_loopx_turn_host_result(_turn_plan(_signed_request()), result)
+    assert verdict["ok"], verdict["errors"]
+
+
+@pytest.mark.parametrize("kind", ["wait", "user_action_required", "iteration_failed"])
+def test_stop_candidates_cannot_smuggle_material_path_fields(kind: str) -> None:
+    candidate, vision_json = _material_replan_candidate()
+    candidate["result_kind"] = kind
+    result = turn_host_adapter.build_result(_signed_request(), candidate)
+
+    assert result["path_delta_mode"] == "material_replan"
+    assert result["agent_vision_json"] == vision_json
+    verdict = validate_loopx_turn_host_result(_turn_plan(_signed_request()), result)
+    assert not verdict["ok"]
+    assert "non-material host results cannot declare path_delta_mode or agent_vision_json" in " ".join(verdict["errors"])
+
+
+@pytest.mark.parametrize("kind", ["wait", "user_action_required", "iteration_failed"])
+def test_codex_schema_stop_with_required_empty_path_fields_passes_executor(kind: str) -> None:
+    request = _signed_request()
+    schema = codex_cli_result_schema(request)
+    required = schema["required"]
+    properties = schema["properties"]
+    assert "path_delta_mode" in required and "agent_vision_json" in required
+    assert properties["path_delta_mode"]["type"] == "string"
+    assert properties["agent_vision_json"]["type"] == "string"
+    value = {field: "" for field in required}
+    value.update(
+        {
+            "schema_version": properties["schema_version"]["enum"][0],
+            "turn_key": request["turn_key"],
+            "result_kind": kind,
+            "completed_phases": ["host_execute", "typed_result"],
+            "classification": "bounded stop",
+            "next_action": "Await the next authorized iteration.",
+        }
+    )
+
+    verdict = validate_loopx_turn_host_result(_turn_plan(request), value)
+    assert verdict["ok"], verdict["errors"]
 
 
 def _turn_plan(request: dict) -> dict:

tests/test_loopx_turn_driver.py

@@ -21,6 +21,7 @@
 from loopx.control_plane.coordination.runtime_shadow import (
     build_todo_runtime_shadow_projection,
 )
+from loopx.control_plane.todos.active_state_todo_parser import parse_active_state_todos
 from loopx.control_plane.quota.turn_envelope import build_turn_envelope
 from loopx.control_plane.turn_driver import (
     LOOPX_TURN_SESSION_BINDING_SCHEMA_VERSION,
@@ -2280,6 +2281,98 @@ def run(turn_instance_id: str, context: str) -> dict[str, object]:
     assert state_path.read_text(encoding="utf-8") == before_state
 
 
+@pytest.mark.parametrize("malformed_packet", [False, True], ids=["material-replan", "invalid-json-array"])
+def test_turn_run_once_cli_dsh_preserves_material_path_delta(
+    tmp_path: Path, malformed_packet: bool,
+) -> None:
+    project, runtime, registry = _write_live_fixture(
+        tmp_path, todo_metadata_extra="no_followup=true"
+    )
+    state_path = project / ".codex/goals/loopx-turn-fixture/ACTIVE_GOAL_STATE.md"
+    parsed = parse_active_state_todos(state_path.read_text(encoding="utf-8"), item_limit=None)
+    projection = build_todo_runtime_shadow_projection(
+        goal_id="loopx-turn-fixture", handoff_mode="soft_claim",
+        todos=parsed["agent_todos"]["items"],
+    )
+    initialize_canonical_authority(
+        runtime, "loopx-turn-fixture", projection, state_path=state_path
+    )
+    host_project = project / "isolated-dsh-workspace"
+    host_project.mkdir()
+    vision = {
+        "schema_version": "goal_vision_replan_contract_v0",
+        "state": "active",
+        "vision_patch": {
+            "vision_summary": "Continue the next bounded fixture check.",
+            "acceptance_summary": "Validate the next fixture outcome independently.",
+        },
+        "path_delta": {
+            "schema_version": "goal_path_delta_v0",
+            "outcome": "replan",
+            "prior_assumption": "The prior fixture stage remained open.",
+            "observed_reality": "The prior stage passed its local validation.",
+            "evidence_refs": ["fixture:validated-closed-stage"],
+            "changed": ["Advance the next bounded fixture check."],
+        },
+    }
+    candidate = {
+        "result_kind": "replan_required",
+        "classification": "fixture_replan_successor",
+        "summary": "The prior stage closed and an active successor was authored.",
+        "recommended_action": "Continue the validated successor stage.",
+        "next_action": "Validate the next bounded fixture check.",
+        "path_delta_mode": "material_replan",
+        "agent_vision_json": "[]" if malformed_packet else json.dumps(vision),
+    }
+    runner = tmp_path / "material_replan_dsh_runner.py"
+    runner.write_text(
+        "import json\nfrom pathlib import Path\n"
+        f"candidate = {candidate!r}\n"
+        "def run_dsh_turn(**kwargs):\n"
+        "    counter = Path(kwargs['workspace']) / 'host-count.txt'\n"
+        "    counter.write_text(str(int(counter.read_text()) + 1 if counter.exists() else 1))\n"
+        "    Path(kwargs['workspace'], 'validated-artifact.txt').write_text('validated')\n"
+        "    return json.dumps(candidate)\n",
+        encoding="utf-8",
+    )
+    validator = (
+        "import pathlib; raise SystemExit(0 if "
+        "pathlib.Path('validated-artifact.txt').read_text() == 'validated' else 7)"
+    )
+    args = [
+        "--registry", str(registry), "--runtime-root", str(runtime),
+        "--format", "json", "turn", "run-once", "--goal-id",
+        "loopx-turn-fixture", "--agent-id", "codex-fixture", "--host", "dsh",
+        "--project", str(host_project), "--dsh-runner", str(runner),
+        "--validation-command-json", json.dumps([sys.executable, "-c", validator]),
+        "--execution-mode", "isolated-headless", "--scan-root", str(project),
+        "--no-global-sync", "--execute",
+    ]
+    output = io.StringIO()
+    with contextlib.redirect_stdout(output):
+        exit_code = cli_main(args)
+    payload = json.loads(output.getvalue())
+    assert (host_project / "host-count.txt").read_text(encoding="utf-8") == "1"
+    if malformed_packet:
+        assert exit_code == 1, payload
+        assert payload["effects"]["host_invoked"] is True
+        assert payload["effects"]["state_written"] is False
+        assert payload["effects"]["quota_spent"] is False
+        assert "agent_vision_json" in json.dumps(payload)
+        assert state_path.read_text(encoding="utf-8").find("successor") == -1
+    else:
+        assert exit_code == 0, payload
+        assert payload["status"] == "committed"
+        assert payload["result_kind"] == "replan_required"
+        assert payload["effects"]["state_written"] is True
+        assert payload["effects"]["quota_spent"] is True
+        rows_path = runtime / "goals/loopx-turn-fixture/runs/index.jsonl"
+        rows = [json.loads(line) for line in rows_path.read_text(encoding="utf-8").splitlines()]
+        durable = next(row for row in rows if row.get("classification") == "fixture_replan_successor")
+        recorded = json.loads(Path(durable["json_path"]).read_text(encoding="utf-8"))
+        assert recorded["agent_vision"]["path_delta"]["outcome"] == "replan"
+
+
 @pytest.mark.parametrize(
     "next_action",
     ["Select the next Todo from a fresh decision.", "Assess remaining work."],