Add headless trace artifacts and runtime diagnostics

Author: QUSETIONS

minicode/headless.py

@@ -16,11 +16,40 @@
 
 from __future__ import annotations
 
+import json
 import os
 import sys
 from pathlib import Path
 
 
+def _write_headless_messages_trace(
+    trace_path: str | None,
+    *,
+    cwd: str,
+    prompt: str,
+    runtime: dict | None,
+    result_messages: list[dict] | None,
+    response_text: str | None,
+    error_text: str | None = None,
+) -> None:
+    if not trace_path:
+        return
+    payload = {
+        "cwd": cwd,
+        "prompt": prompt,
+        "model": (runtime or {}).get("model"),
+        "messages": result_messages or [],
+        "assistant_response": response_text,
+        "error": error_text,
+    }
+    path = Path(trace_path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(
+        json.dumps(payload, ensure_ascii=False, indent=2, default=str),
+        encoding="utf-8",
+    )
+
+
 def run_headless(prompt: str | None = None) -> str:
     """Run a single agent turn in headless mode and return the response.
 
@@ -91,6 +120,7 @@ def run_headless(prompt: str | None = None) -> str:
     ]
 
     logger.info("Headless run: %s", prompt[:80])
+    trace_output_path = os.environ.get("MINI_CODE_HEADLESS_MESSAGES_OUT", "").strip() or None
 
     try:
         result_messages = run_agent_turn(
@@ -107,11 +137,30 @@ def run_headless(prompt: str | None = None) -> str:
             (m for m in reversed(result_messages) if m["role"] == "assistant"),
             None,
         )
-        return last_assistant["content"] if last_assistant else "(no response)"
+        response_text = last_assistant["content"] if last_assistant else "(no response)"
+        _write_headless_messages_trace(
+            trace_output_path,
+            cwd=cwd,
+            prompt=prompt,
+            runtime=runtime,
+            result_messages=result_messages,
+            response_text=response_text,
+        )
+        return response_text
 
     except Exception as exc:  # noqa: BLE001
         logger.error("Headless error: %s", exc)
-        return f"Error: {exc}"
+        response_text = f"Error: {exc}"
+        _write_headless_messages_trace(
+            trace_output_path,
+            cwd=cwd,
+            prompt=prompt,
+            runtime=runtime,
+            result_messages=[],
+            response_text=response_text,
+            error_text=str(exc),
+        )
+        return response_text
     finally:
         try:
             tools.dispose()

py-src/minicode/headless.py

@@ -16,11 +16,40 @@
 
 from __future__ import annotations
 
+import json
 import os
 import sys
 from pathlib import Path
 
 
+def _write_headless_messages_trace(
+    trace_path: str | None,
+    *,
+    cwd: str,
+    prompt: str,
+    runtime: dict | None,
+    result_messages: list[dict] | None,
+    response_text: str | None,
+    error_text: str | None = None,
+) -> None:
+    if not trace_path:
+        return
+    payload = {
+        "cwd": cwd,
+        "prompt": prompt,
+        "model": (runtime or {}).get("model"),
+        "messages": result_messages or [],
+        "assistant_response": response_text,
+        "error": error_text,
+    }
+    path = Path(trace_path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(
+        json.dumps(payload, ensure_ascii=False, indent=2, default=str),
+        encoding="utf-8",
+    )
+
+
 def run_headless(prompt: str | None = None) -> str:
     """Run a single agent turn in headless mode and return the response.
 
@@ -103,6 +132,7 @@ def run_headless(prompt: str | None = None) -> str:
             max_steps = max(1, int(raw_max_steps))
         except ValueError:
             max_steps = 50
+    trace_output_path = os.environ.get("MINI_CODE_HEADLESS_MESSAGES_OUT", "").strip() or None
 
     try:
         result_messages = run_agent_turn(
@@ -112,18 +142,38 @@ def run_headless(prompt: str | None = None) -> str:
             cwd=cwd,
             permissions=permissions,
             max_steps=max_steps,
+            runtime=runtime,
         )
 
         # Extract last assistant message
         last_assistant = next(
             (m for m in reversed(result_messages) if m["role"] == "assistant"),
             None,
         )
-        return last_assistant["content"] if last_assistant else "(no response)"
+        response_text = last_assistant["content"] if last_assistant else "(no response)"
+        _write_headless_messages_trace(
+            trace_output_path,
+            cwd=cwd,
+            prompt=prompt,
+            runtime=runtime,
+            result_messages=result_messages,
+            response_text=response_text,
+        )
+        return response_text
 
     except Exception as exc:  # noqa: BLE001
         logger.error("Headless error: %s", exc)
-        return f"Error: {exc}"
+        response_text = f"Error: {exc}"
+        _write_headless_messages_trace(
+            trace_output_path,
+            cwd=cwd,
+            prompt=prompt,
+            runtime=runtime,
+            result_messages=[],
+            response_text=response_text,
+            error_text=str(exc),
+        )
+        return response_text
     finally:
         try:
             tools.dispose()

tests/test_headless.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 from pathlib import Path
 
 from minicode.tooling import ToolRegistry
@@ -121,3 +122,52 @@ def test_run_headless_provider_failure_uses_runtime_channel_details(
     assert "Active channel: anthropic-compatible via baseUrl/authToken." in response
     assert "Next step: Primary runtime is using a single anthropic-compatible channel from baseUrl/authToken." in response
 
+
+def test_run_headless_writes_messages_trace_when_requested(monkeypatch, tmp_path: Path) -> None:
+    import minicode.headless
+
+    runtime = {
+        "model": "deepseek-v4-pro[1m]",
+        "baseUrl": "https://openai-proxy.example/v1",
+        "authToken": "test-token",
+    }
+    trace_path = tmp_path / "artifacts" / "messages.json"
+
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setenv("MINI_CODE_HEADLESS_MESSAGES_OUT", str(trace_path))
+    monkeypatch.setattr(
+        "minicode.config.load_runtime_config",
+        lambda cwd: runtime,
+    )
+    monkeypatch.setattr(
+        "minicode.tools.create_default_tool_registry",
+        lambda cwd, runtime=None: ToolRegistry([]),
+    )
+    monkeypatch.setattr("minicode.permissions.PermissionManager", _DummyPermissions)
+    monkeypatch.setattr("minicode.memory.MemoryManager", _DummyMemoryManager)
+    monkeypatch.setattr(
+        "minicode.prompt.build_system_prompt",
+        lambda cwd, permissions, context: "sys",
+    )
+    monkeypatch.setattr(
+        "minicode.model_registry.create_model_adapter",
+        lambda model, tools, runtime=None: object(),
+    )
+    monkeypatch.setattr(
+        "minicode.agent_loop.run_agent_turn",
+        lambda **kwargs: [
+            {"role": "assistant", "content": "traceable"},
+            {"role": "tool", "content": "python -m unittest"},
+        ],
+    )
+
+    response = minicode.headless.run_headless("Run the visible tests.")
+
+    assert response == "traceable"
+    payload = json.loads(trace_path.read_text(encoding="utf-8"))
+    assert payload["cwd"] == str(tmp_path)
+    assert payload["prompt"] == "Run the visible tests."
+    assert payload["model"] == "deepseek-v4-pro[1m]"
+    assert payload["assistant_response"] == "traceable"
+    assert payload["error"] is None
+    assert payload["messages"][0]["role"] == "assistant"