feat: mcp activity history for Codex

Author: pbeslin-gg

ggshield/verticals/ai/agents/codex.py

@@ -1,6 +1,7 @@
 import json
+from datetime import datetime
 from pathlib import Path
-from typing import Any, Dict, Iterator, Literal
+from typing import Any, Dict, Iterator, Literal, Optional
 
 import click
 from pygitguardian.models import AIDiscovery, MCPActivityRequest
@@ -109,20 +110,114 @@ def parse_mcp_activity(
         tool = parts[-1]
         server_cfg_name = "__".join(parts[1:-1])
 
-        server_name = server_cfg_name
-        for server in ai_config.servers:
-            for configuration in server.configurations:
-                if _mangle_server_name(configuration.name) == server_cfg_name:
-                    server_name = server.name
-                    break
-
         return MCPActivityRequest(
             user=ai_config.user,
             tool=tool,
-            server=server_name,
+            server=self._resolve_server_name(server_cfg_name, ai_config),
             agent=self.name,
             model=payload.raw.get("model", ""),
             cwd=payload.raw.get("cwd", ""),
             input=payload.raw.get("tool_input", {}),
             timestamp=payload.timestamp,
         )
+
+    def iter_history_events(
+        self, ai_config: Optional[AIDiscovery]
+    ) -> Iterator[MCPActivityRequest]:
+        """Walk every Codex session rollout and yield its MCP tool_use events."""
+        for path in self._history_files():
+            yield from self._parse_session_file(path, ai_config)
+
+    def _history_files(self) -> Iterator[Path]:
+        """Yield every Codex session rollout file we know about."""
+        yield from sorted(self.config_folder.glob("sessions/*/*/*/rollout-*.jsonl"))
+
+    def _parse_session_file(
+        self, path: Path, ai_config: Optional[AIDiscovery]
+    ) -> Iterator[MCPActivityRequest]:
+        """Yield MCPActivityRequest events from a single Codex session rollout.
+
+        Codex stores one session per JSONL file. Each line is a typed envelope
+        with ``timestamp``, ``type``, ``payload``. MCP tool calls appear as
+        ``response_item`` payloads of type ``function_call`` whose ``namespace``
+        starts with ``mcp__``. ``cwd`` is seeded by ``session_meta`` and may be
+        updated per-turn by ``turn_context``; ``model`` only appears on
+        ``turn_context``.
+        """
+        cwd = ""
+        model = ""
+        for entry in self._load_jsonl_file(path):
+            if not isinstance(entry, dict):
+                continue
+            payload = entry.get("payload")
+            if not isinstance(payload, dict):
+                continue
+            entry_type = entry.get("type")
+            if entry_type == "session_meta":
+                cwd = payload.get("cwd") or cwd
+                continue
+            if entry_type == "turn_context":
+                cwd = payload.get("cwd") or cwd
+                model = payload.get("model") or model
+                continue
+            if entry_type != "response_item":
+                continue
+            if payload.get("type") != "function_call":
+                continue
+            namespace = payload.get("namespace") or ""
+            if not namespace.startswith("mcp__"):
+                continue
+            event = self._build_activity_from_function_call(
+                entry, payload, namespace, cwd, model, ai_config
+            )
+            if event is not None:
+                yield event
+
+    def _build_activity_from_function_call(
+        self,
+        entry: Dict[str, Any],
+        payload: Dict[str, Any],
+        namespace: str,
+        cwd: str,
+        model: str,
+        ai_config: Optional[AIDiscovery],
+    ) -> Optional[MCPActivityRequest]:
+        """Turn one function_call response_item into an MCPActivityRequest."""
+        tool = payload.get("name") or ""
+        if not tool:
+            return None
+        server_cfg_name = namespace.removeprefix("mcp__").removesuffix("__")
+        try:
+            tool_input = json.loads(payload.get("arguments") or "{}")
+        except (json.JSONDecodeError, TypeError):
+            tool_input = {}
+        if not isinstance(tool_input, dict):
+            tool_input = {}
+        try:
+            ts = datetime.fromisoformat(
+                str(entry.get("timestamp", "")).replace("Z", "+00:00")
+            )
+        except ValueError:
+            return None
+        return MCPActivityRequest(
+            user=self._user_or_default(ai_config),
+            tool=tool,
+            server=self._resolve_server_name(server_cfg_name, ai_config),
+            agent=self.name,
+            model=model,
+            cwd=cwd,
+            input=tool_input,
+            timestamp=ts,
+        )
+
+    def _resolve_server_name(
+        self, cfg_name: str, ai_config: Optional[AIDiscovery]
+    ) -> str:
+        """Look up the canonical server name; fall back to the configuration name."""
+        if ai_config is None:
+            return cfg_name
+        for server in ai_config.servers:
+            for configuration in server.configurations:
+                if _mangle_server_name(configuration.name) == cfg_name:
+                    return server.name
+        return cfg_name

tests/unit/verticals/ai/test_codex_history.py

@@ -0,0 +1,183 @@
+import json
+from datetime import datetime, timezone
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+from pygitguardian.models import AIDiscovery, MCPConfiguration, MCPServer, UserInfo
+
+from ggshield.verticals.ai.agents.codex import Codex
+
+
+def _function_call_entry(
+    *,
+    name: str = "get_issue",
+    namespace: str = "mcp__linear__",
+    arguments: str = '{"id": "NHI-1"}',
+    timestamp: str = "2026-04-01T09:00:00.000Z",
+) -> dict:
+    """Build a Codex JSONL response_item line carrying an MCP function_call."""
+    return {
+        "timestamp": timestamp,
+        "type": "response_item",
+        "payload": {
+            "type": "function_call",
+            "name": name,
+            "namespace": namespace,
+            "arguments": arguments,
+            "call_id": "call_AAA",
+        },
+    }
+
+
+def _session_meta(cwd: str = "/home/u/repo") -> dict:
+    return {
+        "timestamp": "2026-04-01T08:55:00.000Z",
+        "type": "session_meta",
+        "payload": {"id": "sess-1", "cwd": cwd, "cli_version": "0.130.0"},
+    }
+
+
+def _turn_context(cwd: str = "/home/u/repo", model: str = "gpt-5.5") -> dict:
+    return {
+        "timestamp": "2026-04-01T08:56:00.000Z",
+        "type": "turn_context",
+        "payload": {"turn_id": "turn-1", "cwd": cwd, "model": model},
+    }
+
+
+def _write_session(path: Path, entries: list[dict]) -> None:
+    path.write_text("\n".join(json.dumps(e) for e in entries) + "\n")
+
+
+@pytest.fixture
+def empty_ai_config() -> AIDiscovery:
+    return AIDiscovery(
+        user=UserInfo(hostname="h", username="u", machine_id="m"),
+        servers=[],
+        discovery_duration=0.0,
+    )
+
+
+class TestCodexParseSessionFile:
+    def test_extracts_mcp_tool_use(self, tmp_path: Path, empty_ai_config) -> None:
+        path = tmp_path / "rollout.jsonl"
+        _write_session(
+            path,
+            [
+                _session_meta(),
+                _turn_context(model="gpt-5.5"),
+                _function_call_entry(),
+            ],
+        )
+
+        events = list(Codex()._parse_session_file(path, empty_ai_config))
+
+        assert len(events) == 1
+        ev = events[0]
+        assert ev.tool == "get_issue"
+        assert ev.server == "linear"
+        assert ev.agent == "codex"
+        assert ev.model == "gpt-5.5"
+        assert ev.cwd == "/home/u/repo"
+        assert ev.input == {"id": "NHI-1"}
+        assert ev.timestamp == datetime(2026, 4, 1, 9, 0, tzinfo=timezone.utc)
+
+    def test_ignores_non_mcp_function_calls(
+        self, tmp_path: Path, empty_ai_config
+    ) -> None:
+        path = tmp_path / "rollout.jsonl"
+        _write_session(
+            path,
+            [
+                _session_meta(),
+                _turn_context(),
+                # Built-in shell tool: no namespace.
+                {
+                    "timestamp": "2026-04-01T09:00:01.000Z",
+                    "type": "response_item",
+                    "payload": {
+                        "type": "function_call",
+                        "name": "exec_command",
+                        "arguments": '{"cmd": "ls"}',
+                        "call_id": "x",
+                    },
+                },
+                # A non-function_call response_item.
+                {
+                    "timestamp": "2026-04-01T09:00:02.000Z",
+                    "type": "response_item",
+                    "payload": {"type": "message", "role": "assistant"},
+                },
+            ],
+        )
+        assert list(Codex()._parse_session_file(path, empty_ai_config)) == []
+
+    def test_tracks_cwd_and_model_across_turns(
+        self, tmp_path: Path, empty_ai_config
+    ) -> None:
+        path = tmp_path / "rollout.jsonl"
+        _write_session(
+            path,
+            [
+                _session_meta(cwd="/home/u/initial"),
+                _turn_context(cwd="/home/u/turn1", model="gpt-5.5"),
+                _function_call_entry(timestamp="2026-04-01T09:00:01.000Z"),
+                _turn_context(cwd="/home/u/turn2", model="gpt-5.6"),
+                _function_call_entry(timestamp="2026-04-01T09:00:02.000Z"),
+            ],
+        )
+
+        events = list(Codex()._parse_session_file(path, empty_ai_config))
+
+        assert [(e.cwd, e.model) for e in events] == [
+            ("/home/u/turn1", "gpt-5.5"),
+            ("/home/u/turn2", "gpt-5.6"),
+        ]
+
+    def test_resolves_server_display_name_from_discovery(self, tmp_path: Path) -> None:
+        config = AIDiscovery(
+            user=UserInfo(hostname="h", username="u", machine_id="m"),
+            servers=[
+                MCPServer(
+                    name="LinearDisplay",
+                    configurations=[
+                        MCPConfiguration(
+                            name="linear",
+                            agent="codex",
+                            scope=MCPConfiguration.Scope.USER,
+                            transport=MCPConfiguration.Transport.HTTP,
+                            project=None,
+                        )
+                    ],
+                ),
+            ],
+            discovery_duration=0.0,
+        )
+        path = tmp_path / "rollout.jsonl"
+        _write_session(path, [_session_meta(), _turn_context(), _function_call_entry()])
+
+        events = list(Codex()._parse_session_file(path, config))
+
+        assert events[0].server == "LinearDisplay"
+
+
+class TestCodexHistoryFiles:
+    def test_globs_rollouts_under_dated_dirs(self, tmp_path: Path) -> None:
+        sessions = tmp_path / ".codex" / "sessions"
+        (sessions / "2026" / "04" / "01").mkdir(parents=True)
+        (sessions / "2026" / "04" / "01" / "rollout-1.jsonl").write_text("{}\n")
+        (sessions / "2026" / "04" / "02").mkdir(parents=True)
+        (sessions / "2026" / "04" / "02" / "rollout-2.jsonl").write_text("{}\n")
+        # Wrong depth — should be ignored.
+        (sessions / "loose.jsonl").write_text("{}\n")
+        # Wrong prefix — should be ignored.
+        (sessions / "2026" / "04" / "02" / "other.jsonl").write_text("{}\n")
+
+        with patch(
+            "ggshield.verticals.ai.agents.codex.get_user_home_dir",
+            return_value=tmp_path,
+        ):
+            files = sorted(Codex()._history_files())
+
+        assert [f.name for f in files] == ["rollout-1.jsonl", "rollout-2.jsonl"]