fix(release): stabilize live model qualification

loopx/control_plane/testing/doubao_model_behavior_actor.py

@@ -461,10 +461,14 @@ def _onboarding_decision_instruction(phase: str) -> str:
 - agent_id: copy top-level agent_id, falling back to
   host_loop_activation.agent_id or command_pack.host_loop_activation.agent_id;
   otherwise null.
-- action_command_ids: in this exact order, include each key whose value is a
-  non-empty string in top-level commands or command_pack.commands:
-  goal_start_connect_if_needed, goal_start_refresh_state,
-  goal_start_host_loop_activation, goal_start_quota_should_run.
+- action_command_ids must be a JSON array of command key names, never a
+  key/value object or command text. Start with []. Inspect top-level commands
+  when it is present; otherwise inspect command_pack.commands. Consider only
+  these four keys, in this exact order: goal_start_connect_if_needed,
+  goal_start_refresh_state, goal_start_host_loop_activation,
+  goal_start_quota_should_run. Include a key only when its value is a non-empty
+  string. Never add any other command key, even when another command looks
+  useful for the selected route.
 - host_loop_activation_available: true when top-level host_loop_activation or
   command_pack.host_loop_activation is a non-empty object.
 - host_loop_activation_after_todo_write: copy

loopx/control_plane/testing/replan_vision_closeout_behavior.py

