From 5e62659676aa1cb4f377006c9269462ab5945d96 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Wed, 21 Oct 2026 04:44:39 +0800
Subject: [PATCH 1/4] fix(turn): preserve candidate text for result validation

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../rfcs/agent-loop-effect-interpreter-v0.md  |  9 +++
 .../agent-loop-effect-interpreter-v0.zh-CN.md |  6 ++
 docs/reference/protocols/loopx-turn-v0.md     | 13 +++
 .../turn_driver/host_candidate.py             | 29 +++----
 .../test_host_result_semantics.py             | 81 +++++++++++++++++++
 tests/test_dsh_goal_mode.py                   | 40 ++++++---
 6 files changed, 148 insertions(+), 30 deletions(-)
 create mode 100644 tests/control_plane/test_host_result_semantics.py

diff --git a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md
index 840b67f9df..47fdc14e93 100644
--- a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md
+++ b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md
@@ -599,6 +599,15 @@ This closes the producer-to-detail transport gap only. Live models, packaged
 App/Lark, later authority checks and the remaining convergence matrix retain
 their independent acceptance; no Python projection is retired by this slice.
 
+The shared Host result adapter now retains supplied text for the existing Turn
+validator instead of truncating it into a passing candidate. Over-limit next
+actions could lose a trailing approval constraint while still being accepted.
+File/SQLite native CLI and actual DSH subprocess qualification cover rejection
+before task validation/writeback/spend/ACK, failed-result replay, and a corrected
+fresh Turn settling once. Boundary-length valid results retain their behavior.
+This is result-transport correctness; it does not qualify model adoption, undo
+Host work already performed, or close the three convergence streams.
+
 The first ordinary-CLI consolidation replaces an exactly identical
 `heartbeat_recommendation.blocked_priority_fallback` mirror with
 `blocked_priority_fallback_ref="$.blocked_priority_fallback"`. The complete root
diff --git a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md
index de13ceb6f1..d152b8898a 100644
--- a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md
+++ b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md
@@ -412,6 +412,12 @@ fixture。只读规划与可选 quota capture 默认值不变，原生执行 Tur
 这只关闭 producer 到 detail 的传输缺口。真实模型、打包 App/Lark、后续 authority 检查
 及其余收敛矩阵仍独立验收，本切片不退役任何 Python 投影。
 
+共享 Host 结果适配器现在保留所提供文本，交给既有 Turn 校验器，而不通过截短使
+候选通过。超限后续行动原本可能丢失末尾审批约束却仍获接受。File/SQLite 原生 CLI
+和实际 DSH 子进程验证任务校验、写回、扣费、ACK 前的拒绝、失败结果重放，以及
+修正后的新 Turn 只结算一次。边界长度的合法结果保持原行为。这只证明结果传输
+正确性，不证明模型采用、撤销 Host 已执行工作或关闭原三批收敛验收。
+
 首个普通 CLI 整合将完全相同的
 `heartbeat_recommendation.blocked_priority_fallback` 副本替换为
 `blocked_priority_fallback_ref="$.blocked_priority_fallback"`。根字段的完整事实、
diff --git a/docs/reference/protocols/loopx-turn-v0.md b/docs/reference/protocols/loopx-turn-v0.md
index dea27b7a7c..68eb7116ae 100644
--- a/docs/reference/protocols/loopx-turn-v0.md
+++ b/docs/reference/protocols/loopx-turn-v0.md
@@ -677,6 +677,19 @@ 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.
 
