fix(settlement): preserve producing workspace in combined refresh

docs/reference/protocols/turn-envelope-v0.md

@@ -258,6 +258,10 @@ guard, changes no binding and performs no Todo completion. Failed validation,
 unsatisfied checkpoints, dry runs and no-spend closeout cannot debit quota.
 An ambiguous response is reconciled through the original receipts; exact retry
 cannot debit twice. Ordinary refresh without the option retains its behavior.
+When `--delivery-workspace-path` names the producing worktree, the combined
+operation also runs the owed command there, using the originating CLI's
+implementation. The normal workspace and settlement guards still apply; a
+missing or incompatible workspace leaves the original owed command recoverable.
 
 An in-flight recommendation does not require keeping a completed Todo open.
 Validate its original declaration, use ordinary completion when satisfied, and
@@ -277,6 +281,9 @@ writeback 读取，避免重复交付；当前准入、选择与停止事实仍
 Todo 或 obligation/Turn 的 canonical 待结算命令并验证回执；不重新 guard、换绑或
 完成 Todo。验证失败、检查点未满足、dry-run 与无扣额收尾不能扣额；响应不确定时
 按原回执恢复，精确重试不会重复扣额。不传此选项保持原 refresh 行为。
+指定 `--delivery-workspace-path` 后，组合操作也在该交付工作区执行待结算命令，
+继续使用原调用方的 CLI 实现。原工作区和结算校验保持；目录缺失或不兼容时保留
+原待结算命令以便恢复。
 
 执行中续跑建议不要求把已完成的 Todo 保持开放。原声明验收满足后可正常完成，并
 在原 Turn 上按语义收尾写入有证据的 vision；未完成的阶段进展仍可保留开放状态。

loopx/cli_commands/project_lifecycle_refresh_state.py

@@ -946,4 +946,8 @@ def _settle_requested_refresh(payload, args, registry_path, goal_ref):
         runtime_root=resolve_runtime_root(load_registry(registry_path), args.runtime_root),
         goal_id=args.goal_id, agent_id=args.agent_id, todo_id=args.todo_id,
         replan_obligation_id=args.replan_obligation_id,
-        turn_instance_id=args.turn_instance_id, goal_ref=goal_ref)
+        turn_instance_id=args.turn_instance_id, goal_ref=goal_ref,
+        delivery_workspace_path=(
+            Path(args.delivery_workspace_path)
+            if args.delivery_workspace_path else None
+        ))

loopx/cli_commands/refresh_settlement.py

