Remove redundant legacy envelope builds from Turn CLI

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"

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

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 {}))
 

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
 
 

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