fix: preserve runtime status, verification, and response identities

core/platform_compat.py

@@ -118,12 +118,14 @@ def normalize_stdio_command(
     env: Mapping[str, Any] | None = None,
     *,
     inherit_env: bool = True,
+    map_python_to_current_interpreter: bool = True,
 ) -> tuple[str, list[str], dict[str, str]]:
     """Normalize MCP stdio command/env for Windows and virtualenv safety.
 
-    - ``python`` / ``python3`` map to the current interpreter on every OS, so
-      MCP tool servers run in the same environment as the backend — except in
-      the frozen build, whose executable is not a Python.
+    - By default, bare ``python`` / ``python3`` map to the current interpreter,
+      so MCP tool servers run in the backend environment. Callers running
+      project-owned commands can disable this mapping; frozen builds also leave
+      Python on PATH because their executable is not a Python.
     - Windows ``.cmd``/``.bat`` launchers such as ``npx`` are wrapped in
       ``cmd.exe /d /c`` because CreateProcess does not reliably execute them
       as stdio children across environments.
@@ -136,7 +138,8 @@ def normalize_stdio_command(
     base = _basename(command)
     bare_command = "/" not in command and "\\" not in command
     if (
-        bare_command
+        map_python_to_current_interpreter
+        and bare_command
         and base in {"python", "python.exe", "python3", "python3.exe"}
         and not getattr(sys, "frozen", False)
     ):

core/providers/openai_compat.py

@@ -373,6 +373,7 @@ def _sanitize_messages(
         messages: list[dict[str, Any]],
         *,
         preserve_provider_state: bool = False,
+        normalize_tool_ids: bool = True,
     ) -> list[dict[str, Any]]:
         """Strip non-standard keys, normalize tool_call IDs."""
         allowed_keys = (
@@ -384,7 +385,7 @@ def _sanitize_messages(
         id_map: dict[str, str] = {}
 
         def map_id(value: Any) -> Any:
-            if not isinstance(value, str):
+            if not normalize_tool_ids or not isinstance(value, str):
                 return value
             return id_map.setdefault(value, self._normalize_tool_call_id(value))
 
@@ -624,7 +625,9 @@ def _build_responses_body(
         self.validate_request_capabilities(messages, tools)
         model_name = model or self.default_model
         sanitized_messages = self._sanitize_messages(
-            self._sanitize_empty_content(messages), preserve_provider_state=True
+            self._sanitize_empty_content(messages),
+            preserve_provider_state=True,
+            normalize_tool_ids=False,
         )
         instructions, input_items = convert_messages(sanitized_messages)
 

core/providers/openai_responses/converters.py

@@ -56,13 +56,13 @@ def convert_messages(
                         "id": f"msg_{idx}",
                     }
                 )
-            for tool_call in msg.get("tool_calls", []) or []:
+            for tool_idx, tool_call in enumerate(msg.get("tool_calls", []) or []):
                 fn = tool_call.get("function") or {}
                 call_id, item_id = split_tool_call_id(tool_call.get("id"))
                 input_items.append(
                     {
                         "type": "function_call",
-                        "id": item_id or f"fc_{idx}",
+                        "id": item_id or f"fc_{idx}_{tool_idx}",
                         "call_id": call_id or f"call_{idx}",
                         "name": fn.get("name"),
                         "arguments": fn.get("arguments") or "{}",

core/providers/openai_responses/parsing.py

@@ -27,6 +27,11 @@ def map_finish_reason(status: str | None) -> str:
     return FINISH_REASON_MAP.get(status or "completed", "stop")
 
 
+def _tool_call_id(call_id: str, item_id: Any) -> str:
+    """Keep Responses item IDs when present; converters assign missing IDs uniquely."""
+    return f"{call_id}|{item_id}" if item_id else call_id
+
+
 async def iter_sse(response: httpx.Response) -> AsyncGenerator[dict[str, Any], None]:
     """Yield parsed JSON events from a Responses API SSE stream."""
     buffer: list[str] = []
@@ -83,7 +88,7 @@ async def consume_sse(
                 if not call_id:
                     continue
                 tool_call_buffers[call_id] = {
-                    "id": item.get("id") or "fc_0",
+                    "id": item.get("id"),
                     "name": item.get("name"),
                     "arguments": item.get("arguments") or "",
                 }
@@ -125,7 +130,7 @@ async def consume_sse(
                         args = {"raw": args_raw}
                 tool_calls.append(
                     ToolCallRequest(
-                        id=f"{call_id}|{buf.get('id') or item.get('id') or 'fc_0'}",
+                        id=_tool_call_id(call_id, buf.get("id") or item.get("id")),
                         name=buf.get("name") or item.get("name") or "",
                         arguments=args,
                     )
@@ -175,7 +180,7 @@ def parse_response_output(response: Any) -> LLMResponse:
                     reasoning_content = (reasoning_content or "") + s["text"]
         elif item_type == "function_call":
             call_id = item.get("call_id") or ""
-            item_id = item.get("id") or "fc_0"
+            item_id = item.get("id")
             args_raw = item.get("arguments") or "{}"
             try:
                 args = json.loads(args_raw) if isinstance(args_raw, str) else args_raw
@@ -194,7 +199,7 @@ def parse_response_output(response: Any) -> LLMResponse:
                     args = {"raw": args_raw}
             tool_calls.append(
                 ToolCallRequest(
-                    id=f"{call_id}|{item_id}",
+                    id=_tool_call_id(call_id, item_id),
                     name=item.get("name") or "",
                     arguments=args if isinstance(args, dict) else {},
                 )
@@ -264,7 +269,7 @@ async def consume_sdk_stream(
                 if not call_id:
                     continue
                 tool_call_buffers[call_id] = {
-                    "id": getattr(item, "id", None) or "fc_0",
+                    "id": getattr(item, "id", None),
                     "name": getattr(item, "name", None),
                     "arguments": getattr(item, "arguments", None) or "",
                 }
@@ -312,7 +317,9 @@ async def consume_sdk_stream(
                         args = {"raw": args_raw}
                 tool_calls.append(
                     ToolCallRequest(
-                        id=f"{call_id}|{buf.get('id') or getattr(item, 'id', None) or 'fc_0'}",
+                        id=_tool_call_id(
+                            call_id, buf.get("id") or getattr(item, "id", None)
+                        ),
                         name=buf.get("name") or getattr(item, "name", None) or "",
                         arguments=args,
                     )

core/schedule/monitor.py

@@ -493,7 +493,7 @@ def main(argv: list[str] | None = None) -> int:
             print(f"[monitor] changed: {args.job_id}")
             if outcome.context_block:
                 print(outcome.context_block)
-    return 0
+    return 1 if outcome.error else 0
 
 
 if __name__ == "__main__":

core/verification.py

@@ -11,6 +11,8 @@
 from dataclasses import dataclass
 from pathlib import Path
 
+from core.platform_compat import normalize_stdio_command
+
 
 MAX_VERIFICATION_OUTPUT = 64 * 1024
 MAX_DISCOVERY_FILES = 256
@@ -105,12 +107,18 @@ def run_verification(
     timeout_seconds: int = 300,
 ) -> VerificationResult:
     started = time.monotonic()
+    executable, arguments, environment = normalize_stdio_command(
+        command.argv[0],
+        list(command.argv[1:]),
+        {"CI": "1", "PYTHONUNBUFFERED": "1"},
+        map_python_to_current_interpreter=False,
+    )
     process = subprocess.Popen(
-        command.argv,
+        [executable, *arguments],
         cwd=root,
         stdout=subprocess.PIPE,
         stderr=subprocess.PIPE,
-        env={**os.environ, "CI": "1", "PYTHONUNBUFFERED": "1"},
+        env=environment,
         start_new_session=os.name != "nt",
     )
     stdout_raw, stderr_raw, timed_out, truncated = _capture(process, timeout_seconds)

tests/test_monitor_cli_status.py

@@ -0,0 +1,58 @@
+"""Exercise the documented module CLI, its shell source and persistent state."""
+
+import json
+import subprocess
+import sys
+
+import pytest
+
+from core.schedule.monitor import MonitorStore
+
+
+@pytest.mark.parametrize("as_json", [False, True])
+def test_source_failure_returns_nonzero_without_changing_state(tmp_path, as_json):
+    db = tmp_path / "monitor.sqlite3"
+    store = MonitorStore(str(db))
+    store.save_job("sample", "previous-hash", "previous-output")
+    previous = store.get_job("sample")
+    command = [
+        sys.executable,
+        "-m",
+        "core.schedule.monitor",
+        "--job-id",
+        "sample",
+        "--script",
+        "exit 3",
+        "--db",
+        str(db),
+    ]
+    if as_json:
+        command.append("--json")
+    result = subprocess.run(command, capture_output=True, text=True, timeout=15)
+    assert store.get_job("sample") == previous
+    if as_json:
+        assert json.loads(result.stdout)["error"]
+    else:
+        assert "ERROR" in result.stderr
+    assert result.returncode == 1
+
+
+def test_json_success_and_unchanged_source_return_zero(tmp_path):
+    command = [
+        sys.executable,
+        "-m",
+        "core.schedule.monitor",
+        "--job-id",
+        "sample",
+        "--script",
+        "echo stable",
+        "--db",
+        str(tmp_path / "monitor.sqlite3"),
+        "--json",
+    ]
+    for changed in (True, False):
+        result = subprocess.run(command, capture_output=True, text=True, timeout=15)
+        assert result.returncode == 0
+        outcome = json.loads(result.stdout)
+        assert outcome["changed"] is changed
+        assert not outcome["error"]

tests/test_platform_compat.py

@@ -24,6 +24,17 @@ def test_normalize_stdio_command_maps_only_bare_python_names() -> None:
     assert command == explicit
 
 
+def test_normalize_stdio_command_can_preserve_project_python_name() -> None:
+    command, _args, _environment = normalize_stdio_command(
+        "python3",
+        [],
+        {},
+        inherit_env=False,
+        map_python_to_current_interpreter=False,
+    )
+    assert command == "python3"
+
+
 def test_frozen_build_leaves_python_to_the_path(monkeypatch) -> None:
     """The Desktop build's executable is the App Server, not a Python: an MCP
     server configured as ``python server.py`` needs the user's Python."""

