From 8bcef3766672725bd9afc429fef8cc55c0d48ff6 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 9 Nov 2026 04:47:22 +0800 Subject: [PATCH] Use the complete work source for Turn preparation Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../capabilities/native_chat/steward_turn.py | 4 +- loopx/cli_commands/turn.py | 13 ++---- loopx/control_plane/quota/turn_envelope.py | 31 +++---------- .../test_complete_host_request_cutover.py | 9 +++- tests/test_native_steward_turn_host.py | 43 +++++++++++++++++-- 5 files changed, 60 insertions(+), 40 deletions(-) diff --git a/loopx/capabilities/native_chat/steward_turn.py b/loopx/capabilities/native_chat/steward_turn.py index 8839c731cc..f0242109c1 100644 --- a/loopx/capabilities/native_chat/steward_turn.py +++ b/loopx/capabilities/native_chat/steward_turn.py @@ -700,7 +700,7 @@ def session_binding(envelope: Mapping[str, Any]) -> dict[str, Any]: return {"schema_version": "loopx_turn_session_binding_v0", "goal_id": actor["goal_id"], "agent_id": actor["agent_id"]} - def prepare_native_task(decision: Mapping[str, Any], envelope: Mapping[str, Any]) -> str | None: + def prepare_native_task(decision: Mapping[str, Any]) -> str | None: nonlocal managed_lease, managed_lease_request, preparation_status from ...control_plane.turn_driver.codex_execution_preparation import compact_execution_preparation_request @@ -715,7 +715,7 @@ def prepare_native_task(decision: Mapping[str, Any], envelope: Mapping[str, Any] current() if managed_turn is not None: try: - managed_turn.prepare_validation(request, {"turn_envelope": envelope}) + managed_turn.prepare_validation(request, {"work_contract": decision}) except ManagedTaskValidationUnavailable: managed_turn.validation_unavailable = True preparation_status = "stopped" diff --git a/loopx/cli_commands/turn.py b/loopx/cli_commands/turn.py index bd051e4d29..734a698ad5 100644 --- a/loopx/cli_commands/turn.py +++ b/loopx/cli_commands/turn.py @@ -12,7 +12,7 @@ ) from ..control_plane.operator_provider import operator_provider_environ from ..control_plane.quota.effective_action import EffectiveAction -from ..control_plane.quota.turn_envelope import build_turn_envelope, compose_turn_decision +from ..control_plane.quota.turn_envelope import compose_turn_decision from ..control_plane.runtime.status_projection_cache import ( resolve_status_projection_cache_runtime_root, ) @@ -24,8 +24,8 @@ from .turn_inspection import handle_turn_journal_inspection from .turn_registration import register_turn_commands as register_turn_commands from .turn_rendering import ( - project_turn_plan_inspection, build_turn_error_payload, + project_turn_plan_inspection, render_loopx_turn_execution_markdown as _render_loopx_turn_execution_markdown, render_loopx_turn_plan_markdown as _render_loopx_turn_plan_markdown, ) @@ -50,7 +50,7 @@ def handle_turn_command( print_payload: PrintPayload, native_host_runner: Callable[..., dict[str, Any]] | None = None, native_session_binding_resolver: Callable[[Mapping[str, Any]], dict[str, Any]] | None = None, - native_execution_preparer: Callable[[Mapping[str, Any], Mapping[str, Any]], str | None] | None = None, + native_execution_preparer: Callable[[Mapping[str, Any]], str | None] | None = None, native_completion_lease_resolver: Callable[[], Mapping[str, Any]] | None = None, native_task_validator: Callable[[Mapping[str, Any], Mapping[str, Any]], Mapping[str, Any]] | None = None, ) -> int | None: @@ -259,12 +259,8 @@ def read_decision_owner(): scheduler_context = decision_owner.scheduler_execution_context decision = compose_turn_decision(decision_owner.resolve(), scheduler_execution_context=scheduler_context) resume_requested, session_binding = resolve_turn_resume_session_binding(args) - turn_envelope = build_turn_envelope( - decision, - scheduler_execution_context=scheduler_context, - ) if native_execution_preparer is not None and args.turn_command == "run-once" and args.execute: - prepared_todo = native_execution_preparer(decision, turn_envelope) + prepared_todo = native_execution_preparer(decision) if prepared_todo is not None: # Host preparation changes canonical lease state. Plan and # settle against a fresh read of the same Turn and Todo. @@ -277,7 +273,6 @@ def read_decision_owner(): or (decision.get("selected_todo") or {}).get("todo_id") != prepared_todo or decision.get("normal_delivery_allowed") is not True): raise ValueError("native task preparation did not make the same Todo ready") - turn_envelope = build_turn_envelope(decision, scheduler_execution_context=scheduler_context) # Operation-equipped app-server sessions retain their Todo-bound # approval/handoff contract; this option controls exec conversation reuse. session_scope = (args.session_scope if args.host in {"codex-cli", "codex-app-server"} diff --git a/loopx/control_plane/quota/turn_envelope.py b/loopx/control_plane/quota/turn_envelope.py index 81f0f520d1..9863ddee35 100644 --- a/loopx/control_plane/quota/turn_envelope.py +++ b/loopx/control_plane/quota/turn_envelope.py @@ -5,7 +5,6 @@ from collections.abc import Mapping from copy import deepcopy import json -from dataclasses import dataclass from typing import Any, cast from ..effect_runtime import effect_runtime_projection_result, source_number_tokens @@ -29,27 +28,6 @@ PLANNING_HORIZON_DETAIL_REFS_REF = "$.detail_ref" -@dataclass(frozen=True, init=False, repr=False, slots=True) -class TurnControllerInput: - """Local producer result; never an agent wire or a retained authority cache. - - The envelope and its typed route originate in one projection. Store the - envelope as immutable bytes so a caller cannot mutate it after routing; - each consumer receives its own decoded object. Ordinary wire dictionaries - cannot supply this qualification through an extra JSON field. - """ - - envelope_json: str - route: str - - def __init__(self) -> None: - raise TypeError("Use build_turn_controller_input with the original quota source") - - def envelope(self) -> dict[str, Any]: - envelope: dict[str, Any] = json.loads(self.envelope_json) - return envelope - - def _apply_context_projection_changes( source: dict[str, Any], result: dict[str, Any], *, base: dict[str, Any] | None = None, ) -> dict[str, Any]: @@ -103,7 +81,7 @@ def _turn_envelope_result( payload = params.get("payload") complete = isinstance(payload, Mapping) and payload.get("schema_version") == "loopx_turn_envelope_v1" staged_replan = ( - operation in {"build", "build_controller_input", "quota_signature", "replan_cli_view", "bind_capture"} + operation in {"build", "quota_signature", "replan_cli_view", "bind_capture"} and isinstance(payload, Mapping) and ( (operation == "replan_cli_view" and params.get("source_edits") is True) @@ -194,8 +172,11 @@ def complete_turn_host_source( header = ({"complete_host_request": {key: request[key] for key in ( "schema_version", "turn_key", "source_decision_hash", "detail_ref", ) if key in request}} if request.get("schema_version") == COMPLETE_TURN_HOST_REQUEST_SCHEMA else {}) - return _turn_envelope_result(operation, payload=request["work_contract"], - envelope=_wire_mapping(request.get("turn_envelope")), **header, + # Identity-only current callers need no executor adapter. Keep an explicit + # adapter, even a damaged one, on the typed owner's full Host validation. + adapter = ({"envelope": _wire_mapping(request.get("turn_envelope"))} + if "turn_envelope" in request or operation != "complete_host_lineage" else {}) + return _turn_envelope_result(operation, payload=request["work_contract"], **adapter, **header, **({"recorded_capability_result": True} if recorded_capability_result else {}), **({"include_session_lineage": True} if include_session_lineage else {})) diff --git a/tests/control_plane/test_complete_host_request_cutover.py b/tests/control_plane/test_complete_host_request_cutover.py index 2ddcd1d637..e6aa82e7c5 100644 --- a/tests/control_plane/test_complete_host_request_cutover.py +++ b/tests/control_plane/test_complete_host_request_cutover.py @@ -43,7 +43,14 @@ def test_complete_source_can_plan_without_producing_a_legacy_adapter(complete_ho request = build_loopx_turn_host_request(plan) assert request["schema_version"] == "loopx_turn_host_request_v1" assert request["work_contract"] == contract and "turn_envelope" not in request - assert turn_host_lineage(request) == turn_host_lineage(legacy) + expected_lineage = turn_host_lineage(legacy) + assert turn_host_lineage(request) == expected_lineage + assert turn_host_lineage({"work_contract": contract}) == expected_lineage + for invalid in ({"work_contract": {**contract, "goal_id": "other-goal"}}, + {"work_contract": contract, "turn_envelope": None}, + {"work_contract": contract, "turn_envelope": {}}): + with pytest.raises((ValueError, RuntimeError)): + turn_host_lineage(invalid) assert contract == before diff --git a/tests/test_native_steward_turn_host.py b/tests/test_native_steward_turn_host.py index 28de72cd0b..44f8508b9a 100644 --- a/tests/test_native_steward_turn_host.py +++ b/tests/test_native_steward_turn_host.py @@ -167,11 +167,34 @@ def test_managed_identity_rejections_do_not_create_or_resume_a_host(steward, tmp @pytest.mark.parametrize("provider", ["file", "sqlite"]) -def test_local_managed_session_reuses_native_turn_and_declared_validation(steward, tmp_path, monkeypatch, provider): +@pytest.mark.parametrize("explicit_selection", [False, True], ids=["claim", "ready"]) +def test_local_managed_session_reuses_native_turn_and_declared_validation( + steward, tmp_path, monkeypatch, provider, explicit_selection, +): service, session, actor, todo_id, observation, capture = managed_fixture(steward, tmp_path, monkeypatch, provider) import loopx.cli_commands.turn as turn_owner + from loopx.capabilities.native_chat.steward_managed_mode import ManagedSessionTurn + from loopx.control_plane.quota import turn_envelope + original_plan = turn_owner.build_loopx_turn_plan + original_validation = ManagedSessionTurn.prepare_validation + evaluate = turn_envelope._turn_envelope_result prepared_sources = [] + validation_sources = [] + legacy_builds = [] + + def observe_projection(operation, **kwargs): + if operation == "build": + legacy_builds.append(operation) + return evaluate(operation, **kwargs) + + def checked_validation(self, request, native_request, **kwargs): + validation_sources.append((request["turn_instance_id"], native_request)) + return original_validation(self, request, native_request, **kwargs) + + monkeypatch.setattr(turn_envelope, "_turn_envelope_result", observe_projection) + monkeypatch.setattr(ManagedSessionTurn, "prepare_validation", checked_validation) + def checked_plan(source, *args, **kwargs): preparation = source["interaction_contract"]["agent_channel"]["execution_preparation"] assert preparation["state"] == "ready" @@ -187,7 +210,8 @@ def checked_plan(source, *args, **kwargs): before_history = store.messages(sid) assert len(before_history) == 2 assert service.inspect(sid)["available"] - accepted, created = service.submit(sid, agent_id=actor["agent_id"], operation_id="local-native-1", todo_id=todo_id) + requested_todo = todo_id if explicit_selection else None + accepted, created = service.submit(sid, agent_id=actor["agent_id"], operation_id="local-native-1", todo_id=requested_todo) assert created completed = finish(runtime, accepted) assert completed["status"] == "completed", completed @@ -196,6 +220,19 @@ def checked_plan(source, *args, **kwargs): assert result["effects"]["host_invoked"] and result["effects"]["quota_spent"] assert result["validation"]["status"] == "passed" assert prepared_sources == ["loopx_turn_envelope_v1"] + assert legacy_builds == [] + expected_states = ["ready"] if explicit_selection else ["claim_required", "ready"] + assert len(validation_sources) == len(expected_states) + for (turn_id, native_request), state in zip(validation_sources, expected_states, strict=True): + assert turn_id == accepted["loopx_request"]["turn_instance_id"] + assert "turn_envelope" not in native_request + source = native_request["work_contract"] + assert source["schema_version"] == "loopx_turn_envelope_v1" + assert source["selected_todo"]["todo_id"] == todo_id + assert source["agent_identity"]["agent_id"] == actor["agent_id"] + assert source["execution_preparation"]["state"] == state + if "detail_ref" in native_request: + assert native_request["detail_ref"]["captured_decision"]["turn_instance_id"] == turn_id assert observation.read_text() == "ready" assert store.load_session(sid)["upstream_thread_id"] == thread assert {row["session_id"] for row in store.list_sessions()} == {row["session_id"] for row in before_sessions} @@ -205,7 +242,7 @@ def checked_plan(source, *args, **kwargs): assert all(row.get("params", {}).get("sandbox", "read-only") == "read-only" for row in calls) assert not [row for row in calls if row["method"] in {"thread/start", "thread/goal/set", "thread/inject_items"}] after_calls = capture.read_bytes() - replay, created = service.submit(sid, agent_id=actor["agent_id"], operation_id="local-native-1", todo_id=todo_id) + replay, created = service.submit(sid, agent_id=actor["agent_id"], operation_id="local-native-1", todo_id=requested_todo) assert not created and replay["turn_id"] == accepted["turn_id"] and capture.read_bytes() == after_calls assert service.inspect(sid, operation_id="local-native-1")["turn"]["turn_id"] == accepted["turn_id"] with pytest.raises(ValueError, match="identity conflict"):