From ecbbe1f6736706e6e7cb869834d2587e89624dd1 Mon Sep 17 00:00:00 2001 From: Baris Ozbas Date: Sun, 13 Sep 2026 13:49:14 +0200 Subject: [PATCH] feat(sensor): capture Pi conversations and branch-aware tool activity Summary: Intent: - Add local-file observability for Pi sessions without hooks, credentials, or agent configuration changes. Changes: - Read v1-v3 JSONL sessions and preserve all recorded branches, full tool arguments/results, failures, provider/model metadata, typed content, and recorded usage. - Correlate results along their branch ancestry, preserve extra/orphan results, and distinguish user shell commands from assistant actions. - Register the pi source, refresh resumed session exports, and document native macOS/Linux/Windows storage paths, overrides, and capture limits. - Add contract fixtures and a native three-OS parser CI matrix without new runtime dependencies. Test Plan: Authenticated Pi model sessions have not been run. Before production rollout, run a Pi conversation with read/write/shell tools, resume and branch it, then compare adr-sensor --source pi --save-sessions output with the persisted JSONL. In-memory sessions and content Pi never persisted are outside capture scope. Automated CI replays contract fixtures rather than launching the agent. Revert Plan: Revert this change to remove the pi source; existing exported JSON remains readable. Issue Links: None - user-requested source onboarding without a tracking issue. --- .github/workflows/sensor-pi.yml | 23 ++ README.md | 2 +- Sensor/README.md | 62 +++- Sensor/adr_sensor/__init__.py | 2 +- Sensor/adr_sensor/cli.py | 1 + Sensor/adr_sensor/observer.py | 7 +- Sensor/adr_sensor/parsers/__init__.py | 2 + Sensor/adr_sensor/parsers/pi_parser.py | 295 ++++++++++++++++++ .../adr_sensor/schemas/agent_event_schema.py | 4 +- Sensor/tests/test_pi_parser.py | 250 +++++++++++++++ 10 files changed, 641 insertions(+), 7 deletions(-) create mode 100644 .github/workflows/sensor-pi.yml create mode 100644 Sensor/adr_sensor/parsers/pi_parser.py create mode 100644 Sensor/tests/test_pi_parser.py diff --git a/.github/workflows/sensor-pi.yml b/.github/workflows/sensor-pi.yml new file mode 100644 index 0000000..95678f2 --- /dev/null +++ b/.github/workflows/sensor-pi.yml @@ -0,0 +1,23 @@ +name: Pi parser platforms +on: + pull_request: + paths: ['Sensor/**', '.github/workflows/sensor-pi.yml'] + push: + branches: [main] + paths: ['Sensor/**', '.github/workflows/sensor-pi.yml'] +permissions: + contents: read +jobs: + parser: + strategy: + fail-fast: false + matrix: + os: [ubuntu-latest, macos-latest, windows-latest] + runs-on: ${{ matrix.os }} + steps: + - uses: actions/checkout@v7 + - uses: astral-sh/setup-uv@v7 + - run: uv sync --extra dev --python 3.12 + working-directory: Sensor + - run: uv run pytest tests/test_pi_parser.py -q + working-directory: Sensor diff --git a/README.md b/README.md index 02939c3..d2f4fb5 100644 --- a/README.md +++ b/README.md @@ -23,7 +23,7 @@ This repository contains the open-source **ADR Discovery**, **ADR Sensor**, **AD | Path | ADR component | Description | | -------------------------------------------------- | -------------------------- | ------------------------------------------------------------------------------------ | | [Discovery/](Discovery/) | ADR Discovery | Inventory the AI apps, CLI agents, IDE extensions, model runtimes, and MCP servers on an endpoint, and flag unknown surfaces for review | -| [Sensor/](Sensor/) | ADR Observability | Collect and normalize agent telemetry from Claude Code, Cursor, Codex, GitHub Copilot CLI, opencode, Claude Desktop, and others | +| [Sensor/](Sensor/) | ADR Observability | Collect and normalize agent telemetry from Claude Code, Cursor, Codex, GitHub Copilot CLI, Pi, opencode, Claude Desktop, and others | | [Detection/](Detection/) | ADR Benchmark + Detection | Dual-agent detector, 134 MCP servers, 304 benchmark tasks, baselines, figure scripts | | [docs/REPRODUCIBILITY.md](docs/REPRODUCIBILITY.md) | Evaluation | Step-by-step workflow to reproduce benchmark detection and paper figures | diff --git a/Sensor/README.md b/Sensor/README.md index 215ec70..e5cc951 100644 --- a/Sensor/README.md +++ b/Sensor/README.md @@ -20,6 +20,7 @@ ADR Sensor is a Python library that collects telemetry from AI coding agents to | **GitHub Copilot CLI** | `copilot` | JSONL (`~/.copilot/session-state/`) | macOS, Linux, Windows | | **Warp Terminal** | `warp` | SQLite (`warp.sqlite`) | macOS, Windows | | **opencode** | `opencode` | SQLite (`opencode.db`) or JSON tree | macOS, Linux | +| **Pi coding agent** | `pi` | JSONL (`~/.pi/agent/sessions/`) | macOS, Linux, Windows | ### Claude Desktop Agent Mode @@ -92,13 +93,68 @@ known built-in and contains an underscore is recorded as `tool_type: "mcp_tool"` its `server_name` populated. +### Pi coding agent + +The `pi` source reads Pi's persisted session JSONL files, including legacy v1 +linear sessions and v2/v3 tree-structured sessions. It captures conversations, +tool calls with full recorded arguments and results, tool failures, and user-run +shell commands. No content redaction or additional truncation is applied. + +| Operating system | Default session directory | +| ---------------- | ------------------------- | +| macOS | `/Users//.pi/agent/sessions/` | +| Linux | `/home//.pi/agent/sessions/` | +| Windows | `%USERPROFILE%\.pi\agent\sessions\` | + +Pi supports native Windows with Git Bash by default; its optional PowerShell tool +does not change session storage. ADR Sensor only reads the JSONL and does not +need either shell. Session directories are searched recursively. Set +`PI_CODING_AGENT_DIR` for both processes if Pi's agent directory was moved, or +`PI_CODING_AGENT_SESSION_DIR` to override the entire sessions root. If Pi is +started with `--session-dir`, set the same root through +`PI_CODING_AGENT_SESSION_DIR` for ADR Sensor, or pass `base_path` to `PiParser`. +An explicit parser `base_path` takes precedence over environment variables. + +The export is a forensic history of **all branches recorded in the file**, not +just the active model context. Tool results are matched to calls on their own +ancestor path, so reused call IDs on sibling branches are not mixed. If several +results refer to a shared ancestor call, its first result stays on the invocation +and additional results appear as `tool` messages, with `tool_call_entry_id` in +their metadata. Orphan results are preserved without inventing invocations. +User shell commands have role `user`, not `assistant`. + +`session_context.entries` retains entry IDs/parents, typed content (including +recorded thinking and images), provider/model details, stop reasons, full tool +result details, extension entries, labels, branch summaries, and compactions. +`token_usage.cumulative` sums recorded assistant, nested-tool, compaction, and +branch-summary usage. Raw usage and cost records remain in entry metadata. +Custom extension tools are reported as function calls: the session format does +not establish a universal MCP server identity, so the parser does not guess one. + +The default lookback is 14 days by file modification time; `--all-history` +disables it. Resumed sessions update existing `--save-sessions` exports rather +than producing duplicate files. Malformed JSONL rows are skipped and counted; +unsupported future session versions are skipped explicitly. There is no capture +for `--no-session`/in-memory runs, deleted files, or content Pi itself never wrote. +Pi can truncate shell output before persistence; the parser preserves its +`truncated`/`fullOutputPath` metadata but does not follow external output files. +This source does not collect global settings, credentials, or a tool inventory. + +Storage, format, and platform behavior were checked against Pi's +[session-format reference](https://github.com/earendil-works/pi/blob/71dca871bc80b6bc97be37f0ca3189399d651fff/packages/coding-agent/docs/session-format.md), +[directory configuration](https://github.com/earendil-works/pi/blob/71dca871bc80b6bc97be37f0ca3189399d651fff/packages/coding-agent/src/config.ts), +[session manager](https://github.com/earendil-works/pi/blob/71dca871bc80b6bc97be37f0ca3189399d651fff/packages/coding-agent/src/core/session-manager.ts), +and [Windows guide](https://github.com/earendil-works/pi/blob/71dca871bc80b6bc97be37f0ca3189399d651fff/packages/coding-agent/docs/windows.md). +CI exercises contract fixtures on native macOS, Linux, and Windows; these tests +do not run authenticated Pi model sessions. + ## Architecture ``` ┌─────────────────────────────────────────────────────────────────┐ │ AI Agent Logs │ │ Claude, Cursor, Cline, Codex, Copilot CLI, Warp │ -│ Claude Desktop, opencode │ +│ Claude Desktop, opencode, Pi │ └───────────────────────────────┬─────────────────────────────────┘ ▼ ┌─────────────────────────────────────────────────────────────────┐ @@ -155,6 +211,7 @@ adr-sensor --source codex adr-sensor --source copilot adr-sensor --source claude_desktop adr-sensor --source opencode +adr-sensor --source pi # Save individual session files (incremental) adr-sensor --save-sessions @@ -378,6 +435,8 @@ cannot run on the current platform are skipped rather than failing. | `XDG_CACHE_HOME` | `AgentObserver` | Base for `--save-sessions` output (`$XDG_CACHE_HOME/adr_sensor`, default `~/.cache/adr_sensor`) | | `XDG_DATA_HOME` | opencode parser | Overrides the opencode data directory (default `~/.local/share/opencode`) | | `OPENCODE_DB` | opencode parser | Overrides the opencode SQLite filename or path (`:memory:` is ignored) | +| `PI_CODING_AGENT_DIR` | Pi parser | Agent directory containing `sessions/` (default `~/.pi/agent`) | +| `PI_CODING_AGENT_SESSION_DIR` | Pi parser | Sessions root; takes precedence over `PI_CODING_AGENT_DIR` | | `APPDATA` | Cursor, Cline, Claude Desktop parsers | Windows roaming app-data root. Consulted first so redirected/roaming profiles resolve correctly (default `~/AppData/Roaming`) | | `LOCALAPPDATA` | Warp parser | Windows local app-data root, same redirected-profile handling (default `~/AppData/Local`) | @@ -428,6 +487,7 @@ adr-sensor/ │ │ ├── codex_parser.py │ │ ├── copilot_parser.py │ │ ├── opencode_parser.py +│ │ ├── pi_parser.py │ │ └── warp_parser.py │ ├── schemas/ │ │ ├── agent_event_schema.py # AgentEvent, ChatMessage, ToolUsage diff --git a/Sensor/adr_sensor/__init__.py b/Sensor/adr_sensor/__init__.py index f4d516f..6be1c41 100644 --- a/Sensor/adr_sensor/__init__.py +++ b/Sensor/adr_sensor/__init__.py @@ -3,7 +3,7 @@ Security observability library for AI coding agents. Collects telemetry from Claude Code, Cursor, Cline, OpenAI Codex CLI, GitHub Copilot CLI, Warp -Terminal, opencode, and Claude Desktop Agent Mode (including Dispatch +Terminal, opencode, Pi, and Claude Desktop Agent Mode (including Dispatch sessions) to enable threat detection and security monitoring. Usage: diff --git a/Sensor/adr_sensor/cli.py b/Sensor/adr_sensor/cli.py index bd2974c..ceb63f1 100644 --- a/Sensor/adr_sensor/cli.py +++ b/Sensor/adr_sensor/cli.py @@ -52,6 +52,7 @@ def main(): adr-sensor --source claude_desktop Ingest Claude Desktop agent-mode logs only (macOS/Windows) adr-sensor --source copilot Ingest GitHub Copilot CLI logs only adr-sensor --source opencode Ingest opencode logs only + adr-sensor --source pi Ingest Pi session conversations and tools adr-sensor --save-sessions Save individual session files adr-sensor --output-format jsonl Export as JSONL adr-sensor --all-history Include all logs (not just last 2 weeks) diff --git a/Sensor/adr_sensor/observer.py b/Sensor/adr_sensor/observer.py index fc416a4..6ebf87b 100644 --- a/Sensor/adr_sensor/observer.py +++ b/Sensor/adr_sensor/observer.py @@ -29,6 +29,7 @@ from .parsers.copilot_parser import CopilotParser from .parsers.cursor_parser import CursorParser from .parsers.opencode_parser import OpencodeParser +from .parsers.pi_parser import PiParser from .parsers.warp_parser import WarpParser from .schemas.agent_event_schema import AgentEvent from .schemas.system_config_schema import SystemConfiguration @@ -61,6 +62,7 @@ class AgentObserver: ("codex", "Codex"), ("copilot", "GitHub Copilot CLI"), ("opencode", "opencode"), + ("pi", "Pi"), ) #: Sources that only produce logs on some operating systems. A source absent @@ -69,7 +71,7 @@ class AgentObserver: "claude_desktop": ("Darwin", "Windows"), } - CONTENT_AWARE_INCREMENTAL_SOURCES = frozenset({"codex", "copilot"}) + CONTENT_AWARE_INCREMENTAL_SOURCES = frozenset({"codex", "copilot", "pi"}) def __init__(self, output_dir: Optional[Path] = None, max_age_days: Optional[int] = None): """Initialize the AgentObserver. @@ -92,6 +94,7 @@ def __init__(self, output_dir: Optional[Path] = None, max_age_days: Optional[int self.opencode_parser = ( OpencodeParser(max_age_days=max_age_days) if max_age_days is not None else OpencodeParser() ) + self.pi_parser = PiParser(max_age_days=max_age_days) if max_age_days is not None else PiParser() self.output_dir = output_dir if output_dir else Path("output") self.output_dir.mkdir(exist_ok=True) @@ -129,7 +132,7 @@ def ingest_all( Args: source_filter: Which source to ingest. One of 'all', 'claude', 'cursor', - 'claude_desktop', 'cline', 'warp', 'codex', 'copilot', 'opencode'. + 'claude_desktop', 'cline', 'warp', 'codex', 'copilot', 'opencode', 'pi'. Returns: Tuple of (agent_events, system_configs). diff --git a/Sensor/adr_sensor/parsers/__init__.py b/Sensor/adr_sensor/parsers/__init__.py index e99a362..1d6798c 100644 --- a/Sensor/adr_sensor/parsers/__init__.py +++ b/Sensor/adr_sensor/parsers/__init__.py @@ -12,6 +12,7 @@ from .copilot_parser import CopilotParser from .cursor_parser import CursorParser from .opencode_parser import OpencodeParser +from .pi_parser import PiParser from .warp_parser import WarpParser __all__ = [ @@ -23,5 +24,6 @@ "CopilotParser", "CursorParser", "OpencodeParser", + "PiParser", "WarpParser", ] diff --git a/Sensor/adr_sensor/parsers/pi_parser.py b/Sensor/adr_sensor/parsers/pi_parser.py new file mode 100644 index 0000000..f8f8c1b --- /dev/null +++ b/Sensor/adr_sensor/parsers/pi_parser.py @@ -0,0 +1,295 @@ +"""Read Pi's versioned local JSONL sessions, preserving every recorded branch.""" + +import json +import os +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional + +from ..schemas.agent_event_schema import AgentEvent, ChatMessage, ToolUsage +from ..utils.timestamp_utils import normalize_timestamp +from .base_parser import BaseParser + + +class PiParser(BaseParser): + """Normalize the v1-v3 session contract documented by Pi's session manager.""" + + def __init__(self, max_age_days: int = 14, base_path: Optional[Path] = None): + agent_override = os.environ.get("PI_CODING_AGENT_DIR") + agent_dir = Path(agent_override).expanduser() if agent_override else Path.home() / ".pi" / "agent" + session_override = os.environ.get("PI_CODING_AGENT_SESSION_DIR") + self.base_path = Path(session_override).expanduser() if session_override else agent_dir / "sessions" + if base_path is not None: + self.base_path = Path(base_path).expanduser() + self.max_age_days = max_age_days + + def parse_all(self) -> List[AgentEvent]: + entries: Dict[str, AgentEvent] = {} + if not self.base_path.is_dir(): + return [] + cutoff = (datetime.now(timezone.utc) - timedelta(days=self.max_age_days)).timestamp() + for path in sorted(self.base_path.rglob("*.jsonl")): + try: + if not path.is_file() or (self.max_age_days > 0 and path.stat().st_mtime < cutoff): + continue + entry = self.parse_file(path) + if entry is None or not entry.has_meaningful_content(): + continue + old = entries.get(entry.session_id) + if old is None or self._revision(entry) > self._revision(old): + entries[entry.session_id] = entry + except (OSError, ValueError) as exc: + print(f"[PI] Unable to read {path}: {exc}") + return list(entries.values()) + + @staticmethod + def _revision(entry: AgentEvent) -> tuple: + context = entry.session_context or {} + return normalize_timestamp(context["last_event_at"]), context["event_count"] + + @staticmethod + def _timestamp(value: Any) -> Optional[datetime]: + if value is None or isinstance(value, bool): + return None + try: + return normalize_timestamp(value) + except (TypeError, ValueError, OverflowError, OSError): + return None + + @staticmethod + def _text(content: Any) -> str: + if isinstance(content, str): + return content + if not isinstance(content, list): + return "" + return "\n".join( + part["text"] for part in content if isinstance(part, dict) and isinstance(part.get("text"), str) + ) + + def parse_file(self, path: Path) -> Optional[AgentEvent]: + records = [] + malformed = 0 + try: + modified_at = datetime.fromtimestamp(path.stat().st_mtime, timezone.utc) + with path.open(encoding="utf-8") as handle: + for line_number, line in enumerate(handle, 1): + if not line.strip(): + continue + try: + record = json.loads(line) + except json.JSONDecodeError: + malformed += 1 + continue + if isinstance(record, dict): + records.append((line_number, record)) + else: + malformed += 1 + except (OSError, UnicodeError) as exc: + print(f"[PI] Unable to parse {path}: {exc}") + return None + + if not records or records[0][1].get("type") != "session": + return None + header = records[0][1] + session_id = header.get("id") + version = header.get("version", 1) + if not isinstance(session_id, str) or not session_id: + return None + if type(version) is not int or version not in {1, 2, 3}: + print(f"[PI] Unsupported session version in {path}") + return None + + messages = [] + entry_metadata = [] + parents = {} + # Per-entry call tables avoid correlating identical call IDs on sibling branches. + calls_by_entry = {} + resolved_calls = set() + timestamps = [] + usage = {} + model = None + previous_id = None + + def add_usage(raw: Any) -> None: + if isinstance(raw, dict): + for key, value in raw.items(): + if isinstance(value, (int, float)) and not isinstance(value, bool) and value >= 0: + usage[key] = usage.get(key, 0) + value + + def find_call(parent: Any, call_id: Any) -> Optional[tuple]: + seen = set() + while isinstance(parent, str) and parent not in seen: + seen.add(parent) + calls = calls_by_entry.get(parent, {}) + if isinstance(call_id, str) and call_id in calls: + return parent, calls[call_id] + parent = parents.get(parent) + return None + + for line_number, record in records[1:]: + entry_id = record.get("id") + if not isinstance(entry_id, str): + entry_id = f"line-{line_number}" + parent = record.get("parentId") if version >= 2 else previous_id + parents[entry_id] = parent + previous_id = entry_id + timestamp = self._timestamp(record.get("timestamp")) + if timestamp: + timestamps.append(timestamp) + details = {k: v for k, v in record.items() if k != "message"} + details["entry_id"] = entry_id + entry_metadata.append(details) + kind = record.get("type") + if kind == "model_change" and isinstance(record.get("modelId"), str): + model = record["modelId"] + if kind in {"compaction", "branch_summary"}: + add_usage(record.get("usage")) + if kind == "custom_message": + content = self._text(record.get("content")) + if content: + messages.append({"role": "system", "content": content, "tools": [], "sequence_id": entry_id}) + continue + message = record.get("message") + if kind != "message" or not isinstance(message, dict): + continue + details["message_metadata"] = {k: v for k, v in message.items() if k != "content"} + content = message.get("content") + # Retain source-typed blocks, including recorded thinking and images. + if not isinstance(content, str): + details["content_parts"] = content + timestamp = self._timestamp(message.get("timestamp")) + if timestamp: + timestamps.append(timestamp) + role = message.get("role") + add_usage(message.get("usage")) + if role == "assistant" and isinstance(message.get("model"), str): + model = message["model"] + if role == "toolResult": + match = find_call(parent, message.get("toolCallId")) + result = self._text(content) + status = ( + "error" + if message.get("isError") is True + else ("success" if message.get("isError") is False else None) + ) + details["result_content"] = content + if match is not None: + call_entry_id, tool = match + details["tool_call_entry_id"] = call_entry_id + key = (call_entry_id, message["toolCallId"]) + if key not in resolved_calls: + tool.update(result=result, status=status, error=result if status == "error" else None) + resolved_calls.add(key) + continue + # A branch can reuse an ancestor call with a different result. + # Keep the first result on the invocation and emit subsequent + # results separately, without counting another invocation. + details["additional_tool_result"] = True + else: + details["orphan_tool_result"] = True + if result or isinstance(content, list): + messages.append({"role": "tool", "content": result, "tools": [], "sequence_id": entry_id}) + continue + if role == "bashExecution": + result = message.get("output") + result = result if isinstance(result, str) else "" + code = message.get("exitCode") + status = ( + "cancelled" + if message.get("cancelled") + else (("success" if code == 0 else "error") if type(code) is int else None) + ) + messages.append( + { + "role": "user", + "content": "", + "sequence_id": entry_id, + "tools": [ + { + "tool_name": "bash", + "tool_type": "terminal_command", + "arguments": {"command": message.get("command")}, + "result": result, + "status": status, + "error": result if status == "error" else None, + } + ], + } + ) + continue + if role not in {"user", "assistant", "custom", "hookMessage"}: + details["content"] = content + continue + tools = [] + if role == "assistant" and isinstance(content, list): + for part in content: + if not isinstance(part, dict) or part.get("type") != "toolCall": + continue + name = part.get("name") + if not isinstance(name, str): + continue + args = part.get("arguments", {}) + tool = { + "tool_name": name, + "tool_type": "function_call", + "arguments": args if isinstance(args, dict) else {"raw": args}, + "status": "pending", + } + tools.append(tool) + call_id = part.get("id") + if isinstance(call_id, str): + calls_by_entry.setdefault(entry_id, {})[call_id] = tool + text = self._text(content) + if text or tools or (isinstance(content, list) and content): + messages.append( + { + "role": role if role in {"user", "assistant"} else "system", + "content": text, + "tools": tools, + "sequence_id": entry_id, + } + ) + + history = [ + ChatMessage(**{**message, "tools": [ToolUsage(**tool) for tool in message["tools"]]}) + for message in messages + ] + if not history: + return None + timestamp = self._timestamp(header.get("timestamp")) or (min(timestamps) if timestamps else modified_at) + context = { + "last_event_at": (max(timestamps) if timestamps else timestamp).isoformat(), + "event_count": len(records), + "session_metadata": header, + "entries": entry_metadata, + "history_scope": "all_recorded_branches", + "malformed_records": malformed, + } + if malformed: + print(f"[PI] Skipped {malformed} malformed records in {path}") + return AgentEvent( + timestamp=timestamp, + source="pi", + session_id=f"pi_{session_id}", + project_path=header.get("cwd") if isinstance(header.get("cwd"), str) else None, + model=model, + chat_history=history, + raw_log_path=str(path), + session_context=context, + token_usage={ + "cumulative": { + target: usage[source] + for source, target in ( + ("input", "input_tokens"), + ("output", "output_tokens"), + ("cacheRead", "cached_input_tokens"), + ("cacheWrite", "cache_write_tokens"), + ("reasoning", "reasoning_output_tokens"), + ("totalTokens", "total_tokens"), + ) + if source in usage + } + } + if usage + else None, + ) diff --git a/Sensor/adr_sensor/schemas/agent_event_schema.py b/Sensor/adr_sensor/schemas/agent_event_schema.py index aef5965..867977c 100644 --- a/Sensor/adr_sensor/schemas/agent_event_schema.py +++ b/Sensor/adr_sensor/schemas/agent_event_schema.py @@ -1,7 +1,7 @@ """ Agent Event Schema for AI agent telemetry ingestion. Normalizes logs from Claude Code, Cursor, Cline, Codex, Copilot CLI, Warp, -opencode, and Claude Desktop into a common format for security analysis. +opencode, Pi, and Claude Desktop into a common format for security analysis. """ import hashlib @@ -46,7 +46,7 @@ class AgentEvent: # Core fields timestamp: datetime - source: str # claude, cursor, cline, warp, codex, copilot, claude_desktop, opencode + source: str # claude, cursor, cline, warp, codex, copilot, claude_desktop, opencode, pi session_id: str # Chat history diff --git a/Sensor/tests/test_pi_parser.py b/Sensor/tests/test_pi_parser.py new file mode 100644 index 0000000..277920e --- /dev/null +++ b/Sensor/tests/test_pi_parser.py @@ -0,0 +1,250 @@ +"""Contract fixtures for Pi v1-v3 session files; no account or local agent required.""" + +import json +import os +from pathlib import Path + +import pytest + +from adr_sensor.observer import AgentObserver +from adr_sensor.parsers.pi_parser import PiParser + + +def header(version=3, session_id="example"): + return { + "type": "session", + "version": version, + "id": session_id, + "timestamp": "2026-01-01T00:00:00Z", + "cwd": "C:\\work\\project", + } + + +def message(entry_id, parent, role, content, **kwargs): + return { + "type": "message", + "id": entry_id, + "parentId": parent, + "timestamp": "2026-01-01T00:00:01Z", + "message": {"role": role, "content": content, **kwargs}, + } + + +def call(entry_id="a", parent="u", call_id="c"): + return message( + entry_id, + parent, + "assistant", + [ + {"type": "text", "text": "Reading file"}, + {"type": "toolCall", "id": call_id, "name": "read", "arguments": {"path": "file.txt"}}, + ], + model="model-test", + provider="provider-test", + usage={"input": 10, "output": 2, "totalTokens": 12}, + ) + + +def result(entry_id="r", parent="a", text="result", call_id="c", **kwargs): + return message( + entry_id, + parent, + "toolResult", + [{"type": "text", "text": text}], + toolCallId=call_id, + toolName="read", + isError=False, + **kwargs, + ) + + +def write(path, rows): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("\n".join(json.dumps(row) for row in rows) + "\n", encoding="utf-8") + return path + + +def test_home_and_override_precedence(tmp_path, monkeypatch): + monkeypatch.setattr(Path, "home", lambda: tmp_path) + monkeypatch.setenv("PI_CODING_AGENT_DIR", "") + monkeypatch.setenv("PI_CODING_AGENT_SESSION_DIR", "") + assert PiParser().base_path == tmp_path / ".pi/agent/sessions" + monkeypatch.setenv("PI_CODING_AGENT_DIR", str(tmp_path / "agent")) + assert PiParser().base_path == tmp_path / "agent/sessions" + monkeypatch.setenv("PI_CODING_AGENT_SESSION_DIR", str(tmp_path / "sessions")) + assert PiParser().base_path == tmp_path / "sessions" + assert PiParser(base_path=tmp_path / "explicit").base_path == tmp_path / "explicit" + + +@pytest.mark.parametrize("version", [1, 2, 3]) +def test_versions_tools_usage_and_read_only(tmp_path, version): + rows = [ + header(version), + message("u", None, "user", "Inspect the project"), + call(), + result(text="secret:" + "x" * 4000), + ] + if version == 1: + for row in rows[1:]: + row.pop("id") + row.pop("parentId") + path = write(tmp_path / "--project--/session.jsonl", rows) + before = path.read_bytes(), path.stat().st_mtime_ns + entry = PiParser(base_path=tmp_path).parse_all()[0] + assert entry.source == "pi" and entry.project_path == "C:\\work\\project" + assert len(entry.chat_history) == 2 + tool = entry.chat_history[1].tools[0] + assert tool.result == "secret:" + "x" * 4000 + assert tool.status == "success" and tool.arguments == {"path": "file.txt"} + assert entry.model == "model-test" + assert entry.token_usage["cumulative"]["total_tokens"] == 12 + assert (path.read_bytes(), path.stat().st_mtime_ns) == before + + +def test_sibling_branch_call_ids_are_not_cross_correlated(tmp_path): + rows = [ + header(), + message("u", None, "user", "Inspect the project"), + call("a", "u", "same-call"), + call("b", "u", "same-call"), + result("ra", "a", "branch A", "same-call"), + result("rb", "b", "branch B", "same-call"), + ] + entry = PiParser().parse_file(write(tmp_path / "session.jsonl", rows)) + tools = [msg.tools[0] for msg in entry.chat_history if msg.tools] + assert [tool.result for tool in tools] == ["branch A", "branch B"] + assert entry.session_context["history_scope"] == "all_recorded_branches" + assert entry.session_context["entries"][2]["parentId"] == "u" + + +def test_metadata_reasoning_images_and_custom_messages(tmp_path): + rows = [header(), message("u", None, "user", "Inspect the project"), call()] + rows[-1]["message"]["content"].extend( + [{"type": "thinking", "thinking": "Recorded plan"}, {"type": "image", "data": "abc", "mimeType": "image/png"}] + ) + rows += [ + {"type": "compaction", "id": "compact", "parentId": "a", "summary": "Summary", "usage": {"totalTokens": 5}}, + {"type": "branch_summary", "id": "branch", "parentId": "u", "fromId": "compact", "summary": "Old branch"}, + {"type": "custom_message", "id": "custom", "parentId": "branch", "content": "Extension instructions"}, + {"type": "model_change", "id": "model", "parentId": "custom", "modelId": "next-model"}, + ] + entry = PiParser().parse_file(write(tmp_path / "session.jsonl", rows)) + assert entry.chat_history[-1].role == "system" + assert entry.model == "next-model" + assert entry.token_usage["cumulative"]["total_tokens"] == 17 + assert entry.session_context["entries"][1]["content_parts"][-2]["thinking"] == "Recorded plan" + assert entry.session_context["entries"][2]["summary"] == "Summary" + assert entry.session_context["entries"][3]["fromId"] == "compact" + + +def test_multiple_results_for_shared_ancestor_preserve_first_and_additional_results(tmp_path): + rows = [ + header(), + message("u", None, "user", "Inspect the project"), + call(), + result("first", "a", "original result"), + result("second", "a", "alternate result"), + ] + entry = PiParser().parse_file(write(tmp_path / "session.jsonl", rows)) + tools = [tool for msg in entry.chat_history for tool in msg.tools] + assert len(tools) == 1 and tools[0].result == "original result" + assert entry.chat_history[-1].role == "tool" + assert entry.chat_history[-1].content == "alternate result" + metadata = entry.session_context["entries"][-1] + assert metadata["additional_tool_result"] is True + assert metadata["tool_call_entry_id"] == "a" + assert metadata["parentId"] == "a" + + +def test_legacy_extension_messages_and_image_only_content_are_retained(tmp_path): + rows = [ + header(2), + message("u", None, "user", [{"type": "image", "data": "abc", "mimeType": "image/png"}]), + message("hook", "u", "hookMessage", "Extension instructions", customType="extension"), + message("a", "hook", "assistant", [{"type": "thinking", "thinking": "Recorded thought"}]), + ] + entry = PiParser(base_path=tmp_path).parse_file(write(tmp_path / "session.jsonl", rows)) + assert [msg.role for msg in entry.chat_history] == ["user", "system", "assistant"] + assert entry.session_context["entries"][0]["content_parts"][0]["data"] == "abc" + assert entry.session_context["entries"][1]["message_metadata"]["customType"] == "extension" + assert entry.session_context["entries"][2]["content_parts"][0]["thinking"] == "Recorded thought" + + +def test_nested_usage_and_compaction_are_counted_once(tmp_path): + rows = [ + header(), + call(parent=None), + result(usage={"input": 4, "output": 2, "cacheRead": 3, "totalTokens": 9, "cost": {"total": 0.1}}), + {"type": "compaction", "id": "comp", "parentId": "r", "summary": "Compact", "usage": {"totalTokens": 5}}, + {"type": "branch_summary", "id": "b", "parentId": None, "summary": "Branch", "usage": {"totalTokens": 6}}, + ] + entry = PiParser().parse_file(write(tmp_path / "session.jsonl", rows)) + assert entry.token_usage["cumulative"] == { + "input_tokens": 14, + "output_tokens": 4, + "cached_input_tokens": 3, + "total_tokens": 32, + } + assert entry.session_context["entries"][1]["message_metadata"]["usage"]["cost"]["total"] == 0.1 + + +def test_errors_pending_orphan_and_user_shell_attribution(tmp_path): + failed = result(text="Access denied") + failed["message"]["isError"] = True + rows = [ + header(), + message("u", None, "user", "Inspect project"), + call(), + failed, + call("pending", "r", "pending-call"), + result("orphan", "u", "Detached result", "missing"), + message("shell", "orphan", "bashExecution", None, command="pwd", output="/work", exitCode=0, cancelled=False), + ] + entry = PiParser().parse_file(write(tmp_path / "session.jsonl", rows)) + assert entry.chat_history[1].tools[0].error == "Access denied" + assert entry.chat_history[2].tools[0].status == "pending" + assert entry.chat_history[3].role == "tool" and not entry.chat_history[3].tools + assert entry.chat_history[4].role == "user" + assert entry.chat_history[4].tools[0].tool_type == "terminal_command" + assert entry.chat_history[4].tools[0].result == "/work" + + +def test_bad_files_and_unrecognized_versions_do_not_block_other_sessions(tmp_path): + write(tmp_path / "good.jsonl", [header(), message("u", None, "user", "Inspect project")]) + write(tmp_path / "future.jsonl", [header(99), message("u", None, "user", "Inspect project")]) + write(tmp_path / "wrong.jsonl", [{"type": "unrelated", "content": "Do not ingest"}]) + path = tmp_path / "good.jsonl" + with path.open("a", encoding="utf-8") as handle: + handle.write('[]\n{"partial":') + entries = PiParser(base_path=tmp_path).parse_all() + assert len(entries) == 1 + assert entries[0].session_context["malformed_records"] == 2 + assert PiParser(base_path=tmp_path / "missing").parse_all() == [] + + +def test_age_filter_and_deduplicate_copies(tmp_path): + rows = [header(), message("u", None, "user", "Inspect project")] + path = write(tmp_path / "old/session.jsonl", rows) + os.utime(path, (1, 1)) + assert PiParser(base_path=tmp_path).parse_all() == [] + assert len(PiParser(max_age_days=0, base_path=tmp_path).parse_all()) == 1 + write(tmp_path / "copy/session.jsonl", rows) + assert len(PiParser(max_age_days=0, base_path=tmp_path).parse_all()) == 1 + + +def test_resumed_export_updates_without_duplicate_files(tmp_path): + path = write(tmp_path / "input/session.jsonl", [header(), message("u", None, "user", "Inspect project"), call()]) + observer = AgentObserver(output_dir=tmp_path / "output", max_age_days=0) + assert observer.pi_parser.max_age_days == 0 + observer.pi_parser = PiParser(base_path=tmp_path / "input") + event = observer.ingest_all("pi")[0][0] + saved = observer.save_sessions_to_individual_files([event], tmp_path / "output") + assert observer.filter_entries_by_existing_files([event], tmp_path / "output") == [] + with path.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(result()) + "\n") + resumed = observer.ingest_all("pi")[0][0] + assert resumed.timestamp == event.timestamp + selected = observer.filter_entries_by_existing_files([resumed], tmp_path / "output") + assert len(selected) == 1 + assert observer.save_sessions_to_individual_files(selected, tmp_path / "output") == saved + assert json.loads(saved[0].read_text(encoding="utf-8"))["chat_history"][1]["tools"][0]["result"] == "result"