fix: preserve raw quota decisions for machine consumers

docs/reference/required-work-context.md

@@ -243,6 +243,25 @@ machine default, with off when absent. The existing worker environment's explici
 not enable delivery. Benchmark trials keep their existing `context_delivery=on`
 opt-in. Workers propagate the resolved trial choice, including default off, to
 both preflight and solver subprocesses without changing the caller's settings.
+
+Machine consumers that parse `should_run`, selected work or settlement identities
+must explicitly request raw output on internal `quota should-run` calls using
+`--context-delivery off` or the subprocess `LOOPX_CONTEXT_DELIVERY=off` override,
+and retain the exact registry, Goal, Agent and Turn. Enabled native output is a
+presentation/capture envelope, not a raw authority decision; a missing
+`should_run` in that envelope is not a denial. MCP settlement and Claude status
+readers use the existing `LOOPX_CONTEXT_DELIVERY=off` subprocess override, which
+also preserves compatibility with older CLI parsers. This does not change the
+machine preference or disable differential context for the interactive Agent.
+
+机器消费者若解析 `should_run`、选中工作或结算身份，应在内部 `quota should-run`
+调用中通过 `--context-delivery off` 或子进程 `LOOPX_CONTEXT_DELIVERY=off`
+显式请求原始输出，并保留原 registry、Goal、Agent 和 Turn。
+开启差分后的原生输出是展示／快照封装；封装没有 `should_run` 不表示准入拒绝。
+MCP 结算和 Claude 状态读取在子进程内设置已有的 `LOOPX_CONTEXT_DELIVERY=off`，
+兼容旧版 CLI 解析器；不修改机器开关，也不关闭交互
+Agent 的差分上下文。
+
 On accepts the declared native scenarios: Codex App heartbeat (`--codex-app` or its equivalent complete scheduler
 declaration), Codex App SSH Goal (`--runtime-profile codex_app_ssh_goal`), and the existing Codex benchmark heartbeat worker (`generic_cli`
 plus its explicit heartbeat worker environment). Missing, contradictory or

loopx/claude_goal_mode/hooks/goal_policy.py

@@ -52,6 +52,7 @@
 from __future__ import annotations
 
 import json
+import os
 import shutil
 import subprocess
 import sys
@@ -156,13 +157,15 @@ def should_run(registry, goal_id, agent_id=None) -> bool | None:
             capture_output=True,
             text=True, encoding="utf-8", errors="replace",
             timeout=10,
+            env={**os.environ, "LOOPX_CONTEXT_DELIVERY": "off"},
         )
         if _runtime_profile_flag_is_unsupported(out):
             out = subprocess.run(
                 [*cmd, *CLAUDE_LEGACY_SCHEDULER_ARGS],
                 capture_output=True,
                 text=True, encoding="utf-8", errors="replace",
                 timeout=10,
+                env={**os.environ, "LOOPX_CONTEXT_DELIVERY": "off"},
             )
     except Exception:
         return None  # caller fails CLOSED on None for non-read-only tools

loopx/claude_goal_mode/scripts/goalmode_cmd.py

@@ -24,6 +24,7 @@
 from __future__ import annotations
 
 import json
+import os
 import re
 import shutil
 import subprocess
@@ -148,7 +149,8 @@ def goal_detail(ctx):
             ["--format", "json", "quota", "should-run", "--goal-id", gid]
         if agent:
             cmd += ["--agent-id", agent]
-        out = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=15)
+        out = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=15,
+            env={**os.environ, "LOOPX_CONTEXT_DELIVERY": "off"})
         payload = json.loads(out.stdout or "{}")
     except Exception:
         pass

loopx/claude_goal_mode/statusline/goal_status.py

@@ -16,6 +16,7 @@
 from __future__ import annotations
 
 import json
+import os
 import subprocess
 import sys
 from pathlib import Path
@@ -86,7 +87,8 @@ def main():
         cmd += ["--format", "json", "quota", "should-run", "--goal-id", gid]
         if st.get("agent_id"):
             cmd += ["--agent-id", st["agent_id"]]
-        out = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=8)
+        out = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=8,
+            env={**os.environ, "LOOPX_CONTEXT_DELIVERY": "off"})
         d = json.loads(out.stdout or "{}")
         print(_render(gid, d))
     except Exception:

