From a109856370b786a009f2f436ea556fdf309bde88 Mon Sep 17 00:00:00 2001 From: hyk <4408344+hhyykk@users.noreply.github.com> Date: Thu, 8 Oct 2026 16:40:05 +0800 Subject: [PATCH 1/2] fix(turn): preserve managed host material replan packets Signed-off-by: hyk <4408344+hhyykk@users.noreply.github.com> --- loopx/control_plane/turn_driver/executor.py | 16 ++- .../turn_driver/host_candidate.py | 41 ++++-- tests/test_dsh_goal_mode.py | 133 ++++++++++++++++++ tests/test_loopx_turn_driver.py | 93 ++++++++++++ 4 files changed, 270 insertions(+), 13 deletions(-) diff --git a/loopx/control_plane/turn_driver/executor.py b/loopx/control_plane/turn_driver/executor.py index a7d93fd263..0643d2a61b 100644 --- a/loopx/control_plane/turn_driver/executor.py +++ b/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) diff --git a/loopx/control_plane/turn_driver/host_candidate.py b/loopx/control_plane/turn_driver/host_candidate.py index a72e7de3e8..f7ce5a6f52 100644 --- a/loopx/control_plane/turn_driver/host_candidate.py +++ b/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 diff --git a/tests/test_dsh_goal_mode.py b/tests/test_dsh_goal_mode.py index 684b38dc30..cb03b4d6f1 100644 --- a/tests/test_dsh_goal_mode.py +++ b/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: diff --git a/tests/test_loopx_turn_driver.py b/tests/test_loopx_turn_driver.py index 6f44e65068..7c93bdff98 100644 --- a/tests/test_loopx_turn_driver.py +++ b/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."], From eac312f26acfc0bcc79faf6aa4db463e76d1aa60 Mon Sep 17 00:00:00 2001 From: hyk <4408344+hhyykk@users.noreply.github.com> Date: Thu, 8 Oct 2026 16:40:05 +0800 Subject: [PATCH 2/2] docs(turn): clarify managed path delta compatibility Signed-off-by: hyk <4408344+hhyykk@users.noreply.github.com> --- docs/reference/protocols/loopx-turn-v0.md | 26 +++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/docs/reference/protocols/loopx-turn-v0.md b/docs/reference/protocols/loopx-turn-v0.md index 941c755673..88e06354bd 100644 --- a/docs/reference/protocols/loopx-turn-v0.md +++ b/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