+Shared dsh/optional-cloud adapters preserve candidate text through the existing
+Turn result limits. They must not truncate a supplied classification, summary,
+next action, recommendation or unchanged-path reason to make it pass validation:
+a trailing constraint can change the meaning. The executor rejects over-limit
+text before task validation, durable writeback, quota spend or scheduler ACK.
+Generated adapter diagnostics remain bounded. Existing transport/output limits
+still apply, and a failed result does not undo work the Host already performed.
+
+共享 dsh/可选云端适配器保留候选结果的完整文本，交给现有 Turn 结果校验器判定上限；
+不得截短分类、摘要、后续行动、建议或路径不变理由以使其通过，否则可能丢失末尾约束。
+超限结果在任务验证、持久写回、quota 扣费和 scheduler ACK 前被拒绝。适配器生成的
+诊断仍有界，原传输和输出上限保留；结果被拒绝不会撤销 Host 已经执行的工作。
+
 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`;
diff --git a/loopx/control_plane/turn_driver/host_candidate.py b/loopx/control_plane/turn_driver/host_candidate.py
index c5ba0ac51e..ee29ba120c 100644
--- a/loopx/control_plane/turn_driver/host_candidate.py
+++ b/loopx/control_plane/turn_driver/host_candidate.py
@@ -254,22 +254,14 @@ def _complete_material_fields(result: dict[str, Any], kind: str) -> None:
     """Fill the required delivery fields of a material host candidate."""
     result["delivery_batch_scale"] = "single_surface"
     result["delivery_outcome"] = "outcome_progress"
-    # Material results require these bounded text fields; fill them from
-    # adjacent fields if the model returned a sparse block.
+    # Preserve sparse legacy candidates; the executor validates both supplied
+    # fields and these aliases against the same result contract.
     if not result.get("recommended_action"):
-        result["recommended_action"] = _bounded(
-            result.get("next_action") or result.get("classification") or kind,
-            limit=TEXT_LIMITS["recommended_action"],
-        )
+        result["recommended_action"] = result.get("next_action") or result.get("classification") or kind
     if not result.get("next_action"):
-        result["next_action"] = _bounded(
-            result.get("recommended_action"),
-            limit=TEXT_LIMITS["next_action"],
-        )
+        result["next_action"] = result.get("recommended_action")
     if not result.get("classification"):
-        result["classification"] = _bounded(
-            kind, limit=TEXT_LIMITS["classification"]
-        )
+        result["classification"] = kind
 
 
 def build_result(
@@ -327,11 +319,14 @@ def build_result(
         "result_kind": kind,
         "completed_phases": list(COMPLETED_PHASES),
     }
-    for field, limit in TEXT_LIMITS.items():
+    for field in TEXT_LIMITS:
         if field == "vision_unchanged_reason":
             continue
         value = candidate.get(field)
-        text = _bounded(value, limit=limit)
+        # The executor must see an over-limit candidate and reject it. Clipping
+        # can remove a trailing constraint and turn invalid input into a valid
+        # but semantically different next action or progress claim.
+        text = str(value or "").strip()
         if text:
             result[field] = text
 
@@ -346,9 +341,7 @@ def build_result(
     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"]
-        )
+        result["vision_unchanged_reason"] = str(unchanged_reason).strip()
     elif not has_path_delta:
         # Keep the legacy material-without-packet behavior for older hosts.
         result["vision_unchanged_reason"] = _bounded(
diff --git a/tests/control_plane/test_host_result_semantics.py b/tests/control_plane/test_host_result_semantics.py
new file mode 100644
index 0000000000..bf57e6a118
--- /dev/null
+++ b/tests/control_plane/test_host_result_semantics.py
@@ -0,0 +1,81 @@
+"""A real DSH subprocess cannot hide invalid result text from native settlement."""
+from __future__ import annotations
+
+import json
+import sys
+
+import pytest
+
+from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
+from tests.control_plane.canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime
+from tests.control_plane.test_turn_host_decision_capture import _fixture_module
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_native_turn_rejects_full_overlimit_candidate_before_writeback(tmp_path, monkeypatch, provider):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    f = _fixture_module()
+    project, runtime, workspace, registry = f._write_fixture(tmp_path, turn_count=2)
+    prefix = ["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json"]
+    code, listed = f._run_cli([*prefix, "todo", "list", "--goal-id", f.GOAL_ID])
+    assert code == 0, listed
+    state = project / ".codex/goals" / f.GOAL_ID / "ACTIVE_GOAL_STATE.md"
+    initialize_canonical_authority(runtime, f.GOAL_ID,
+        build_todo_runtime_shadow_projection(goal_id=f.GOAL_ID, handoff_mode="soft_claim", todos=listed["todos"]),
+        state_path=state, provider=provider)
+    candidate = {
+        "result_kind": "validated_progress", "classification": "artifact_checked",
+        "summary": "The isolated artifact was checked.",
+        "recommended_action": "Continue the accepted task.",
+        "next_action": "Review the evidence. " * 65 + "Stop before publication; approval is still required.",
+        "vision_unchanged_reason": "The accepted Goal remains unchanged.",
+    }
+    (workspace / "candidate.json").write_text(json.dumps(candidate))
+    runner = tmp_path / "runner.py"
+    runner.write_text(
+        "import json\nfrom pathlib import Path\n"
+        "def run_dsh_turn(*, prompt, workspace, **kwargs):\n"
+        "    root = Path(workspace)\n"
+        "    with (root / 'invocations.txt').open('a') as out: out.write('host\\n')\n"
+        "    (root / 'received.txt').write_text(prompt)\n"
+        "    return (root / 'candidate.json').read_text()\n"
+    )
+    validator = [sys.executable, "-c",
+        "import json,pathlib,sys; json.load(sys.stdin); "
+        "pathlib.Path('validator-ran').write_text('verified')"]
+    base = [*prefix, "turn", "run-once", "--goal-id", f.GOAL_ID, "--agent-id", f.AGENT_ID,
+        "--host", "generic-cli", "--execution-mode", "isolated-headless", "--project", str(workspace),
+        "--host-command-json", json.dumps([sys.executable, "-m", "loopx.dsh_goal_mode",
+            "--workspace", str(workspace), "--dsh-home", str(tmp_path / "dsh-home"), "--dsh-runner", str(runner)]),
+        "--validation-command-json", json.dumps(validator), "--scan-root", str(project),
+        "--no-global-sync", "--timeout-seconds", "30"]
+    code, failed = f._run_cli([*base, "--turn-instance-id", "overlimit-result", "--execute"])
+    assert code == 1 and failed["status"] == "failed", failed
+    assert "next_action exceeds 1200 characters" in json.dumps(failed)
+    assert failed["validation_stage"] == "host_result_contract"
+    assert failed["effects"] == {
+        "host_invoked": True, "state_written": False,
+        "quota_spent": False, "scheduler_acknowledged": False,
+    }
+    assert not (workspace / "validator-ran").exists()
+    assert f._quota_spend_count(runtime) == 0
+    code, current = f._run_cli([*prefix, "todo", "list", "--goal-id", f.GOAL_ID])
+    assert code == 0 and current["todos"] == listed["todos"]
+
+    # A correction is a fresh admitted Turn, not a replay of an invalid result.
+    candidate["next_action"] = "Stop before publication; approval is still required."
+    (workspace / "candidate.json").write_text(json.dumps(candidate))
+    code, repeated = f._run_cli([*base, "--resume-turn-key", failed["resume_turn_key"], "--execute"])
+    assert code == 1 and repeated["status"] == "failed", repeated
+    assert not any(repeated["effects"].values())
+    assert (workspace / "invocations.txt").read_text() == "host\n"
+    assert not (workspace / "validator-ran").exists()
+    code, recovered = f._run_cli([*base, "--turn-instance-id", "corrected-result", "--execute"])
+    assert code == 0 and recovered["status"] == "committed", recovered
+    assert (workspace / "validator-ran").read_text() == "verified"
+    assert f._quota_spend_count(runtime) == 1
+    code, replay = f._run_cli([*base, "--resume-turn-key", recovered["resume_turn_key"], "--execute"])
+    assert code == 0 and replay["ok"], replay
+    assert not any(replay["effects"].values())
+    assert f._quota_spend_count(runtime) == 1
+    assert (workspace / "invocations.txt").read_text() == "host\nhost\n"
diff --git a/tests/test_dsh_goal_mode.py b/tests/test_dsh_goal_mode.py
index 50215fd1aa..0a1f22dedb 100644
--- a/tests/test_dsh_goal_mode.py
+++ b/tests/test_dsh_goal_mode.py
@@ -632,20 +632,36 @@ def test_build_result_is_accepted_by_the_real_host_result_validator(
     assert verdict["ok"], verdict["errors"]
 
 
-def test_build_result_bounds_every_free_text_field() -> None:
+@pytest.mark.parametrize("field,limit", [
+    ("classification", 120), ("summary", 400), ("next_action", 1_200),
+    ("recommended_action", 1_200), ("vision_unchanged_reason", 240),
+])
+@pytest.mark.parametrize("kind", ["validated_progress", "wait"])
+@pytest.mark.parametrize("extra", [0, 1])
+def test_candidate_text_reaches_executor_limits_without_losing_its_tail(
+    field: str, limit: int, kind: str, extra: int,
+) -> None:
+    # Independent limits from the Turn result contract. A trailing restriction
+    # must reach the validator, including on stop and legacy unchanged paths.
+    tail = "Stop before publication."
+    value = "验" * (limit + extra - len(tail)) + tail
+    candidate = {
+        "result_kind": kind, "classification": "implementation_refined",
+        "summary": "The isolated artifact was checked.",
+        "next_action": "Continue the accepted work.",
+        "recommended_action": "Read the next current decision.",
+        "vision_unchanged_reason": "The accepted Goal remains unchanged.",
+        field: value,
+    }
+    request = _signed_request()
     result = turn_host_adapter.build_result(
-        _signed_request(),
-        {
-            "result_kind": "user_action_required",
-            "classification": "x" * 500,
-            "summary": "y" * 900,
-            "next_action": "z" * 5000,
-        },
+        request, candidate,
     )
-    # Independent limits from the Turn result contract, not the adapter's table.
-    assert len(result["classification"]) <= 120
-    assert len(result["summary"]) <= 400
-    assert len(result["next_action"]) <= 1_200
+    assert result[field] == value
+    verdict = validate_loopx_turn_host_result(_turn_plan(request), result)
+    assert verdict["ok"] is (extra == 0), verdict["errors"]
+    if extra:
+        assert f"{field} exceeds {limit} characters" in verdict["errors"]
 
 
 def test_adapter_runs_hermetically_through_the_module_entry() -> None:

From 052ef1179ea49dca2eb84a7e15ad42c2dba7cfaf Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Wed, 21 Oct 2026 05:13:10 +0800
Subject: [PATCH 2/4] fix(turn): reject nontext candidates without string
 coercion

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../rfcs/agent-loop-effect-interpreter-v0.md  |  3 ++
 .../agent-loop-effect-interpreter-v0.zh-CN.md |  4 +-
 docs/reference/protocols/loopx-turn-v0.md     |  8 ++++
 loopx/control_plane/turn_driver/executor.py   |  8 +++-
 .../turn_driver/host_candidate.py             | 16 ++++----
 .../test_host_result_semantics.py             | 16 ++++++--
 tests/test_dsh_goal_mode.py                   | 38 +++++++++++++++++++
 7 files changed, 80 insertions(+), 13 deletions(-)

diff --git a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md
index 47fdc14e93..564900d886 100644
--- a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md
+++ b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md
@@ -602,6 +602,9 @@ their independent acceptance; no Python projection is retired by this slice.
 The shared Host result adapter now retains supplied text for the existing Turn
 validator instead of truncating it into a passing candidate. Over-limit next
 actions could lose a trailing approval constraint while still being accepted.
+JSON booleans, numbers and containers also retain their types for rejection by
+that validator; false/zero cannot disappear into sparse-field defaults. Legacy
+null/omitted optional text retains its existing absence semantics.
 File/SQLite native CLI and actual DSH subprocess qualification cover rejection
 before task validation/writeback/spend/ACK, failed-result replay, and a corrected
 fresh Turn settling once. Boundary-length valid results retain their behavior.
diff --git a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md
index d152b8898a..bf6ef4198e 100644
--- a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md
+++ b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md
@@ -413,7 +413,9 @@ fixture。只读规划与可选 quota capture 默认值不变，原生执行 Tur
 及其余收敛矩阵仍独立验收，本切片不退役任何 Python 投影。
 
 共享 Host 结果适配器现在保留所提供文本，交给既有 Turn 校验器，而不通过截短使
-候选通过。超限后续行动原本可能丢失末尾审批约束却仍获接受。File/SQLite 原生 CLI
+候选通过。超限后续行动原本可能丢失末尾审批约束却仍获接受。JSON 布尔值、数字和
+容器也保留原类型，交给同一校验器拒绝；false/零不能消失到
+稀疏字段默认值，旧 null/省略的可选文本仍按缺省处理。File/SQLite 原生 CLI
 和实际 DSH 子进程验证任务校验、写回、扣费、ACK 前的拒绝、失败结果重放，以及
 修正后的新 Turn 只结算一次。边界长度的合法结果保持原行为。这只证明结果传输
 正确性，不证明模型采用、撤销 Host 已执行工作或关闭原三批收敛验收。
diff --git a/docs/reference/protocols/loopx-turn-v0.md b/docs/reference/protocols/loopx-turn-v0.md
index 68eb7116ae..8dc9679ee0 100644
--- a/docs/reference/protocols/loopx-turn-v0.md
+++ b/docs/reference/protocols/loopx-turn-v0.md
@@ -684,11 +684,19 @@ a trailing constraint can change the meaning. The executor rejects over-limit
 text before task validation, durable writeback, quota spend or scheduler ACK.
 Generated adapter diagnostics remain bounded. Existing transport/output limits
 still apply, and a failed result does not undo work the Host already performed.
+Candidate text fields retain their JSON types until validation: booleans,
+numbers, arrays and objects are rejected, including false, zero and empty
+containers. They must not become instructions through string coercion or
+missing-field fallbacks. Legacy null/omitted optional text keeps its absence
+semantics; sparse adapter candidates retain their established defaults.
 
 共享 dsh/可选云端适配器保留候选结果的完整文本，交给现有 Turn 结果校验器判定上限；
 不得截短分类、摘要、后续行动、建议或路径不变理由以使其通过，否则可能丢失末尾约束。
 超限结果在任务验证、持久写回、quota 扣费和 scheduler ACK 前被拒绝。适配器生成的
 诊断仍有界，原传输和输出上限保留；结果被拒绝不会撤销 Host 已经执行的工作。
+候选文本字段保留原 JSON 类型直到校验；布尔值、数字、数组和对象（含 false、零和
+空容器）会被拒绝，不得经字符串转换或缺字段兜底变成行动。旧 null/省略的可选
+文本继续按缺省处理，稀疏适配器候选保留既有默认值。
 
 The existing TypeScript Vision normalizer owns packet interpretation and emits
 the canonical `goal_vision_replan_contract_v0` shape. Its current compatibility
diff --git a/loopx/control_plane/turn_driver/executor.py b/loopx/control_plane/turn_driver/executor.py
index abc4c89c9f..b7d617d9e2 100644
--- a/loopx/control_plane/turn_driver/executor.py
+++ b/loopx/control_plane/turn_driver/executor.py
@@ -218,7 +218,13 @@ def _bounded_public_text(
     required: bool,
     errors: list[str],
 ) -> str | None:
-    text = str(result.get(field) or "").strip()
+    value = result.get(field)
+    # Missing/null legacy optional text remains absent. Other JSON types must
+    # not acquire instruction meaning through Python string coercion.
+    if value is not None and not isinstance(value, str):
+        errors.append(f"{field} must be a string")
+        return None
+    text = (value or "").strip()
     if required and not text:
         errors.append(f"{field} is required")
         return None
diff --git a/loopx/control_plane/turn_driver/host_candidate.py b/loopx/control_plane/turn_driver/host_candidate.py
index ee29ba120c..dc908726d3 100644
--- a/loopx/control_plane/turn_driver/host_candidate.py
+++ b/loopx/control_plane/turn_driver/host_candidate.py
@@ -256,11 +256,11 @@ def _complete_material_fields(result: dict[str, Any], kind: str) -> None:
     result["delivery_outcome"] = "outcome_progress"
     # Preserve sparse legacy candidates; the executor validates both supplied
     # fields and these aliases against the same result contract.
-    if not result.get("recommended_action"):
+    if "recommended_action" not in result:
         result["recommended_action"] = result.get("next_action") or result.get("classification") or kind
-    if not result.get("next_action"):
+    if "next_action" not in result:
         result["next_action"] = result.get("recommended_action")
-    if not result.get("classification"):
+    if "classification" not in result:
         result["classification"] = kind
 
 
@@ -326,8 +326,8 @@ def build_result(
         # The executor must see an over-limit candidate and reject it. Clipping
         # can remove a trailing constraint and turn invalid input into a valid
         # but semantically different next action or progress claim.
-        text = str(value or "").strip()
-        if text:
+        text = value.strip() if isinstance(value, str) else value
+        if text or (value is not None and not isinstance(value, str)):
             result[field] = text
 
     if kind in MATERIAL_KINDS:
@@ -340,8 +340,10 @@ def build_result(
 
     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"] = str(unchanged_reason).strip()
+    if unchanged_reason is not None and not isinstance(unchanged_reason, str):
+        result["vision_unchanged_reason"] = unchanged_reason
+    elif unchanged_reason:
+        result["vision_unchanged_reason"] = unchanged_reason.strip()
     elif not has_path_delta:
         # Keep the legacy material-without-packet behavior for older hosts.
         result["vision_unchanged_reason"] = _bounded(
diff --git a/tests/control_plane/test_host_result_semantics.py b/tests/control_plane/test_host_result_semantics.py
index bf57e6a118..a3a5145228 100644
--- a/tests/control_plane/test_host_result_semantics.py
+++ b/tests/control_plane/test_host_result_semantics.py
@@ -12,7 +12,15 @@
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
-def test_native_turn_rejects_full_overlimit_candidate_before_writeback(tmp_path, monkeypatch, provider):
+@pytest.mark.parametrize("invalid_action,error", [
+    ("Review the evidence. " * 65 + "Stop before publication; approval is still required.",
+     "next_action exceeds 1200 characters"),
+    ({"action": "Continue", "constraint": "Stop before publication."}, "next_action must be a string"),
+    (False, "next_action must be a string"),
+], ids=["overlimit", "object", "false"])
+def test_native_turn_rejects_invalid_candidate_before_writeback(
+    tmp_path, monkeypatch, provider, invalid_action, error,
+):
     isolate_sqlite_runtime(tmp_path, monkeypatch)
     f = _fixture_module()
     project, runtime, workspace, registry = f._write_fixture(tmp_path, turn_count=2)
@@ -27,7 +35,7 @@ def test_native_turn_rejects_full_overlimit_candidate_before_writeback(tmp_path,
         "result_kind": "validated_progress", "classification": "artifact_checked",
         "summary": "The isolated artifact was checked.",
         "recommended_action": "Continue the accepted task.",
-        "next_action": "Review the evidence. " * 65 + "Stop before publication; approval is still required.",
+        "next_action": invalid_action,
         "vision_unchanged_reason": "The accepted Goal remains unchanged.",
     }
     (workspace / "candidate.json").write_text(json.dumps(candidate))
@@ -49,9 +57,9 @@ def test_native_turn_rejects_full_overlimit_candidate_before_writeback(tmp_path,
             "--workspace", str(workspace), "--dsh-home", str(tmp_path / "dsh-home"), "--dsh-runner", str(runner)]),
         "--validation-command-json", json.dumps(validator), "--scan-root", str(project),
         "--no-global-sync", "--timeout-seconds", "30"]
-    code, failed = f._run_cli([*base, "--turn-instance-id", "overlimit-result", "--execute"])
+    code, failed = f._run_cli([*base, "--turn-instance-id", "invalid-result", "--execute"])
     assert code == 1 and failed["status"] == "failed", failed
-    assert "next_action exceeds 1200 characters" in json.dumps(failed)
+    assert error in json.dumps(failed)
     assert failed["validation_stage"] == "host_result_contract"
     assert failed["effects"] == {
         "host_invoked": True, "state_written": False,
diff --git a/tests/test_dsh_goal_mode.py b/tests/test_dsh_goal_mode.py
index 0a1f22dedb..8b3f19a507 100644
--- a/tests/test_dsh_goal_mode.py
+++ b/tests/test_dsh_goal_mode.py
@@ -664,6 +664,44 @@ def test_candidate_text_reaches_executor_limits_without_losing_its_tail(
         assert f"{field} exceeds {limit} characters" in verdict["errors"]
 
 
+@pytest.mark.parametrize("field", [
+    "classification", "summary", "next_action", "recommended_action", "vision_unchanged_reason",
+])
+@pytest.mark.parametrize("value", [True, False, 42, 0, 1.25, {}, [], {"action": "Continue"}, ["Continue"]])
+@pytest.mark.parametrize("kind", ["validated_progress", "wait"])
+def test_nontext_candidate_fields_reach_the_executor_without_coercion(field, value, kind):
+    request = _signed_request()
+    valid = {
+        "result_kind": kind, "classification": "artifact_checked", "summary": "Evidence checked.",
+        "next_action": "Continue the accepted task.", "recommended_action": "Read current requirements.",
+        "vision_unchanged_reason": "The accepted Goal is unchanged.",
+    }
+    adapted = turn_host_adapter.build_result(request, {**valid, field: value})
+    assert type(adapted[field]) is type(value) and adapted[field] == value
+    # The generic host route must reject the same raw JSON type even without
+    # the shared DSH/Ark adapter. A transport cannot repair malformed meaning.
+    direct = {**turn_host_adapter.build_result(request, valid), field: value}
+    for result in (adapted, direct):
+        verdict = validate_loopx_turn_host_result(_turn_plan(request), result)
+        assert not verdict["ok"]
+        assert f"{field} must be a string" in verdict["errors"]
+
+
+@pytest.mark.parametrize("field", [
+    "classification", "summary", "next_action", "recommended_action", "vision_unchanged_reason",
+])
+@pytest.mark.parametrize("kind", ["validated_progress", "wait"])
+def test_legacy_null_candidate_text_still_uses_the_existing_absent_field_behavior(field, kind):
+    request = _signed_request()
+    result = turn_host_adapter.build_result(request, {
+        "result_kind": kind, "classification": "artifact_checked", "summary": "Evidence checked.",
+        "next_action": "Continue the accepted task.", "recommended_action": "Read current requirements.",
+        "vision_unchanged_reason": "The accepted Goal is unchanged.", field: None,
+    })
+    verdict = validate_loopx_turn_host_result(_turn_plan(request), result)
+    assert verdict["ok"], verdict["errors"]
+
+
 def test_adapter_runs_hermetically_through_the_module_entry() -> None:
     # `python -m loopx.dsh_goal_mode` must drive the same stdin/stdout
     # contract as the legacy script, using a fake dsh runner.

From c450a3867c55eeff10dc91d288bcceb5b886fe3b Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Wed, 21 Oct 2026 07:38:35 +0800
Subject: [PATCH 3/4] fix(turn): preserve declared delivery outcomes through
 host adapters

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../rfcs/agent-loop-effect-interpreter-v0.md  |  8 ++
 .../agent-loop-effect-interpreter-v0.zh-CN.md |  8 +-
 docs/reference/protocols/loopx-turn-v0.md     | 14 ++++
 .../turn_driver/host_candidate.py             | 16 +++-
 .../test_host_result_semantics.py             | 75 +++++++++++++------
 tests/test_dsh_goal_mode.py                   | 31 ++++++++
 6 files changed, 126 insertions(+), 26 deletions(-)

diff --git a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md
index 564900d886..aa478cdbdf 100644
--- a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md
+++ b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md
@@ -605,6 +605,14 @@ actions could lose a trailing approval constraint while still being accepted.
 JSON booleans, numbers and containers also retain their types for rejection by
 that validator; false/zero cannot disappear into sparse-field defaults. Legacy
 null/omitted optional text retains its existing absence semantics.
+Explicit delivery scale and outcome now retain their values and JSON types;
+only omitted metadata uses the existing sparse defaults. Previously a declared
+`surface_only`, unqualified `outcome_gap` or invalid value became
+`outcome_progress`, bypassing the existing refusal and recording a quota spend.
+The shared prompt uses the existing delivery vocabularies. Native File/SQLite
+tests preserve valid scale history, reject invalid vocabulary before task
+validation, and reject nonqualifying outcomes before writeback/spend/ACK even
+when the independent task check passes. No settlement policy is added here.
 File/SQLite native CLI and actual DSH subprocess qualification cover rejection
 before task validation/writeback/spend/ACK, failed-result replay, and a corrected
 fresh Turn settling once. Boundary-length valid results retain their behavior.
diff --git a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md
index bf6ef4198e..61cdc5961e 100644
--- a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md
+++ b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md
@@ -415,7 +415,13 @@ fixture。只读规划与可选 quota capture 默认值不变，原生执行 Tur
 共享 Host 结果适配器现在保留所提供文本，交给既有 Turn 校验器，而不通过截短使
 候选通过。超限后续行动原本可能丢失末尾审批约束却仍获接受。JSON 布尔值、数字和
 容器也保留原类型，交给同一校验器拒绝；false/零不能消失到
-稀疏字段默认值，旧 null/省略的可选文本仍按缺省处理。File/SQLite 原生 CLI
+稀疏字段默认值，旧 null/省略的可选文本仍按缺省处理。明确提供的交付规模与结果
+现在也保留原值及 JSON 类型，只有省略元数据时才沿用稀疏默认值。此前声明的
+`surface_only`、不满足条件的 `outcome_gap` 或非法值都会变成 `outcome_progress`，
+绕过既有拒绝并记入 quota 扣费。共享 prompt 使用既有交付词汇。原生 File/SQLite
+测试保留合法规模历史，在任务验证前拒绝非法词汇，并证明独立任务检查通过后，
+不满足条件的结果仍在写回／扣费／ACK 前被拒绝；此处没有新增结算策略。
+File/SQLite 原生 CLI
 和实际 DSH 子进程验证任务校验、写回、扣费、ACK 前的拒绝、失败结果重放，以及
 修正后的新 Turn 只结算一次。边界长度的合法结果保持原行为。这只证明结果传输
 正确性，不证明模型采用、撤销 Host 已执行工作或关闭原三批收敛验收。
diff --git a/docs/reference/protocols/loopx-turn-v0.md b/docs/reference/protocols/loopx-turn-v0.md
index 8dc9679ee0..bf493a52b8 100644
--- a/docs/reference/protocols/loopx-turn-v0.md
+++ b/docs/reference/protocols/loopx-turn-v0.md
@@ -689,6 +689,14 @@ numbers, arrays and objects are rejected, including false, zero and empty
 containers. They must not become instructions through string coercion or
 missing-field fallbacks. Legacy null/omitted optional text keeps its absence
 semantics; sparse adapter candidates retain their established defaults.
+Explicit `delivery_batch_scale` and `delivery_outcome` also pass through to
+their existing validators; only omitted metadata uses the legacy
+`single_surface` / `outcome_progress` defaults. The adapter must not replace a
+declared result with progress or accept read-only scale aliases for new writes.
+Valid vocabulary does not establish a qualifying settlement: `surface_only`
+and an `outcome_gap` without its required typed blocker remain subject to the
+existing refresh-state refusal after task validation, before writeback, spend
+or ACK. Valid declared scales reach durable delivery history unchanged.
 
 共享 dsh/可选云端适配器保留候选结果的完整文本，交给现有 Turn 结果校验器判定上限；
 不得截短分类、摘要、后续行动、建议或路径不变理由以使其通过，否则可能丢失末尾约束。
@@ -697,6 +705,12 @@ semantics; sparse adapter candidates retain their established defaults.
 候选文本字段保留原 JSON 类型直到校验；布尔值、数字、数组和对象（含 false、零和
 空容器）会被拒绝，不得经字符串转换或缺字段兜底变成行动。旧 null/省略的可选
 文本继续按缺省处理，稀疏适配器候选保留既有默认值。
+明确提供的 `delivery_batch_scale` 与 `delivery_outcome` 也原样交给既有校验器；
+只有省略元数据时才使用旧 `single_surface` / `outcome_progress` 默认值。适配器
+不得将所声明结果替换为进展，也不得让只读 scale 别名用于新写入。词汇合法不代表
+满足结算条件：`surface_only` 及缺少规定 typed blocker 的 `outcome_gap` 仍会在
+任务验证后、写回／扣费／ACK 前被既有 refresh-state 拒绝。合法声明的规模原样
+进入持久交付历史。
 
 The existing TypeScript Vision normalizer owns packet interpretation and emits
 the canonical `goal_vision_replan_contract_v0` shape. Its current compatibility
diff --git a/loopx/control_plane/turn_driver/host_candidate.py b/loopx/control_plane/turn_driver/host_candidate.py
index dc908726d3..4887e2c154 100644
--- a/loopx/control_plane/turn_driver/host_candidate.py
+++ b/loopx/control_plane/turn_driver/host_candidate.py
@@ -12,6 +12,8 @@
 from typing import Any
 
 from ..quota.turn_envelope import turn_envelope_action_signature_document
+from ..work_items.delivery_batch_scale import DELIVERY_BATCH_SCALE_CHOICES
+from ..work_items.delivery_outcome import DELIVERY_OUTCOME_CHOICES
 from .driver import selected_turn_todo
 
 LOOPX_TURN_HOST_REQUEST_SCHEMA = "loopx_turn_host_request_v0"
@@ -192,6 +194,8 @@ 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"
+        f"- delivery_batch_scale (material results): {' | '.join(DELIVERY_BATCH_SCALE_CHOICES)}\n"
+        f"- delivery_outcome (material results): {' | '.join(DELIVERY_OUTCOME_CHOICES)}\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 "
@@ -250,10 +254,14 @@ def parse_model_json(text: str) -> dict[str, Any] | None:
     return parsed if isinstance(parsed, dict) else None
 
 
-def _complete_material_fields(result: dict[str, Any], kind: str) -> None:
+def _complete_material_fields(
+    result: dict[str, Any], candidate: Mapping[str, Any], kind: str,
+) -> None:
     """Fill the required delivery fields of a material host candidate."""
-    result["delivery_batch_scale"] = "single_surface"
-    result["delivery_outcome"] = "outcome_progress"
+    # Only omitted metadata gets the legacy defaults. Supplied values, including
+    # null and invalid JSON types, belong to the existing executor validators.
+    result["delivery_batch_scale"] = candidate.get("delivery_batch_scale", "single_surface")
+    result["delivery_outcome"] = candidate.get("delivery_outcome", "outcome_progress")
     # Preserve sparse legacy candidates; the executor validates both supplied
     # fields and these aliases against the same result contract.
     if "recommended_action" not in result:
@@ -331,7 +339,7 @@ def build_result(
             result[field] = text
 
     if kind in MATERIAL_KINDS:
-        _complete_material_fields(result, kind)
+        _complete_material_fields(result, candidate, kind)
     for field in ("path_delta_mode", "agent_vision_json"):
         if field in candidate:
             # Preserve the structured executor input verbatim. The executor
diff --git a/tests/control_plane/test_host_result_semantics.py b/tests/control_plane/test_host_result_semantics.py
index a3a5145228..bfcf49fd5e 100644
--- a/tests/control_plane/test_host_result_semantics.py
+++ b/tests/control_plane/test_host_result_semantics.py
@@ -1,4 +1,4 @@
-"""A real DSH subprocess cannot hide invalid result text from native settlement."""
+"""A real DSH subprocess preserves the result that native settlement validates."""
 from __future__ import annotations
 
 import json
@@ -12,14 +12,19 @@
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
-@pytest.mark.parametrize("invalid_action,error", [
-    ("Review the evidence. " * 65 + "Stop before publication; approval is still required.",
-     "next_action exceeds 1200 characters"),
-    ({"action": "Continue", "constraint": "Stop before publication."}, "next_action must be a string"),
-    (False, "next_action must be a string"),
-], ids=["overlimit", "object", "false"])
+@pytest.mark.parametrize("candidate_fields,error,before_task_validation", [
+    ({"next_action": "Review the evidence. " * 65 + "Stop before publication; approval is still required."},
+     "next_action exceeds 1200 characters", True),
+    ({"next_action": {"action": "Continue", "constraint": "Stop before publication."}},
+     "next_action must be a string", True),
+    ({"next_action": False}, "next_action must be a string", True),
+    ({"delivery_batch_scale": "single_segment"}, "legacy alias 'single_segment' is read-only", True),
+    ({"delivery_outcome": False}, "delivery_outcome must be one of", True),
+    ({"delivery_outcome": "surface_only"}, "cannot settle a Turn", False),
+    ({"delivery_outcome": "outcome_gap"}, "blocked", False),
+], ids=["overlimit", "object", "false", "legacy-scale", "invalid-outcome", "surface-only", "untyped-gap"])
 def test_native_turn_rejects_invalid_candidate_before_writeback(
-    tmp_path, monkeypatch, provider, invalid_action, error,
+    tmp_path, monkeypatch, provider, candidate_fields, error, before_task_validation,
 ):
     isolate_sqlite_runtime(tmp_path, monkeypatch)
     f = _fixture_module()
@@ -35,8 +40,9 @@ def test_native_turn_rejects_invalid_candidate_before_writeback(
         "result_kind": "validated_progress", "classification": "artifact_checked",
         "summary": "The isolated artifact was checked.",
         "recommended_action": "Continue the accepted task.",
-        "next_action": invalid_action,
+        "next_action": "Continue the accepted task.",
         "vision_unchanged_reason": "The accepted Goal remains unchanged.",
+        **candidate_fields,
     }
     (workspace / "candidate.json").write_text(json.dumps(candidate))
     runner = tmp_path / "runner.py"
@@ -58,30 +64,57 @@ def test_native_turn_rejects_invalid_candidate_before_writeback(
         "--validation-command-json", json.dumps(validator), "--scan-root", str(project),
         "--no-global-sync", "--timeout-seconds", "30"]
     code, failed = f._run_cli([*base, "--turn-instance-id", "invalid-result", "--execute"])
-    assert code == 1 and failed["status"] == "failed", failed
+    assert code == 1 and not failed["ok"], failed
     assert error in json.dumps(failed)
-    assert failed["validation_stage"] == "host_result_contract"
-    assert failed["effects"] == {
-        "host_invoked": True, "state_written": False,
-        "quota_spent": False, "scheduler_acknowledged": False,
-    }
-    assert not (workspace / "validator-ran").exists()
+    if before_task_validation:
+        assert failed["status"] == "failed"
+        assert failed["validation_stage"] == "host_result_contract"
+        assert failed["effects"] == {
+            "host_invoked": True, "state_written": False,
+            "quota_spent": False, "scheduler_acknowledged": False,
+        }
+        resume_key = failed["resume_turn_key"]
+    else:
+        # Vocabulary-valid outcomes still face the existing native settlement
+        # rule. Its CLI error leaves the validated journal for recovery.
+        journal_path, = (runtime / "goals" / f.GOAL_ID / "turns").glob("*.json")
+        journal = json.loads(journal_path.read_text())
+        assert journal["host_result"]["delivery_outcome"] == candidate_fields["delivery_outcome"]
+        assert journal["completed_phases"] == ["host_execute", "typed_result", "validation"]
+        assert "writeback" not in journal and "quota_spend" not in journal
+        resume_key = journal["turn_key"]
+    assert (workspace / "validator-ran").exists() is not before_task_validation
     assert f._quota_spend_count(runtime) == 0
     code, current = f._run_cli([*prefix, "todo", "list", "--goal-id", f.GOAL_ID])
     assert code == 0 and current["todos"] == listed["todos"]
 
     # A correction is a fresh admitted Turn, not a replay of an invalid result.
-    candidate["next_action"] = "Stop before publication; approval is still required."
+    candidate.update({
+        "next_action": "Stop before publication; approval is still required.",
+        "delivery_batch_scale": "implementation", "delivery_outcome": "outcome_progress",
+    })
     (workspace / "candidate.json").write_text(json.dumps(candidate))
-    code, repeated = f._run_cli([*base, "--resume-turn-key", failed["resume_turn_key"], "--execute"])
-    assert code == 1 and repeated["status"] == "failed", repeated
-    assert not any(repeated["effects"].values())
+    code, repeated = f._run_cli([*base, "--resume-turn-key", resume_key, "--execute"])
+    assert code == 1 and not repeated["ok"], repeated
+    if before_task_validation:
+        assert repeated["status"] == "failed" and not any(repeated["effects"].values())
+    else:
+        assert error in json.dumps(repeated)
+        journal = json.loads(journal_path.read_text())
+        assert journal["completed_phases"] == ["host_execute", "typed_result", "validation"]
+        assert "writeback" not in journal and "quota_spend" not in journal
+    assert f._quota_spend_count(runtime) == 0
     assert (workspace / "invocations.txt").read_text() == "host\n"
-    assert not (workspace / "validator-ran").exists()
+    assert (workspace / "validator-ran").exists() is not before_task_validation
     code, recovered = f._run_cli([*base, "--turn-instance-id", "corrected-result", "--execute"])
     assert code == 0 and recovered["status"] == "committed", recovered
     assert (workspace / "validator-ran").read_text() == "verified"
     assert f._quota_spend_count(runtime) == 1
+    rows = [json.loads(line) for line in
+        (runtime / "goals" / f.GOAL_ID / "runs/index.jsonl").read_text().splitlines()]
+    progress = next(row for row in rows if row.get("classification") == "artifact_checked")
+    assert progress["delivery_batch_scale"] == "implementation"
+    assert progress["delivery_outcome"] == "outcome_progress"
     code, replay = f._run_cli([*base, "--resume-turn-key", recovered["resume_turn_key"], "--execute"])
     assert code == 0 and replay["ok"], replay
     assert not any(replay["effects"].values())
diff --git a/tests/test_dsh_goal_mode.py b/tests/test_dsh_goal_mode.py
index 8b3f19a507..cc796e28e2 100644
--- a/tests/test_dsh_goal_mode.py
+++ b/tests/test_dsh_goal_mode.py
@@ -459,6 +459,37 @@ def test_build_result_shapes_material_results_with_required_fields() -> None:
     assert "agent_vision_json" not in result
 
 
+@pytest.mark.parametrize("kind", ["validated_progress", "repair_required", "replan_required"])
+@pytest.mark.parametrize("field,value,valid", [
+    ("delivery_batch_scale", "test_only", True),
+    ("delivery_batch_scale", "multi_surface", True),
+    ("delivery_batch_scale", "implementation", True),
+    ("delivery_outcome", "surface_only", True),
+    ("delivery_outcome", "outcome_gap", True),
+    ("delivery_outcome", "primary_goal_outcome", True),
+    ("delivery_batch_scale", "single_segment", False),
+    ("delivery_batch_scale", "bounded_segment", False),
+    *[(field, value, False)
+      for field in ("delivery_batch_scale", "delivery_outcome")
+      for value in (None, False, 0, {}, [], "", "unknown")],
+])
+def test_explicit_delivery_metadata_reaches_its_existing_validator(kind, field, value, valid):
+    request = _signed_request()
+    base = {
+        "result_kind": kind, "classification": "artifact_checked",
+        "next_action": "Continue the accepted task.",
+        "vision_unchanged_reason": "The accepted Goal remains unchanged.",
+    }
+    adapted = turn_host_adapter.build_result(request, {**base, field: value})
+    assert type(adapted[field]) is type(value) and adapted[field] == value
+    direct = {**turn_host_adapter.build_result(request, base), field: value}
+    # Legal outcome names are not proof that native settlement can accept
+    # them. This checks only the transport and existing vocabulary contract.
+    for result in (adapted, direct):
+        verdict = validate_loopx_turn_host_result(_turn_plan(request), result)
+        assert verdict["ok"] is valid, verdict["errors"]
+
+
 def _material_replan_candidate() -> tuple[dict[str, object], str]:
     vision = {
         "schema_version": "goal_vision_replan_contract_v0",

From 712e88ed9419349743b962586a1b4827fd50bfbf Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Wed, 21 Oct 2026 08:58:58 +0800
Subject: [PATCH 4/4] fix(turn): reject nonqualifying candidates before
 preparing writes

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../rfcs/agent-loop-effect-interpreter-v0.md  | 13 +++-
 .../agent-loop-effect-interpreter-v0.zh-CN.md |  9 ++-
 docs/reference/protocols/loopx-turn-v0.md     | 25 +++++---
 loopx/control_plane/turn_driver/executor.py   | 15 ++++-
 .../test_host_result_semantics.py             | 62 ++++++++-----------
 tests/test_dsh_goal_mode.py                   |  8 +--
 tests/test_loopx_turn_executor.py             | 48 ++++++++++++++
 7 files changed, 128 insertions(+), 52 deletions(-)

diff --git a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md
index aa478cdbdf..362729b842 100644
--- a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md
+++ b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.md
@@ -610,9 +610,16 @@ only omitted metadata uses the existing sparse defaults. Previously a declared
 `surface_only`, unqualified `outcome_gap` or invalid value became
 `outcome_progress`, bypassing the existing refusal and recording a quota spend.
 The shared prompt uses the existing delivery vocabularies. Native File/SQLite
-tests preserve valid scale history, reject invalid vocabulary before task
-validation, and reject nonqualifying outcomes before writeback/spend/ACK even
-when the independent task check passes. No settlement policy is added here.
+tests preserve valid scale history. New material results also reuse the existing
+settlement rule before task validation or effect preparation: `surface_only`
+and `outcome_gap` produce a typed `validation_failed` without a prepared
+writeback. Previously vocabulary-valid but nonqualifying results reached
+refresh-state, raised a generic CLI error and left an in-progress prepared
+writeback that repeated the same refusal. The Host contract carries no typed
+blocker observation; ordinary CLI blocked settlement is unchanged. Historical
+cached results retain provider readback/recovery, including an unknown provider
+observation followed by a committed receipt, without relaunching the Host or
+repeating writeback. No settlement policy is added here.
 File/SQLite native CLI and actual DSH subprocess qualification cover rejection
 before task validation/writeback/spend/ACK, failed-result replay, and a corrected
 fresh Turn settling once. Boundary-length valid results retain their behavior.
diff --git a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md
index 61cdc5961e..09cf650df5 100644
--- a/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md
+++ b/docs/architecture/rfcs/agent-loop-effect-interpreter-v0.zh-CN.md
@@ -419,8 +419,13 @@ fixture。只读规划与可选 quota capture 默认值不变，原生执行 Tur
 现在也保留原值及 JSON 类型，只有省略元数据时才沿用稀疏默认值。此前声明的
 `surface_only`、不满足条件的 `outcome_gap` 或非法值都会变成 `outcome_progress`，
 绕过既有拒绝并记入 quota 扣费。共享 prompt 使用既有交付词汇。原生 File/SQLite
-测试保留合法规模历史，在任务验证前拒绝非法词汇，并证明独立任务检查通过后，
-不满足条件的结果仍在写回／扣费／ACK 前被拒绝；此处没有新增结算策略。
+测试保留合法规模历史。新材料结果也在任务验证或效果准备前复用既有结算规则：
+`surface_only` 与 `outcome_gap` 返回 typed `validation_failed`，不留下 prepared
+写回。此前词汇合法但不满足结算条件的结果会进入 refresh-state，抛出通用 CLI
+错误，留下反复遭遇同一拒绝的 in-progress prepared 写回。Host 合同不携带 typed
+blocker observation；普通 CLI 的阻塞结算保持原行为。历史缓存结果仍走 provider
+读回／恢复，覆盖 provider 返回 unknown 后读到 committed receipt，不重启 Host 或
+重复写回；此处没有新增结算策略。
 File/SQLite 原生 CLI
 和实际 DSH 子进程验证任务校验、写回、扣费、ACK 前的拒绝、失败结果重放，以及
 修正后的新 Turn 只结算一次。边界长度的合法结果保持原行为。这只证明结果传输
diff --git a/docs/reference/protocols/loopx-turn-v0.md b/docs/reference/protocols/loopx-turn-v0.md
index bf493a52b8..fe80dc7b4b 100644
--- a/docs/reference/protocols/loopx-turn-v0.md
+++ b/docs/reference/protocols/loopx-turn-v0.md
@@ -693,10 +693,17 @@ Explicit `delivery_batch_scale` and `delivery_outcome` also pass through to
 their existing validators; only omitted metadata uses the legacy
 `single_surface` / `outcome_progress` defaults. The adapter must not replace a
 declared result with progress or accept read-only scale aliases for new writes.
-Valid vocabulary does not establish a qualifying settlement: `surface_only`
-and an `outcome_gap` without its required typed blocker remain subject to the
-existing refresh-state refusal after task validation, before writeback, spend
-or ACK. Valid declared scales reach durable delivery history unchanged.
+Valid vocabulary does not establish a qualifying settlement. For new material
+Host results, the validator reuses the existing settlement rule before task
+validation or effect preparation. `surface_only` and `outcome_gap` return
+`validation_failed` at `host_result_contract`: this Host contract carries no
+typed blocker observation. The failed journal retains no prepared writeback;
+ordinary replay has no effects, and a corrected candidate needs a new Turn or
+an explicitly admitted failed-result retry. Historical cached results keep
+their provider readback and settlement recovery; a possibly committed effect
+cannot be discarded by this new-candidate check. Ordinary CLI typed-blocker
+settlement remains unchanged. Valid declared scales reach durable history
+unchanged.
 
 共享 dsh/可选云端适配器保留候选结果的完整文本，交给现有 Turn 结果校验器判定上限；
 不得截短分类、摘要、后续行动、建议或路径不变理由以使其通过，否则可能丢失末尾约束。
@@ -708,9 +715,13 @@ or ACK. Valid declared scales reach durable delivery history unchanged.
 明确提供的 `delivery_batch_scale` 与 `delivery_outcome` 也原样交给既有校验器；
 只有省略元数据时才使用旧 `single_surface` / `outcome_progress` 默认值。适配器
 不得将所声明结果替换为进展，也不得让只读 scale 别名用于新写入。词汇合法不代表
-满足结算条件：`surface_only` 及缺少规定 typed blocker 的 `outcome_gap` 仍会在
-任务验证后、写回／扣费／ACK 前被既有 refresh-state 拒绝。合法声明的规模原样
-进入持久交付历史。
+满足结算条件。新材料 Host 结果在任务校验或效果准备前复用既有结算规则：
+`surface_only` 与 `outcome_gap` 返回 `host_result_contract` 阶段的
+`validation_failed`，因为当前 Host 合同不携带 typed blocker observation。
+失败 journal 不留下 prepared 写回；普通重放无效果，修正候选须通过新 Turn 或明确
+准入的失败结果重试。历史缓存结果继续走 provider 读回和结算恢复，不能由新增候选
+校验丢弃可能已经提交的效果。普通 CLI 的 typed blocker 结算保持原行为。合法声明
+的规模原样进入持久历史。
 
 The existing TypeScript Vision normalizer owns packet interpretation and emits
 the canonical `goal_vision_replan_contract_v0` shape. Its current compatibility
diff --git a/loopx/control_plane/turn_driver/executor.py b/loopx/control_plane/turn_driver/executor.py
index b7d617d9e2..26bcc6e1f5 100644
--- a/loopx/control_plane/turn_driver/executor.py
+++ b/loopx/control_plane/turn_driver/executor.py
@@ -24,7 +24,10 @@
 )
 from ..goals.goal_vision import normalize_goal_vision_packet
 from ..work_items.delivery_batch_scale import require_delivery_batch_scale
-from ..work_items.delivery_outcome import require_delivery_outcome
+from ..work_items.delivery_outcome import (
+    explain_turn_scoped_settlement_gap,
+    require_delivery_outcome,
+)
 from . import subagent_execution_topology as subagent
 from .command_validation import (
     TaskValidator,
@@ -434,6 +437,16 @@ def validate_loopx_turn_host_result(
             normalized["delivery_outcome"] = require_delivery_outcome(
                 result.get("delivery_outcome")
             ).value
+            if not from_journal:
+                # This Host contract carries no typed blocker observation.
+                # Refuse new candidates before preparing an impossible write.
+                # Saved results retain provider readback/recovery: an effect
+                # may already have committed under an older producer.
+                settlement_gap = explain_turn_scoped_settlement_gap(
+                    normalized["delivery_outcome"], None,
+                )
+                if settlement_gap:
+                    errors.append(settlement_gap)
         except ValueError as exc:
             errors.append(str(exc))
 
diff --git a/tests/control_plane/test_host_result_semantics.py b/tests/control_plane/test_host_result_semantics.py
index bfcf49fd5e..186d38ff2b 100644
--- a/tests/control_plane/test_host_result_semantics.py
+++ b/tests/control_plane/test_host_result_semantics.py
@@ -12,19 +12,19 @@
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
-@pytest.mark.parametrize("candidate_fields,error,before_task_validation", [
+@pytest.mark.parametrize("candidate_fields,error", [
     ({"next_action": "Review the evidence. " * 65 + "Stop before publication; approval is still required."},
-     "next_action exceeds 1200 characters", True),
+     "next_action exceeds 1200 characters"),
     ({"next_action": {"action": "Continue", "constraint": "Stop before publication."}},
-     "next_action must be a string", True),
-    ({"next_action": False}, "next_action must be a string", True),
-    ({"delivery_batch_scale": "single_segment"}, "legacy alias 'single_segment' is read-only", True),
-    ({"delivery_outcome": False}, "delivery_outcome must be one of", True),
-    ({"delivery_outcome": "surface_only"}, "cannot settle a Turn", False),
-    ({"delivery_outcome": "outcome_gap"}, "blocked", False),
+     "next_action must be a string"),
+    ({"next_action": False}, "next_action must be a string"),
+    ({"delivery_batch_scale": "single_segment"}, "legacy alias 'single_segment' is read-only"),
+    ({"delivery_outcome": False}, "delivery_outcome must be one of"),
+    ({"delivery_outcome": "surface_only"}, "cannot settle a Turn"),
+    ({"delivery_outcome": "outcome_gap"}, "blocked"),
 ], ids=["overlimit", "object", "false", "legacy-scale", "invalid-outcome", "surface-only", "untyped-gap"])
 def test_native_turn_rejects_invalid_candidate_before_writeback(
-    tmp_path, monkeypatch, provider, candidate_fields, error, before_task_validation,
+    tmp_path, monkeypatch, provider, candidate_fields, error,
 ):
     isolate_sqlite_runtime(tmp_path, monkeypatch)
     f = _fixture_module()
@@ -66,24 +66,22 @@ def test_native_turn_rejects_invalid_candidate_before_writeback(
     code, failed = f._run_cli([*base, "--turn-instance-id", "invalid-result", "--execute"])
     assert code == 1 and not failed["ok"], failed
     assert error in json.dumps(failed)
-    if before_task_validation:
-        assert failed["status"] == "failed"
-        assert failed["validation_stage"] == "host_result_contract"
-        assert failed["effects"] == {
-            "host_invoked": True, "state_written": False,
-            "quota_spent": False, "scheduler_acknowledged": False,
-        }
-        resume_key = failed["resume_turn_key"]
-    else:
-        # Vocabulary-valid outcomes still face the existing native settlement
-        # rule. Its CLI error leaves the validated journal for recovery.
-        journal_path, = (runtime / "goals" / f.GOAL_ID / "turns").glob("*.json")
-        journal = json.loads(journal_path.read_text())
-        assert journal["host_result"]["delivery_outcome"] == candidate_fields["delivery_outcome"]
-        assert journal["completed_phases"] == ["host_execute", "typed_result", "validation"]
-        assert "writeback" not in journal and "quota_spend" not in journal
-        resume_key = journal["turn_key"]
-    assert (workspace / "validator-ran").exists() is not before_task_validation
+    assert failed["schema_version"] == "loopx_turn_execution_v0"
+    assert failed["status"] == "failed"
+    assert failed["result_kind"] == "validation_failed"
+    assert failed["validation_stage"] == "host_result_contract"
+    assert failed["effects"] == {
+        "host_invoked": True, "state_written": False,
+        "quota_spent": False, "scheduler_acknowledged": False,
+    }
+    resume_key = failed["resume_turn_key"]
+    journal_path, = (runtime / "goals" / f.GOAL_ID / "turns").glob("*.json")
+    journal = json.loads(journal_path.read_text())
+    assert journal["status"] == "failed"
+    assert journal["completed_phases"] == ["host_execute", "typed_result"]
+    assert not journal.get("effect_attempts")
+    assert "writeback" not in journal and "quota_spend" not in journal
+    assert not (workspace / "validator-ran").exists()
     assert f._quota_spend_count(runtime) == 0
     code, current = f._run_cli([*prefix, "todo", "list", "--goal-id", f.GOAL_ID])
     assert code == 0 and current["todos"] == listed["todos"]
@@ -96,16 +94,10 @@ def test_native_turn_rejects_invalid_candidate_before_writeback(
     (workspace / "candidate.json").write_text(json.dumps(candidate))
     code, repeated = f._run_cli([*base, "--resume-turn-key", resume_key, "--execute"])
     assert code == 1 and not repeated["ok"], repeated
-    if before_task_validation:
-        assert repeated["status"] == "failed" and not any(repeated["effects"].values())
-    else:
-        assert error in json.dumps(repeated)
-        journal = json.loads(journal_path.read_text())
-        assert journal["completed_phases"] == ["host_execute", "typed_result", "validation"]
-        assert "writeback" not in journal and "quota_spend" not in journal
+    assert repeated["status"] == "failed" and not any(repeated["effects"].values())
     assert f._quota_spend_count(runtime) == 0
     assert (workspace / "invocations.txt").read_text() == "host\n"
-    assert (workspace / "validator-ran").exists() is not before_task_validation
+    assert not (workspace / "validator-ran").exists()
     code, recovered = f._run_cli([*base, "--turn-instance-id", "corrected-result", "--execute"])
     assert code == 0 and recovered["status"] == "committed", recovered
     assert (workspace / "validator-ran").read_text() == "verified"
diff --git a/tests/test_dsh_goal_mode.py b/tests/test_dsh_goal_mode.py
index cc796e28e2..59e0ed2a03 100644
--- a/tests/test_dsh_goal_mode.py
+++ b/tests/test_dsh_goal_mode.py
@@ -464,8 +464,8 @@ def test_build_result_shapes_material_results_with_required_fields() -> None:
     ("delivery_batch_scale", "test_only", True),
     ("delivery_batch_scale", "multi_surface", True),
     ("delivery_batch_scale", "implementation", True),
-    ("delivery_outcome", "surface_only", True),
-    ("delivery_outcome", "outcome_gap", True),
+    ("delivery_outcome", "surface_only", False),
+    ("delivery_outcome", "outcome_gap", False),
     ("delivery_outcome", "primary_goal_outcome", True),
     ("delivery_batch_scale", "single_segment", False),
     ("delivery_batch_scale", "bounded_segment", False),
@@ -483,8 +483,8 @@ def test_explicit_delivery_metadata_reaches_its_existing_validator(kind, field,
     adapted = turn_host_adapter.build_result(request, {**base, field: value})
     assert type(adapted[field]) is type(value) and adapted[field] == value
     direct = {**turn_host_adapter.build_result(request, base), field: value}
-    # Legal outcome names are not proof that native settlement can accept
-    # them. This checks only the transport and existing vocabulary contract.
+    # Legal outcome names alone cannot settle a material Turn. The current
+    # Host result has no typed blocker observation, so outcome_gap is refused.
     for result in (adapted, direct):
         verdict = validate_loopx_turn_host_result(_turn_plan(request), result)
         assert verdict["ok"] is valid, verdict["errors"]
diff --git a/tests/test_loopx_turn_executor.py b/tests/test_loopx_turn_executor.py
index 0eed761ed1..c6fd88ce84 100644
--- a/tests/test_loopx_turn_executor.py
+++ b/tests/test_loopx_turn_executor.py
@@ -724,6 +724,23 @@ def test_host_result_requires_bounded_public_material_fields(field: str) -> None
     assert "unsupported host result fields" in " ".join(validation["errors"])
 
 
+@pytest.mark.parametrize("kind", [
+    "validated_progress", "validated_completion", "repair_required", "replan_required",
+])
+@pytest.mark.parametrize("outcome", ["surface_only", "outcome_gap"])
+def test_new_material_host_result_requires_qualifying_settlement(kind: str, outcome: str) -> None:
+    plan = _plan()
+    result = {**_host_result(plan), "result_kind": kind, "delivery_outcome": outcome}
+
+    validation = validate_loopx_turn_host_result(
+        plan, result, completion_lifecycle_configured=True,
+    )
+
+    assert validation["ok"] is False
+    assert outcome in " ".join(validation["errors"])
+    assert validation["result"]["delivery_outcome"] == outcome
+
+
 @pytest.mark.parametrize("packet, expected_error", [
     (None, "cached agent_vision must be a JSON object"),
     ([], "cached agent_vision must be a JSON object"),
@@ -1869,9 +1886,11 @@ def test_run_once_commits_once_and_replays_without_duplicate_effects(
     assert resumed["route"] == stored["plan"]["route"]
 
 
+@pytest.mark.parametrize("saved_outcome", ["outcome_progress", "surface_only", "outcome_gap"])
 def test_provider_can_commit_before_its_journal_checkpoint(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
+    saved_outcome: str,
 ) -> None:
     plan = _plan()
     calls = {"writeback": 0, "spend": 0, "scheduler": 0}
@@ -1927,6 +1946,33 @@ def fail_before_writeback_checkpoint(
     assert prepared["effect_ref"] in provider_records
 
     monkeypatch.setattr(turn_executor, "_write_journal", write_journal)
+    # Model metadata accepted by a historical producer. A new candidate gate
+    # cannot erase this unresolved provider commit or replace its saved result.
+    journal["host_result"]["delivery_outcome"] = saved_outcome
+    write_journal(
+        turn_journal_path(
+            tmp_path / "runtime", goal_id="fixture-goal", turn_key=journal["turn_key"],
+        ),
+        journal,
+    )
+    unresolved = run_loopx_turn_once(
+        plan,
+        host_runner=lambda _request: pytest.fail("host must not run during recovery"),
+        project=tmp_path,
+        runtime_root=tmp_path / "runtime",
+        goal_id="fixture-goal",
+        timeout_seconds=5,
+        execute=True,
+        task_validator=_passing_validator,
+        writeback=writeback,
+        writeback_resolver=lambda _effect_ref: {"kind": "unknown", "reason": "provider unavailable"},
+        spend=spend,
+        scheduler=scheduler,
+    )
+    assert unresolved["result_kind"] == "writeback_failed"
+    assert unresolved["receipt"]["failed_phase"] == "durable_writeback"
+    assert _journal(tmp_path / "runtime")["effect_attempts"]["durable_writeback"] == prepared
+    assert calls == {"writeback": 1, "spend": 0, "scheduler": 0}
     resumed = run_loopx_turn_once(
         plan,
         host_runner=lambda _request: pytest.fail("host must not run during recovery"),
@@ -1941,6 +1987,7 @@ def fail_before_writeback_checkpoint(
             "kind": "committed",
             "payload": provider_records[effect_ref],
         },
+        retry_failed=True,
         spend=spend,
         scheduler=scheduler,
     )
@@ -1948,6 +1995,7 @@ def fail_before_writeback_checkpoint(
     assert resumed["status"] == "committed"
     assert calls == {"writeback": 1, "spend": 1, "scheduler": 1}
     assert "effect_attempts" not in _journal(tmp_path / "runtime")
+    assert _journal(tmp_path / "runtime")["host_result"]["delivery_outcome"] == saved_outcome
 
 
 def test_recreation_waits_for_admitted_provider_checkpoint(