loopx/goal_mode_mcp.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import json
+import os
 import shutil
 import subprocess
 import sys
@@ -132,12 +133,17 @@ def run_cli(
         if registry:
             command += ["--registry", registry]
         command += ["--format", "json"]
+        # MCP parses/projects raw decisions itself. Pin only this subprocess;
+        # the existing environment opt-out also works with older CLI parsers.
+        environment = {**os.environ, "LOOPX_CONTEXT_DELIVERY": "off"}
         result = subprocess.run(
-            [*command, *args], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=30
+            [*command, *args], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=30,
+            env=environment,
         )
         if legacy_args is not None and self._runtime_profile_flag_is_unsupported(result):
             result = subprocess.run(
-                [*command, *legacy_args], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=30
+                [*command, *legacy_args], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=30,
+                env=environment,
             )
         return (result.stdout or "") + (
             ("\n" + result.stderr) if result.returncode else ""

tests/control_plane/test_machine_guard_context_contract.py

@@ -0,0 +1,153 @@
+"""Machine consumers retain quota semantics when native presentation is enabled."""
+from __future__ import annotations
+
+import io
+import json
+import os
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+from loopx.claude_goal_mode.scripts import goalmode_cmd
+from loopx.claude_goal_mode.statusline import goal_status
+from loopx.claude_goal_mode.hooks import goal_policy
+from loopx.control_plane.host_adapter_settlement import HostTodoSettlementRequest, _prepare
+from loopx.goal_mode_mcp import GoalModeMCPConfig, GoalModeMCPControlPlane
+from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime
+from tests.control_plane.test_context_machine_configuration import configure
+from tests.control_plane.test_quota_settlement_cli import (
+    AGENT_ID, GOAL_ID, TODO_ID, _initialize_fixture_authority, _spend_run_count, _write_fixture,
+)
+
+
+@pytest.fixture(params=["off", "on"])
+def machine(request, tmp_path, monkeypatch):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    monkeypatch.delenv("LOOPX_CONTEXT_DELIVERY", raising=False)
+    monkeypatch.delenv("LOOPX_EXECUTION_MODE", raising=False)
+    monkeypatch.setenv("CODEX_THREAD_ID", "machine-guard-fixture")
+    project, runtime, registry = _write_fixture(tmp_path)
+    _initialize_fixture_authority(project, runtime)
+    assert configure(registry, runtime, tmp_path, request.param)[0] == 0
+    monkeypatch.chdir(project)
+    prefix = [sys.executable, "-m", "loopx.cli", "--runtime-root", str(runtime)]
+    context = {"goal_id": GOAL_ID, "agent_id": AGENT_ID, "registry": str(registry)}
+    config_path = runtime / "machine/configuration.json"
+    configuration = config_path.read_bytes()
+    yield context, prefix, runtime
+    assert config_path.read_bytes() == configuration
+    assert "LOOPX_CONTEXT_DELIVERY" not in os.environ
+
+
+def assert_decision(payload, runtime):
+    assert payload["ok"] is True, payload
+    assert payload["should_run"] is True, payload
+    assert payload["goal_id"] == GOAL_ID
+    assert payload["selected_todo"]["todo_id"] == TODO_ID
+    assert _spend_run_count(runtime) == 0
+    assert not (runtime / "context-delivery").exists()
+
+
+def test_claude_status_detail_reads_full_decision(machine, monkeypatch):
+    context, prefix, runtime = machine
+    monkeypatch.setattr(goalmode_cmd, "gh_prefix", lambda: prefix)
+    _, payload = goalmode_cmd.goal_detail(context)
+    assert_decision(payload, runtime)
+
+
+def test_claude_statusline_reads_full_decision(machine, monkeypatch, capsys):
+    context, prefix, runtime = machine
+    monkeypatch.setattr(goal_status, "_gh_prefix", lambda: prefix)
+    monkeypatch.setattr(goal_status, "active_context", lambda _: context)
+    monkeypatch.setattr(goal_status, "_render", lambda _, payload: json.dumps(payload))
+    monkeypatch.setattr(sys, "stdin", io.StringIO("{}"))
+    goal_status.main()
+    assert_decision(json.loads(capsys.readouterr().out), runtime)
+
+
+def test_claude_tool_gate_reads_raw_decision(machine, monkeypatch):
+    context, prefix, runtime = machine
+    monkeypatch.setattr(goal_policy, "_gh_prefix", lambda: prefix)
+    assert goal_policy.should_run(context["registry"], GOAL_ID, AGENT_ID) is True
+    assert _spend_run_count(runtime) == 0
+    assert not (runtime / "context-delivery").exists()
+
+
+@pytest.mark.parametrize("completion", [False, True])
+def test_mcp_and_typed_completion_guards_use_raw_transport(machine, completion):
+    context, prefix, runtime = machine
+    control = GoalModeMCPControlPlane(
+        GoalModeMCPConfig(server_name="fixture", runtime_profile="codex_app_ssh_goal",
+                          legacy_host_surface="codex_app_ssh"),
+        lambda: context,
+    )
+    control.command_prefix = lambda: prefix
+    if completion:
+        request = HostTodoSettlementRequest(
+            goal_id=GOAL_ID, agent_id=AGENT_ID, todo_id=TODO_ID,
+            runtime_profile="codex_app_ssh_goal", legacy_host_surface="codex_app_ssh",
+            scheduler_owner="agent_cli_loop", execution_mode="interactive",
+            completion_args=(),
+        )
+        plan = _prepare(request)
+        step = plan["provider_effect"]["steps"][0]
+        args, legacy = step["args"], step["legacy_args"]
+    else:
+        args, legacy = control.should_run_args(GOAL_ID, AGENT_ID)
+    payload = json.loads(control.run_cli(args, legacy_args=legacy))
+    assert_decision(payload, runtime)
+    if completion:
+        assert payload["heartbeat_receipt"]["settlement_identity"] == plan["identity"]
+    # Both current and legacy host declarations use the same machine contract.
+    legacy_payload = json.loads(control.run_cli(legacy))
+    assert_decision(legacy_payload, runtime)
+
+
+def test_rejected_guard_never_completes_or_spends_with_machine_on(machine):
+    context, prefix, runtime = machine
+    registry = Path(context["registry"])
+    document = json.loads(registry.read_text())
+    document["goals"][0]["adapter"].pop("status")
+    registry.write_text(json.dumps(document))
+    state = registry.parent.parent / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md"
+    before = state.read_bytes()
+    control = GoalModeMCPControlPlane(
+        GoalModeMCPConfig(server_name="fixture", runtime_profile="codex_app_ssh_goal",
+                          legacy_host_surface="codex_app_ssh"),
+        lambda: context,
+    )
+    control.command_prefix = lambda: prefix
+    result = json.loads(control.complete_task(TODO_ID, AGENT_ID, "must not complete"))
+    assert result["ok"] is False and result["settlement_blocked_completion"] is True
+    assert result["settlement"]["failed_stage"] == "guard"
+    assert state.read_bytes() == before
+    assert _spend_run_count(runtime) == 0
+    assert not (runtime / "context-delivery").exists()
+
+
+def test_mcp_legacy_retry_keeps_raw_transport_without_new_parser_flags(monkeypatch):
+    import loopx.goal_mode_mcp as mcp
+
+    monkeypatch.setenv("LOOPX_CONTEXT_DELIVERY", "on")
+    calls = []
+
+    def run(argv, **kwargs):
+        calls.append((argv, kwargs["env"]))
+        assert "--context-delivery" not in argv
+        assert kwargs["env"]["LOOPX_CONTEXT_DELIVERY"] == "off"
+        if "--runtime-profile" in argv:
+            return SimpleNamespace(returncode=2, stdout="", stderr="unrecognized arguments: --runtime-profile")
+        return SimpleNamespace(returncode=0, stdout='{"ok":true,"should_run":false}', stderr="")
+
+    monkeypatch.setattr(mcp.subprocess, "run", run)
+    control = GoalModeMCPControlPlane(
+        GoalModeMCPConfig(server_name="fixture", runtime_profile="claude_code", legacy_host_surface="claude_code"),
+        lambda: {"goal_id": GOAL_ID, "agent_id": AGENT_ID},
+    )
+    args, legacy = control.should_run_args(GOAL_ID, AGENT_ID)
+    result = json.loads(control.run_cli(args, legacy_args=legacy))
+    assert result == {"ok": True, "should_run": False}
+    assert len(calls) == 2 and os.environ["LOOPX_CONTEXT_DELIVERY"] == "on"
+