tests/test_reasoning_providers.py

@@ -375,3 +375,36 @@ async def on_reasoning(delta: str, channel: ReasoningChannel) -> None:
         ("private-1", expected_channel),
         ("private-2", expected_channel),
     ]
+
+
+@pytest.mark.parametrize("compound_ids", [False, True])
+def test_responses_parallel_tool_calls_preserve_distinct_identities(compound_ids):
+    provider = _openai_provider()
+    ids = [f"call_{i}|fc_{i}" if compound_ids else f"call_{i}" for i in range(2)]
+    messages = [
+        {
+            "role": "assistant",
+            "content": None,
+            "tool_calls": [
+                {
+                    "id": identifier,
+                    "type": "function",
+                    "function": {"name": "probe", "arguments": "{}"},
+                }
+                for identifier in ids
+            ],
+        }
+    ] + [
+        {"role": "tool", "tool_call_id": identifier, "content": "ok"}
+        for identifier in ids
+    ]
+    body = provider._build_responses_body(
+        messages, None, "gpt-5.4", 1024, 0.1, None, None
+    )
+    calls = [item for item in body["input"] if item["type"] == "function_call"]
+    outputs = [item for item in body["input"] if item["type"] == "function_call_output"]
+    assert [item["call_id"] for item in calls] == ["call_0", "call_1"]
+    assert [item["call_id"] for item in outputs] == ["call_0", "call_1"]
+    assert len({item["id"] for item in calls}) == 2
+    if compound_ids:
+        assert [item["id"] for item in calls] == ["fc_0", "fc_1"]