@@ -33,7 +33,19 @@ def __init__(self, code: str, detail: str = "") -> None:
 VISION_HOST_INSTRUCTION = (
     "You are an agent working in an isolated LoopX project. Follow the heartbeat "
     "and current control-plane packet. Use the shell normally to inspect evidence, "
-    "author your decision and execute the real CLI writeback and settlement."
+    "author your decision and execute the real CLI writeback and settlement. "
+    "The ordered interaction_contract.cli_channel.next_cli_actions are templates, "
+    "not literal commands: first read the named source, start the decision file "
+    "from vision_authoring.minimal_example. Add path_delta.outcome, "
+    "prior_assumption, observed_reality, one retained/changed/stopped item and "
+    "observed evidence_refs. Keep every prose string under 80 characters; cite "
+    "the source instead of pasting it. Omit other optional fields unless the "
+    "evidence requires them, and check the projected vision_authoring limits "
+    "before the first refresh-state. Replace the first "
+    "action's path placeholder with your file path. Correct a failed action "
+    "using its error instead of repeating it unchanged. After a successful "
+    "writeback, execute the returned settlement_owed.command once in the same "
+    "turn and verify the settlement readback."
 )
 
 VISION_EXEC_TOOL_DESCRIPTION = (
@@ -107,6 +119,8 @@ def observe_shell_evidence(output: str, state: _QualificationState) -> None:
 def _rows(state: _QualificationState) -> list[dict[str, Any]]:
     goal_id = str((state.quota_packet or {})["goal_id"])
     index = state.fixture.runtime_root / "goals" / goal_id / "runs" / "index.jsonl"
+    if not index.is_file():
+        return []
     return [json.loads(line) for line in index.read_text(encoding="utf-8").splitlines() if line]
 
 
@@ -142,7 +156,13 @@ def dispatch_vision_closeout(
         if not refs.intersection(observed_refs):
             raise VisionHostAdmissionRejected("vision_closeout_evidence_not_observed", "No refresh was executed: path_delta.evidence_refs must identify the source evidence actually read. Accepted references: " + json.dumps(sorted(observed_refs)))
         output = execute(command, fixture=state.fixture, turn_instance_id=state.turn_instance_id)
-        row = _rows(state)[-1]
+        rows = _rows(state)
+        if not rows:
+            raise VisionHostAdmissionRejected(
+                "vision_closeout_durable_writeback_missing",
+                "No run receipt was written; revise the vision decision and retry.",
+            )
+        row = rows[-1]
         semantic = dict(row.get("autonomous_replan_ack") or {}).get("semantic_delta") or {}
         checkpoint = dict(row.get("vision_checkpoint") or {})
         identity = dict(row.get("settlement_identity") or {})

scripts/qualify-doubao-model-behavior-live.py

@@ -4,9 +4,11 @@
 import argparse
 import json
 import sys
+from collections.abc import Mapping
 from hashlib import sha256
 from pathlib import Path
 from tempfile import TemporaryDirectory
+from typing import Any
 
 REPO_ROOT = Path(__file__).resolve().parents[1]
 repo_root_text = str(REPO_ROOT)
@@ -52,6 +54,7 @@ def _parser() -> argparse.ArgumentParser:
     )
     parser.add_argument("--qualification-id", required=True)
     parser.add_argument("--timeout-seconds", type=float, default=90.0)
+    parser.add_argument("--required-vision-timeout-seconds", type=float, default=180.0)
     parser.add_argument("--repo-root", type=Path, default=Path.cwd())
     return parser
 
@@ -66,11 +69,17 @@ def main() -> int:
     provider_call_count = 0
     provider_models: set[str] = set()
 
-    def counted_transport(**kwargs):
+    def counted_transport(
+        *, endpoint: str, headers: Mapping[str, str], body: bytes,
+        timeout_seconds: float,
+    ) -> Mapping[str, Any]:
         nonlocal provider_call_count
         provider_call_count += 1
-        provider_models.add(json.loads(kwargs["body"])["model"])
-        return _direct_ark_transport(**kwargs)
+        provider_models.add(json.loads(body)["model"])
+        return _direct_ark_transport(
+            endpoint=endpoint, headers=headers, body=body,
+            timeout_seconds=timeout_seconds,
+        )
 
     turn_actor = DoubaoModelBehaviorActor.from_environment(
         timeout_seconds=args.timeout_seconds, transport=counted_transport
@@ -82,7 +91,8 @@ def counted_transport(**kwargs):
         timeout_seconds=args.timeout_seconds, transport=counted_transport
     )
     replan_semantic_action_actor = DoubaoReplanSemanticActionBehaviorActor.from_environment(
-        timeout_seconds=args.timeout_seconds, transport=counted_transport
+        timeout_seconds=args.required_vision_timeout_seconds,
+        transport=counted_transport,
     )
     scoped_gate_successor_actor = (
         DoubaoScopedGateSuccessorToolBehaviorActor.from_environment(

tests/control_plane/test_release_commit_qualification.py

@@ -3,10 +3,12 @@
 import copy
 import json
 import os
+import runpy
 import subprocess
 import sys
 from datetime import datetime, timezone
 from pathlib import Path
+from types import SimpleNamespace
 
 import pytest
 
@@ -327,3 +329,45 @@ def test_live_doubao_script_prefers_candidate_checkout_over_pythonpath(
     assert result.returncode == 0, result.stderr
     assert "Run the actual-default behavior portfolio" in result.stdout
     assert "loaded shadow loopx" not in result.stderr
+
+
+@pytest.mark.parametrize(
+    ("extra_args", "ordinary_timeout", "vision_timeout"),
+    [([], 90.0, 180.0), (["--timeout-seconds", "12", "--required-vision-timeout-seconds", "34"], 12.0, 34.0)],
+)
+def test_live_doubao_timeout_extension_is_scoped_to_required_vision(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+    extra_args: list[str],
+    ordinary_timeout: float,
+    vision_timeout: float,
+) -> None:
+    script = runpy.run_path(str(REPO_ROOT / "scripts" / "qualify-doubao-model-behavior-live.py"))
+    main = script["main"]
+    globals_ = main.__globals__
+    observed: dict[str, float] = {}
+    actor_names = (
+        "DoubaoModelBehaviorActor",
+        "DoubaoOnboardingModelBehaviorActor",
+        "DoubaoSelectedTodoToolBehaviorActor",
+        "DoubaoReplanSemanticActionBehaviorActor",
+        "DoubaoScopedGateSuccessorToolBehaviorActor",
+        "DoubaoCapabilityMonitorRepairToolBehaviorActor",
+        "DoubaoTerminalSettlementToolBehaviorActor",
+    )
+    for name in actor_names:
+        def make_actor(actor_name: str) -> SimpleNamespace:
+            def from_environment(**kwargs: object) -> object:
+                observed[actor_name] = float(kwargs["timeout_seconds"])
+                return object()
+            return SimpleNamespace(from_environment=from_environment)
+        monkeypatch.setitem(globals_, name, make_actor(name))
+    monkeypatch.setitem(globals_, "collect_release_source_identity", lambda _root: {"git_dirty": False})
+    monkeypatch.setitem(globals_, "build_actual_default_model_behavior_scenario_inputs", lambda _root: ({}, {}))
+    monkeypatch.setitem(globals_, "run_actual_default_model_behavior_portfolio", lambda *args, **kwargs: {"qualification_passed": True})
+    monkeypatch.setattr(sys, "argv", ["qualify-doubao-model-behavior-live.py", "--qualification-id", "timeout-scope", *extra_args])
+
+    assert main() == 0
+    assert json.loads(capsys.readouterr().out)["qualification_passed"] is True
+    assert observed["DoubaoReplanSemanticActionBehaviorActor"] == vision_timeout
+    assert all(observed[name] == ordinary_timeout for name in actor_names if name != "DoubaoReplanSemanticActionBehaviorActor")

tests/control_plane/test_replan_vision_closeout_receipt.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+import json
+from types import SimpleNamespace
+
+import pytest
+
+from loopx.control_plane.testing.replan_vision_closeout_behavior import (
+    VisionHostAdmissionRejected,
+    dispatch_vision_closeout,
+)
+
+
+def test_successful_cli_output_without_durable_receipt_is_correctable(tmp_path) -> None:
+    project = tmp_path / "project"
+    project.mkdir()
+    source = project / "work.json"
+    source.write_text("{}", encoding="utf-8")
+    frontier = project / "frontier.json"
+    frontier.write_text(
+        json.dumps({"uncovered": [{"source_ref": "work.json", "evidence_id": "evidence-a"}]}),
+        encoding="utf-8",
+    )
+    (project / "vision.json").write_text(
+        json.dumps({"path_delta": {"evidence_refs": ["evidence-a"]}}),
+        encoding="utf-8",
+    )
+    fixture = SimpleNamespace(
+        project_root=project,
+        runtime_root=tmp_path / "runtime",
+        frontier_target=frontier,
+        work_source_target=source,
+    )
+    state = SimpleNamespace(
+        fixture=fixture,
+        turn_instance_id="turn-a",
+        work_source_read=True,
+        quota_packet={
+            "goal_id": "goal-a",
+            "interaction_contract": {
+                "cli_channel": {
+                    "replan_settlement_contract": {
+                        "settlement_binding": {"cli_argument": "--binding", "id": "obligation-a"}
+                    }
+                }
+            },
+        },
+    )
+    executed = []
+
+    def execute(command, **_kwargs):
+        executed.append(command)
+        return "{}"
+
+    with pytest.raises(
+        VisionHostAdmissionRejected, match="vision_closeout_durable_writeback_missing"
+    ):
+        dispatch_vision_closeout(
+            "loopx refresh-state --binding obligation-a --turn-instance-id turn-a "
+            "--agent-vision-json vision.json",
+            state,
+            execute=execute,
+        )
+    assert len(executed) == 1

tests/control_plane/test_required_vision_closeout_behavior.py

@@ -14,7 +14,7 @@
 from loopx.control_plane.testing.replan_semantic_action_behavior import (
     DoubaoReplanSemanticActionBehaviorActor, _build_fixture,
 )
-from loopx.control_plane.testing import vision_shell_host
+from loopx.control_plane.testing import replan_semantic_action_behavior, vision_shell_host
 from loopx.control_plane.testing.vision_shell_host import VisionShellHost, shell_isolation_available
 
 pytestmark = pytest.mark.skipif(not shell_isolation_available(), reason="Native shell needs sandbox-exec or bubblewrap")
@@ -123,6 +123,40 @@ def correct(request: Mapping[str, Any]) -> ScriptedExecToolAction:
     assert result["vision_closeout"]["spend_count"] == 1
 
 
+def test_missing_durable_receipt_reaches_shell_and_can_be_retried(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    fixture = _build_fixture(tmp_path / "oracle", required_vision=True)
+    execute = replan_semantic_action_behavior._execute_loopx
+    intercepted = False
+
+    def first_refresh_without_receipt(command: str, **kwargs: Any) -> str:
+        nonlocal intercepted
+        if "refresh-state" in command and not intercepted:
+            intercepted = True
+            return '{"ok": true}'
+        return execute(command, **kwargs)
+
+    def retry_after_error(request: Mapping[str, Any]) -> ScriptedExecToolAction:
+        feedback = json.loads(request["messages"][-1]["content"])
+        assert feedback["exit_code"] != 0
+        assert "vision_closeout_durable_writeback_missing" in feedback["output"]
+        return projected_refresh(request)
+
+    monkeypatch.setattr(replan_semantic_action_behavior, "_execute_loopx", first_refresh_without_receipt)
+    result = _qualify(tmp_path, [
+        ScriptedExecToolAction(fixture.quota_guard_command),
+        ScriptedExecToolAction("cat fixture/permission-config.json"),
+        vision_patch_action,
+        projected_refresh,
+        retry_after_error,
+        projected_spend,
+    ])
+    assert intercepted is True
+    assert result["qualification_passed"] is True
+    assert result["vision_closeout"]["spend_count"] == 1
+
+
 @pytest.mark.parametrize("invalid", ["no_source", "wrong_turn", "unread_reference", "no_spend"])
 def test_native_host_does_not_qualify_unproven_closeout(tmp_path: Path, invalid: str) -> None:
     fixture = _build_fixture(tmp_path / "oracle", required_vision=True)