@@ -23,6 +23,7 @@ def settle_refreshed_turn(
     goal_id: str, agent_id: str, todo_id: str | None,
     replan_obligation_id: str | None, turn_instance_id: str,
     goal_ref: Mapping[str, str] | None,
+    delivery_workspace_path: Path | None = None,
 ) -> None:
     """Follow one owed step, then verify its exact receipts; never collect a guard.
 
@@ -45,6 +46,7 @@ def settle_refreshed_turn(
     error = None
     spent: dict[str, Any] = {}
     try:
+        workspace = delivery_workspace_path.expanduser().resolve() if delivery_workspace_path else None
         command = shlex.split(owed["command"])
         # Validate the owner-rendered argv through the actual CLI parser too:
         # a projector bug must not debit a different valid Turn before readback.
@@ -60,11 +62,19 @@ def settle_refreshed_turn(
                     ("goal_id", goal_id), ("agent_id", agent_id), ("todo_id", todo_id),
                     ("replan_obligation_id", replan_obligation_id), ("turn_instance_id", turn_instance_id)))):
             raise ValueError("owed quota command does not preserve the exact original binding")
-        # Execute this checkout's complete CLI, including native guards, hooks
-        # and rollout receipts. Do not resolve an ambient LoopX from PATH or
-        # call the lower-level debit without its receipt-producing adapter.
+        # Keep the validated producing cwd for accounting, but pin imports to
+        # this CLI's package: the delivered worktree may contain another LoopX.
+        # -P excludes the child cwd from Python's initial module search path.
+        # Retain the complete CLI guards, hooks and rollout receipts; neither
+        # ambient PATH nor a lower-level debit may replace that adapter.
+        entry = (
+            "import sys; sys.path.insert(0, sys.argv.pop(1)); "
+            "import runpy; runpy.run_module('loopx.cli', run_name='__main__')"
+        )
         result = subprocess.run(
-            [sys.executable, "-m", "loopx.cli", "--format", "json", *command[1:]],
+            [sys.executable, "-P", "-c", entry, str(Path(__file__).resolve().parents[2]),
+             "--format", "json", *command[1:]],
+            cwd=workspace,
             capture_output=True, text=True, encoding="utf-8", timeout=60, check=False,
         )
         value = json.loads(result.stdout)
@@ -73,7 +83,7 @@ def settle_refreshed_turn(
         spent = value
         if result.returncode or spent.get("ok") is not True:
             error = str(spent.get("error") or spent.get("reason") or "quota settlement CLI failed")
-    except (OSError, ValueError, TypeError, KeyError, subprocess.TimeoutExpired, SystemExit) as exc:
+    except (OSError, RuntimeError, ValueError, TypeError, KeyError, subprocess.TimeoutExpired, SystemExit) as exc:
         error = str(exc)
     try:
         readback = read_heartbeat_settlement(

tests/control_plane/test_refresh_settle_turn.py

@@ -15,6 +15,88 @@
 )
 
 
+def _repository_delivery(tmp_path, provider):
+    project, runtime, registry = _write_fixture(tmp_path)
+    state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md"
+    state.write_text(state.read_text().replace(
+        "action_kind=validate", "action_kind=validate task_repository=git:github.com/example/delivery"))
+    subprocess.run(["git", "init", "-q", str(project)], check=True)
+    subprocess.run(["git", "-C", str(project), "remote", "add", "origin",
+                    "https://github.com/example/delivery.git"], check=True)
+    subprocess.run(["git", "-C", str(project), "-c", "user.name=Fixture", "-c",
+                    "user.email=fixture@example.test", "commit", "--allow-empty", "-qm", "fixture"], check=True)
+    delivery = tmp_path / "producing workspace"
+    subprocess.run(["git", "-C", str(project), "worktree", "add", "--detach", str(delivery)],
+                   check=True, capture_output=True)
+    # The delivered project may itself contain a different LoopX implementation.
+    # Changing the accounting cwd must not execute that implementation.
+    trap = delivery / "loopx"
+    trap.mkdir()
+    (trap / "__init__.py").write_text("raise RuntimeError('wrong implementation selected')\n")
+    _initialize_fixture_authority(project, runtime, provider=provider)
+    return project, runtime, registry, delivery
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_combined_settlement_uses_producing_worktree_and_originating_cli(
+    tmp_path, monkeypatch, provider,
+):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    project, runtime, registry, _delivery = _repository_delivery(tmp_path, provider)
+    binding = ["--goal-id", GOAL_ID, "--agent-id", AGENT_ID,
+               "--todo-id", TODO_ID, "--turn-instance-id", TURN_ID]
+    rc, guard = _run_cli(registry, runtime, "quota", "should-run", "--codex-app", *binding, cwd=project)
+    assert rc == 0, guard
+    identity = guard["interaction_contract"]["cli_channel"]["settlement_plan"]["identity"]
+    refresh = ["refresh-state", *binding, "--classification", "verified_worktree",
+               "--delivery-batch-scale", "implementation", "--delivery-outcome", "outcome_progress",
+               "--delivery-boundary", "in_flight_continuation", "--no-global-sync", "--suppress-external-sinks",
+               "--delivery-workspace-path", "../producing workspace", "--settle-turn"]
+    for replay in (False, True):
+        rc, result = _run_cli(registry, runtime, *refresh, cwd=project)
+        assert rc == 0, result
+        assert result["settlement_progress"]["state"] == "settled"
+        assert result["settlement_identity"] == identity
+        assert result["delivery_workspace"]["workspace_kind"] == "independent_git_worktree"
+        assert _spend_run_count(runtime) == 1
+        assert result.get("idempotent_replay", False) is replay
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_wrong_or_missing_child_workspace_retains_owed_command_for_exact_recovery(
+    tmp_path, monkeypatch, provider,
+):
+    from loopx.cli_commands.refresh_settlement import settle_refreshed_turn
+
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    project, runtime, registry, delivery = _repository_delivery(tmp_path, provider)
+    binding = ["--goal-id", GOAL_ID, "--agent-id", AGENT_ID,
+               "--todo-id", TODO_ID, "--turn-instance-id", TURN_ID]
+    rc, guard = _run_cli(registry, runtime, "quota", "should-run", "--codex-app", *binding, cwd=project)
+    assert rc == 0, guard
+    refresh = ["refresh-state", *binding, "--classification", "verified_worktree",
+               "--delivery-batch-scale", "implementation", "--delivery-outcome", "outcome_progress",
+               "--delivery-boundary", "in_flight_continuation", "--no-global-sync", "--suppress-external-sinks",
+               "--delivery-workspace-path", str(delivery)]
+    rc, original = _run_cli(registry, runtime, *refresh, cwd=project)
+    assert rc == 0 and _spend_run_count(runtime) == 0, original
+    for workspace in (project, tmp_path / "missing"):
+        failed = copy.deepcopy(original)
+        settle_refreshed_turn(failed, registry_path=registry, runtime_root=runtime,
+            goal_id=GOAL_ID, agent_id=AGENT_ID, todo_id=TODO_ID,
+            replan_obligation_id=None, turn_instance_id=TURN_ID, goal_ref=None,
+            delivery_workspace_path=workspace)
+        assert failed["ok"] is False, failed
+        assert failed["settlement_progress"]["state"] == "spend_required"
+        assert failed["settlement_owed"]["command"] == original["settlement_owed"]["command"]
+        assert _spend_run_count(runtime) == 0
+    rc, recovered = _run_cli(registry, runtime, *refresh, "--settle-turn", cwd=project)
+    assert rc == 0, recovered
+    assert recovered["settlement_progress"]["state"] == "settled"
+    assert recovered["settlement_identity"] == original["settlement_identity"]
+    assert _spend_run_count(runtime) == 1
+
+
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
 @pytest.mark.parametrize("finish_todo", [False, True])
 def test_same_turn_progress_or_completion_and_replay_without_extra_guard(
@@ -88,6 +170,9 @@ def test_failed_writeback_dry_run_and_missing_binding_cannot_spend(tmp_path, mon
             "--no-global-sync", "--suppress-external-sinks", "--settle-turn"]
     rc, rejected = _run_cli(registry, runtime, *base, "--delivery-outcome", "surface_only")
     assert rc != 0 and _spend_run_count(runtime) == 0, rejected
+    rc, bad_path = _run_cli(registry, runtime, *base,
+        "--delivery-workspace-path", "~loopx-missing-fixture-user")
+    assert rc != 0 and bad_path["ok"] is False and _spend_run_count(runtime) == 0, bad_path
     rc, preview = _run_cli(registry, runtime, *base, "--delivery-boundary", "in_flight_continuation", "--dry-run")
     assert rc == 0 and preview["dry_run"] and _spend_run_count(runtime) == 0, preview
     # Invalid vision fails before writeback; no direct-debit bypass is allowed.