From bdcf83a53a06030041a1139e457766d4209e396d Mon Sep 17 00:00:00 2001 From: "Li Zexian (External - Orange CyberDefense Sweden AB)" Date: Wed, 12 Aug 2026 12:01:41 +0800 Subject: [PATCH 1/3] Add GitHub Copilot parser and fix incremental timestamps for resumed sessions --- Sensor/README.md | 3 + Sensor/adr_sensor/observer.py | 5 +- Sensor/adr_sensor/parsers/__init__.py | 2 + Sensor/adr_sensor/parsers/claude_parser.py | 2 +- Sensor/adr_sensor/parsers/codex_parser.py | 10 +- Sensor/adr_sensor/parsers/copilot_parser.py | 589 ++++++++++++++++++++ Sensor/tests/test_observer.py | 2 + Sensor/tests/test_parsers.py | 165 ++++++ 8 files changed, 775 insertions(+), 3 deletions(-) create mode 100644 Sensor/adr_sensor/parsers/copilot_parser.py diff --git a/Sensor/README.md b/Sensor/README.md index bd81ecf..6d79f02 100644 --- a/Sensor/README.md +++ b/Sensor/README.md @@ -17,6 +17,7 @@ ADR Sensor is a Python library that collects telemetry from AI coding agents to | **Cline (Claude Dev)** | `cline` | JSON task files | macOS, Linux, Windows | | **Claude Desktop** | `claude_desktop` | JSONL audit logs | macOS, Windows | | **OpenAI Codex CLI** | `codex` | JSONL (`~/.codex/sessions/`) | macOS, Linux, Windows | +| **GitHub Copilot** | `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 | @@ -110,6 +111,7 @@ adr-sensor adr-sensor --source claude adr-sensor --source cursor adr-sensor --source codex +adr-sensor --source copilot adr-sensor --source claude_desktop adr-sensor --source opencode @@ -349,6 +351,7 @@ adr-sensor/ │ │ ├── cline_parser.py │ │ ├── claude_desktop_parser.py │ │ ├── codex_parser.py +│ │ ├── copilot_parser.py │ │ ├── opencode_parser.py │ │ └── warp_parser.py │ ├── schemas/ diff --git a/Sensor/adr_sensor/observer.py b/Sensor/adr_sensor/observer.py index 97e2109..514f22b 100644 --- a/Sensor/adr_sensor/observer.py +++ b/Sensor/adr_sensor/observer.py @@ -19,6 +19,7 @@ from .parsers.claude_desktop_parser import ClaudeDesktopParser from .parsers.claude_parser import ClaudeParser from .parsers.cline_parser import ClineParser +from .parsers.copilot_parser import CopilotParser from .parsers.codex_parser import CodexParser from .parsers.cursor_parser import CursorParser from .parsers.opencode_parser import OpencodeParser @@ -50,6 +51,7 @@ class AgentObserver: ("cline", "Cline"), ("warp", "Warp Terminal"), ("codex", "Codex"), + ("copilot", "GitHub Copilot"), ("opencode", "opencode"), ) @@ -72,6 +74,7 @@ def __init__(self, output_dir: Optional[Path] = None, max_age_days: Optional[int ClaudeDesktopParser(max_age_days=max_age_days) if max_age_days is not None else ClaudeDesktopParser() ) self.codex_parser = CodexParser() + self.copilot_parser = CopilotParser() self.cline_parser = ClineParser() self.warp_parser = WarpParser(max_age_days=max_age_days) if max_age_days is not None else WarpParser() self.opencode_parser = ( @@ -114,7 +117,7 @@ def ingest_all( Args: source_filter: Which source to ingest. One of 'all', 'claude', 'cursor', - 'claude_desktop', 'cline', 'warp', 'codex', 'opencode'. + 'claude_desktop', 'cline', 'warp', 'codex', 'copilot', 'opencode'. Returns: Tuple of (agent_events, system_configs). diff --git a/Sensor/adr_sensor/parsers/__init__.py b/Sensor/adr_sensor/parsers/__init__.py index a3b1281..c94fff0 100644 --- a/Sensor/adr_sensor/parsers/__init__.py +++ b/Sensor/adr_sensor/parsers/__init__.py @@ -8,6 +8,7 @@ from .claude_desktop_parser import ClaudeDesktopParser from .claude_parser import ClaudeParser from .cline_parser import ClineParser +from .copilot_parser import CopilotParser from .codex_parser import CodexParser from .cursor_parser import CursorParser from .opencode_parser import OpencodeParser @@ -18,6 +19,7 @@ "ClaudeDesktopParser", "ClaudeParser", "ClineParser", + "CopilotParser", "CodexParser", "CursorParser", "OpencodeParser", diff --git a/Sensor/adr_sensor/parsers/claude_parser.py b/Sensor/adr_sensor/parsers/claude_parser.py index 856de38..05000b7 100644 --- a/Sensor/adr_sensor/parsers/claude_parser.py +++ b/Sensor/adr_sensor/parsers/claude_parser.py @@ -127,7 +127,7 @@ def parse_jsonl_file(self, file_path: Path) -> List[AgentEvent]: if "timestamp" in obj: try: ts = normalize_timestamp(obj["timestamp"]) - if sessions[session_id]["timestamp"] is None or ts < sessions[session_id]["timestamp"]: + if sessions[session_id]["timestamp"] is None or ts > sessions[session_id]["timestamp"]: sessions[session_id]["timestamp"] = ts except Exception: pass diff --git a/Sensor/adr_sensor/parsers/codex_parser.py b/Sensor/adr_sensor/parsers/codex_parser.py index 4987c95..42c430f 100644 --- a/Sensor/adr_sensor/parsers/codex_parser.py +++ b/Sensor/adr_sensor/parsers/codex_parser.py @@ -48,6 +48,7 @@ def parse_jsonl_file(self, file_path: Path) -> Optional[AgentEvent]: session_data: Dict[str, Any] = { "id": None, "timestamp": None, + "last_event_timestamp": None, "cwd": None, "model": None, "messages": [], @@ -94,7 +95,7 @@ def parse_jsonl_file(self, file_path: Path) -> Optional[AgentEvent]: ) return AgentEvent( - timestamp=session_data["timestamp"] or datetime.now(timezone.utc), + timestamp=session_data["last_event_timestamp"] or session_data["timestamp"] or datetime.now(timezone.utc), source="codex", session_id=f"codex_{session_data['id']}", project_path=session_data["cwd"], @@ -174,6 +175,13 @@ def _process_event(self, event: Dict[str, Any], session_data: Dict[str, Any]): evt_type = event.get("type") payload = event.get("payload", {}) + event_timestamp = event.get("timestamp") + if event_timestamp: + normalized = normalize_timestamp(event_timestamp) + current = session_data.get("last_event_timestamp") + if current is None or normalized > current: + session_data["last_event_timestamp"] = normalized + if evt_type == "session_meta": session_data["id"] = payload.get("id") if payload.get("timestamp"): diff --git a/Sensor/adr_sensor/parsers/copilot_parser.py b/Sensor/adr_sensor/parsers/copilot_parser.py new file mode 100644 index 0000000..a39845d --- /dev/null +++ b/Sensor/adr_sensor/parsers/copilot_parser.py @@ -0,0 +1,589 @@ +""" +Parser for GitHub Copilot session-state logs. + +Reads per-session event streams from ``~/.copilot/session-state//`` +and normalizes them into ADR's ``AgentEvent`` schema. + +The primary signal lives in ``events.jsonl``. When available, the parser also +enriches sessions with lightweight metadata from ``workspace.yaml`` and +``vscode.metadata.json``. +""" + +import json +import traceback +from collections import Counter +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional + +from ..schemas.agent_event_schema import AgentEvent, ChatMessage, ToolUsage +from ..utils.string_utils import truncate_middle +from ..utils.timestamp_utils import normalize_timestamp +from .base_parser import BaseParser + +MAX_STRING_LENGTH = 1000 +EDGE_CHARS = 400 + + +class CopilotParser(BaseParser): + """Parser for GitHub Copilot session-state logs.""" + + def __init__(self, base_path: Optional[Path] = None): + self.base_path = Path(base_path) if base_path else Path.home() / ".copilot" / "session-state" + + def parse_all(self) -> List[AgentEvent]: + """Parse all available Copilot sessions.""" + entries: List[AgentEvent] = [] + + if not self.base_path.exists(): + print(f"[COPILOT] No logs found at {self.base_path}") + return entries + + session_dirs = [path for path in self.base_path.iterdir() if path.is_dir() and (path / "events.jsonl").exists()] + session_dirs.sort(key=lambda path: path.stat().st_mtime, reverse=True) + print(f"[COPILOT] Found {len(session_dirs)} session directories") + + for session_dir in session_dirs: + try: + entry = self.parse_session_dir(session_dir) + if entry and entry.has_meaningful_content(): + entries.append(entry) + except Exception as exc: + print(f"[COPILOT] Error parsing {session_dir}: {exc}") + + return entries + + def parse_session_dir(self, session_dir: Path) -> Optional[AgentEvent]: + """Parse one Copilot session directory.""" + events_path = session_dir / "events.jsonl" + if not events_path.exists(): + return None + + workspace_meta = self._load_workspace_yaml(session_dir / "workspace.yaml") + vscode_meta = self._load_json_file(session_dir / "vscode.metadata.json") + + session_data: Dict[str, Any] = { + "id": session_dir.name, + "timestamp": None, + "cwd": workspace_meta.get("cwd"), + "model": None, + "messages": [], + "pending_tool_calls": {}, + "first_event_at": None, + "last_event_at": None, + "event_counts": Counter(), + "workspace_metadata": workspace_meta, + "vscode_metadata": self._sanitize_for_context(vscode_meta) if isinstance(vscode_meta, dict) else {}, + "model_changes": [], + "usage_checkpoints": [], + "permissions": [], + "skills_invoked": [], + "subagents": [], + "session_info": [], + "mode_changes": [], + "hooks": [], + "system_messages": [], + "system_notifications": [], + "workspace_file_changes": [], + "plan_changes": [], + "transformed_user_messages": [], + } + + try: + with open(events_path, encoding="utf-8") as handle: + for line in handle: + line = line.strip() + if not line: + continue + try: + event = json.loads(line) + except json.JSONDecodeError: + continue + self._process_event(event, session_data) + except Exception as exc: + print(f"[COPILOT] Error reading {events_path}: {exc}") + traceback.print_exc() + return None + + chat_history: List[ChatMessage] = [] + for index, msg_dict in enumerate(session_data["messages"]): + tools = [ + ToolUsage( + tool_name=tool_dict["tool_name"], + tool_type=tool_dict["tool_type"], + arguments=tool_dict["arguments"], + result=tool_dict.get("result"), + status=tool_dict.get("status"), + error=tool_dict.get("error"), + ) + for tool_dict in msg_dict["tools"] + ] + + chat_history.append( + ChatMessage( + role=msg_dict["role"], + content=msg_dict["content"], + tools=tools, + sequence_id=msg_dict.get("sequence_id") or f"{session_data['id']}_msg_{index}", + ) + ) + + if not chat_history: + return None + + timestamp = ( + self._normalize_optional_timestamp(workspace_meta.get("updated_at")) + or self._normalize_optional_timestamp(vscode_meta.get("modified")) + or self._normalize_optional_timestamp(session_data["last_event_at"]) + or session_data["timestamp"] + or self._normalize_optional_timestamp(workspace_meta.get("created_at")) + or self._normalize_optional_timestamp(vscode_meta.get("created")) + or self._normalize_optional_timestamp(session_data["first_event_at"]) + or datetime.now(timezone.utc) + ) + + context = self._build_session_context(session_data) + + return AgentEvent( + timestamp=timestamp, + source="copilot", + session_id=f"copilot_{session_data['id']}", + project_path=session_data["cwd"], + model=session_data["model"], + chat_history=chat_history, + raw_log_path=str(events_path), + session_context=context or None, + ) + + def _process_event(self, event: Dict[str, Any], session_data: Dict[str, Any]): + """Process a single Copilot event from events.jsonl.""" + event_type = event.get("type") + if not event_type: + return + + data = event.get("data", {}) + event_time = self._normalize_optional_timestamp(event.get("timestamp")) + + session_data["event_counts"][event_type] += 1 + if event_time and session_data["first_event_at"] is None: + session_data["first_event_at"] = event_time + if event_time: + session_data["last_event_at"] = event_time + + if event_type == "session.start": + session_id = data.get("sessionId") + if session_id: + session_data["id"] = session_id + session_data["timestamp"] = self._normalize_optional_timestamp(data.get("startTime")) or event_time + session_data["model"] = data.get("selectedModel") or session_data["model"] + context = data.get("context", {}) + if isinstance(context, dict): + session_data["cwd"] = context.get("cwd") or session_data["cwd"] + return + + if event_type == "session.resume": + session_id = data.get("sessionId") + if session_id: + session_data["id"] = session_id + context = data.get("context", {}) + if isinstance(context, dict): + session_data["cwd"] = context.get("cwd") or session_data["cwd"] + return + + if event_type == "user.message": + content = (data.get("content") or "").strip() + transformed = (data.get("transformedContent") or "").strip() + if transformed and transformed != content: + session_data["transformed_user_messages"].append( + { + "sequence_id": event.get("id"), + "content": truncate_middle(transformed, max_length=MAX_STRING_LENGTH, edge_chars=EDGE_CHARS), + } + ) + if content: + session_data["messages"].append( + { + "role": "user", + "content": content, + "tools": [], + "sequence_id": event.get("id"), + } + ) + return + + if event_type == "assistant.message": + content = (data.get("content") or "").strip() + tools = [] + for request in data.get("toolRequests") or []: + tool_call_id = request.get("toolCallId") + tool_dict = { + "tool_name": request.get("name") or "unknown", + "tool_type": request.get("type") or "tool_request", + "arguments": self._truncate_large_arguments(request.get("arguments") or {}), + "result": None, + "status": "pending", + "error": None, + } + tools.append(tool_dict) + if tool_call_id: + session_data["pending_tool_calls"][tool_call_id] = tool_dict + + if content or tools: + session_data["messages"].append( + { + "role": "assistant", + "content": content, + "tools": tools, + "sequence_id": data.get("messageId") or event.get("id"), + } + ) + + session_data["model"] = data.get("model") or session_data["model"] + return + + if event_type == "tool.execution_start": + tool = self._get_or_create_tool( + session_data, + data.get("toolCallId"), + default_tool_name=data.get("toolName"), + ) + if tool is not None: + tool["tool_name"] = data.get("toolName") or tool["tool_name"] + tool["arguments"] = self._truncate_large_arguments(data.get("arguments") or tool["arguments"]) + tool["status"] = "running" + session_data["model"] = data.get("model") or session_data["model"] + return + + if event_type == "tool.execution_complete": + tool = self._get_or_create_tool(session_data, data.get("toolCallId"), default_tool_name=None) + if tool is not None: + success = bool(data.get("success")) + result_text = self._normalize_tool_result(data.get("result")) + tool["result"] = result_text + tool["status"] = "success" if success else "error" + if not success and result_text: + tool["error"] = result_text + session_data["model"] = data.get("model") or session_data["model"] + return + + if event_type == "permission.requested": + permission_record = { + "request_id": data.get("requestId"), + "tool_call_id": self._extract_tool_call_id_from_permission(data), + "permission_kind": self._extract_permission_kind(data), + "command": self._extract_permission_command(data), + "timestamp": event.get("timestamp"), + } + session_data["permissions"].append(self._sanitize_for_context(permission_record)) + + tool = self._get_or_create_tool( + session_data, + permission_record["tool_call_id"], + default_tool_name="permission_request", + ) + if tool is not None and tool.get("status") == "pending": + tool["status"] = "permission_requested" + return + + if event_type == "permission.completed": + permission_result = data.get("result", {}) + permission_record = { + "request_id": data.get("requestId"), + "tool_call_id": data.get("toolCallId"), + "result": permission_result, + "timestamp": event.get("timestamp"), + } + session_data["permissions"].append(self._sanitize_for_context(permission_record)) + + tool = self._get_or_create_tool(session_data, data.get("toolCallId"), default_tool_name=None) + kind = permission_result.get("kind") if isinstance(permission_result, dict) else None + if tool is not None and tool.get("status") in {"pending", "permission_requested"} and kind == "approved": + tool["status"] = "approved" + return + + if event_type == "session.model_change": + session_data["model_changes"].append( + self._sanitize_for_context( + { + "previous_model": data.get("previousModel"), + "new_model": data.get("newModel"), + "previous_reasoning_effort": data.get("previousReasoningEffort"), + "reasoning_effort": data.get("reasoningEffort"), + "timestamp": event.get("timestamp"), + } + ) + ) + session_data["model"] = data.get("newModel") or session_data["model"] + return + + if event_type == "session.usage_checkpoint": + session_data["usage_checkpoints"].append(self._sanitize_for_context(data)) + return + + if event_type == "skill.invoked": + session_data["skills_invoked"].append( + self._sanitize_for_context( + { + "name": data.get("name"), + "path": data.get("path"), + "content": data.get("content"), + } + ) + ) + return + + if event_type == "subagent.started": + session_data["subagents"].append( + self._sanitize_for_context( + { + "agent_id": event.get("agentId"), + "tool_call_id": data.get("toolCallId"), + "agent_name": data.get("agentName"), + "display_name": data.get("agentDisplayName"), + "description": data.get("agentDescription"), + "timestamp": event.get("timestamp"), + } + ) + ) + return + + if event_type == "session.info": + session_data["session_info"].append(self._sanitize_for_context(data)) + return + + if event_type == "session.mode_changed": + session_data["mode_changes"].append(self._sanitize_for_context(data)) + return + + if event_type in {"hook.start", "hook.end"}: + session_data["hooks"].append( + self._sanitize_for_context( + { + "type": event_type, + "timestamp": event.get("timestamp"), + "data": data, + } + ) + ) + return + + if event_type == "system.message": + content = data.get("content") + if isinstance(content, str) and content.strip(): + session_data["system_messages"].append( + truncate_middle(content.strip(), max_length=MAX_STRING_LENGTH, edge_chars=EDGE_CHARS) + ) + return + + if event_type == "system.notification": + session_data["system_notifications"].append(self._sanitize_for_context(data)) + return + + if event_type == "session.workspace_file_changed": + session_data["workspace_file_changes"].append(self._sanitize_for_context(data)) + return + + if event_type == "session.plan_changed": + session_data["plan_changes"].append(self._sanitize_for_context(data)) + + def _get_or_create_tool( + self, + session_data: Dict[str, Any], + tool_call_id: Optional[str], + default_tool_name: Optional[str], + ) -> Optional[Dict[str, Any]]: + """Return a mutable tool dict for a toolCallId, creating an orphan entry if needed.""" + if not tool_call_id: + return None + + existing = session_data["pending_tool_calls"].get(tool_call_id) + if existing is not None: + return existing + + tool_dict = { + "tool_name": default_tool_name or "unknown", + "tool_type": "tool_execution", + "arguments": {}, + "result": None, + "status": "pending", + "error": None, + } + session_data["pending_tool_calls"][tool_call_id] = tool_dict + session_data["messages"].append( + { + "role": "assistant", + "content": "", + "tools": [tool_dict], + "sequence_id": f"{session_data['id']}_tool_{tool_call_id}", + } + ) + return tool_dict + + def _build_session_context(self, session_data: Dict[str, Any]) -> Dict[str, Any]: + """Build session-level context from side-channel event metadata.""" + context = { + "workspace_metadata": session_data["workspace_metadata"], + "vscode_metadata": session_data["vscode_metadata"], + "first_event_at": self._format_datetime(session_data["first_event_at"]), + "last_event_at": self._format_datetime(session_data["last_event_at"]), + "event_counts": dict(session_data["event_counts"]), + "model_changes": session_data["model_changes"], + "usage_checkpoints": session_data["usage_checkpoints"], + "permissions": session_data["permissions"], + "skills_invoked": session_data["skills_invoked"], + "subagents": session_data["subagents"], + "session_info": session_data["session_info"], + "mode_changes": session_data["mode_changes"], + "hooks": session_data["hooks"], + "system_messages": session_data["system_messages"][:5], + "system_notifications": session_data["system_notifications"], + "workspace_file_changes": session_data["workspace_file_changes"], + "plan_changes": session_data["plan_changes"], + "transformed_user_messages": session_data["transformed_user_messages"], + } + + filtered: Dict[str, Any] = {} + for key, value in context.items(): + if isinstance(value, dict) and value: + filtered[key] = value + elif isinstance(value, list) and value: + filtered[key] = value + elif isinstance(value, str) and value: + filtered[key] = value + return filtered + + def _normalize_tool_result(self, result: Any) -> Optional[str]: + """Normalize tool result payloads to a truncated string.""" + if result is None: + return None + + if isinstance(result, str): + text = result + elif isinstance(result, dict): + text = result.get("detailedContent") or result.get("content") or json.dumps(result, ensure_ascii=False) + else: + text = str(result) + + if not text: + return text + + return truncate_middle(text, max_length=MAX_STRING_LENGTH, edge_chars=EDGE_CHARS) + + def _truncate_large_arguments(self, arguments: Dict[str, Any]) -> Dict[str, Any]: + """Truncate large string values within tool argument dictionaries.""" + if not isinstance(arguments, dict): + return {"raw": arguments} + + return {key: self._sanitize_for_context(value) for key, value in arguments.items()} + + def _sanitize_for_context(self, value: Any) -> Any: + """Recursively truncate long strings so session_context stays bounded.""" + if isinstance(value, str): + return truncate_middle(value, max_length=MAX_STRING_LENGTH, edge_chars=EDGE_CHARS) + if isinstance(value, dict): + return {str(key): self._sanitize_for_context(val) for key, val in value.items()} + if isinstance(value, list): + return [self._sanitize_for_context(item) for item in value] + return value + + def _load_json_file(self, path: Path) -> Dict[str, Any]: + """Read a JSON file if it exists, else return an empty dict.""" + if not path.exists(): + return {} + + try: + with open(path, encoding="utf-8") as handle: + value = json.load(handle) + return value if isinstance(value, dict) else {} + except Exception: + return {} + + def _load_workspace_yaml(self, path: Path) -> Dict[str, Any]: + """Parse the simple key/value workspace.yaml emitted by Copilot.""" + if not path.exists(): + return {} + + data: Dict[str, Any] = {} + try: + with open(path, encoding="utf-8") as handle: + for raw_line in handle: + line = raw_line.strip() + if not line or ":" not in line: + continue + key, value = line.split(":", 1) + data[key.strip()] = self._coerce_scalar(value.strip()) + except Exception: + return {} + return data + + def _coerce_scalar(self, value: str) -> Any: + """Coerce simple YAML scalars without pulling in a YAML dependency.""" + if value == "": + return "" + if value.lower() == "true": + return True + if value.lower() == "false": + return False + try: + return int(value) + except ValueError: + return value + + def _normalize_optional_timestamp(self, value: Any) -> Optional[datetime]: + """Normalize a timestamp when possible.""" + if value in (None, ""): + return None + try: + return normalize_timestamp(value) + except Exception: + return None + + @staticmethod + def _format_datetime(value: Optional[datetime]) -> Optional[str]: + """Format datetimes for session_context.""" + return value.isoformat() if isinstance(value, datetime) else None + + @staticmethod + def _extract_tool_call_id_from_permission(data: Dict[str, Any]) -> Optional[str]: + """Extract toolCallId from the permission event payload.""" + if data.get("toolCallId"): + return data.get("toolCallId") + + permission_request = data.get("permissionRequest", {}) + if isinstance(permission_request, dict) and permission_request.get("toolCallId"): + return permission_request.get("toolCallId") + + prompt_request = data.get("promptRequest", {}) + if isinstance(prompt_request, dict): + return prompt_request.get("toolCallId") + + return None + + @staticmethod + def _extract_permission_kind(data: Dict[str, Any]) -> Optional[str]: + """Return the requested permission kind if present.""" + permission_request = data.get("permissionRequest", {}) + if isinstance(permission_request, dict) and permission_request.get("kind"): + return permission_request.get("kind") + + prompt_request = data.get("promptRequest", {}) + if isinstance(prompt_request, dict): + return prompt_request.get("kind") + + return None + + def _extract_permission_command(self, data: Dict[str, Any]) -> Optional[str]: + """Return the shell command associated with a permission request.""" + permission_request = data.get("permissionRequest", {}) + if isinstance(permission_request, dict): + command = permission_request.get("fullCommandText") + if isinstance(command, str) and command.strip(): + return truncate_middle(command.strip(), max_length=MAX_STRING_LENGTH, edge_chars=EDGE_CHARS) + + prompt_request = data.get("promptRequest", {}) + if isinstance(prompt_request, dict): + command = prompt_request.get("fullCommandText") + if isinstance(command, str) and command.strip(): + return truncate_middle(command.strip(), max_length=MAX_STRING_LENGTH, edge_chars=EDGE_CHARS) + + return None diff --git a/Sensor/tests/test_observer.py b/Sensor/tests/test_observer.py index 2854b0f..6802cef 100644 --- a/Sensor/tests/test_observer.py +++ b/Sensor/tests/test_observer.py @@ -16,6 +16,8 @@ def test_init_default(self, tmp_path): """Test default initialization.""" observer = AgentObserver(output_dir=tmp_path) assert observer.output_dir == tmp_path + assert hasattr(observer, "copilot_parser") + assert ("copilot", "GitHub Copilot") in observer.SOURCES def test_display_summary_empty(self, tmp_path, capsys): """Test display summary with no data.""" diff --git a/Sensor/tests/test_parsers.py b/Sensor/tests/test_parsers.py index afea155..816336f 100644 --- a/Sensor/tests/test_parsers.py +++ b/Sensor/tests/test_parsers.py @@ -12,6 +12,7 @@ from adr_sensor.parsers.claude_desktop_parser import ClaudeDesktopParser from adr_sensor.parsers.claude_parser import ClaudeParser from adr_sensor.parsers.cline_parser import ClineParser +from adr_sensor.parsers.copilot_parser import CopilotParser from adr_sensor.parsers.codex_parser import CodexParser from adr_sensor.parsers.cursor_parser import CursorParser from adr_sensor.parsers.opencode_parser import OpencodeParser @@ -101,6 +102,37 @@ def test_truncate_large_arguments(self): assert len(result["long"]) < 2000 assert "[truncated" in result["long"] + def test_uses_latest_timestamp_for_resumed_session(self, tmp_path): + """Incremental export should see a resumed session as updated.""" + jsonl_file = tmp_path / "resumed.jsonl" + messages = [ + { + "type": "user", + "sessionId": "session1", + "timestamp": "2025-06-15T10:00:00Z", + "message": {"content": "first prompt"}, + }, + { + "type": "assistant", + "sessionId": "session1", + "timestamp": "2025-06-15T10:00:01Z", + "message": {"content": [{"type": "text", "text": "first reply"}]}, + }, + { + "type": "user", + "sessionId": "session1", + "timestamp": "2025-06-16T12:30:00Z", + "message": {"content": "continued next day"}, + }, + ] + with open(jsonl_file, "w", encoding="utf-8") as f: + for msg in messages: + f.write(json.dumps(msg) + "\n") + + entries = ClaudeParser().parse_jsonl_file(jsonl_file) + assert len(entries) == 1 + assert entries[0].timestamp == datetime(2025, 6, 16, 12, 30, 0, tzinfo=timezone.utc) + class TestClineParser: def test_parse_cline_log(self, tmp_path): @@ -434,6 +466,139 @@ def test_parse_no_directory(self): entries = parser.parse_all() assert entries == [] + def test_uses_latest_event_timestamp_for_resumed_session(self, tmp_path): + """Incremental export should use the last event, not only session_meta.""" + jsonl_file = tmp_path / "resumed-codex.jsonl" + events = [ + { + "type": "session_meta", + "timestamp": "2025-06-15T10:00:00Z", + "payload": {"id": "sess-resume", "timestamp": "2025-06-15T10:00:00Z", "cwd": "/tmp"}, + }, + { + "type": "response_item", + "timestamp": "2025-06-15T10:00:01Z", + "payload": { + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": "first question"}], + }, + }, + { + "type": "response_item", + "timestamp": "2025-06-16T12:30:00Z", + "payload": { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": "later follow-up"}], + }, + }, + ] + with open(jsonl_file, "w", encoding="utf-8") as f: + for event in events: + f.write(json.dumps(event) + "\n") + + entry = CodexParser().parse_jsonl_file(jsonl_file) + assert entry is not None + assert entry.timestamp == datetime(2025, 6, 16, 12, 30, 0, tzinfo=timezone.utc) + + +class TestCopilotParser: + def test_parse_session_dir(self, tmp_path): + """Parse Copilot session directories into chat history and tool usage.""" + session_dir = tmp_path / "abc-session" + session_dir.mkdir() + + (session_dir / "workspace.yaml").write_text( + "\n".join( + [ + "id: abc-session", + "cwd: C:\\repo", + "client_name: vscode", + "name: Sample session", + "created_at: 2026-08-10T10:00:00.000Z", + "updated_at: 2026-08-10T11:00:00.000Z", + ] + ), + encoding="utf-8", + ) + (session_dir / "vscode.metadata.json").write_text( + json.dumps( + { + "origin": "vscode", + "created": 1785470000000, + "modified": 1785476400000, + "firstUserMessage": "inspect the repo", + } + ), + encoding="utf-8", + ) + + events = [ + { + "type": "session.start", + "timestamp": "2026-08-10T10:00:00.000Z", + "data": { + "sessionId": "abc-session", + "startTime": "2026-08-10T10:00:00.000Z", + "selectedModel": "gpt-5.6-sol", + "context": {"cwd": "C:\\repo"}, + }, + }, + { + "type": "user.message", + "id": "user-1", + "timestamp": "2026-08-10T10:00:05.000Z", + "data": {"content": "inspect the repo", "transformedContent": "...\ninspect the repo"}, + }, + { + "type": "assistant.message", + "id": "assistant-1", + "timestamp": "2026-08-10T10:00:06.000Z", + "data": { + "messageId": "assistant-message-1", + "model": "gpt-5.6-sol", + "content": "I will inspect it.", + "toolRequests": [ + { + "toolCallId": "call-1", + "name": "powershell", + "type": "function", + "arguments": {"command": "Get-ChildItem", "description": "List files"}, + } + ], + }, + }, + { + "type": "tool.execution_complete", + "timestamp": "2026-08-10T11:00:00.000Z", + "data": { + "toolCallId": "call-1", + "model": "gpt-5.6-sol", + "success": True, + "result": {"content": "file1\nfile2"}, + }, + }, + ] + with open(session_dir / "events.jsonl", "w", encoding="utf-8") as f: + for event in events: + f.write(json.dumps(event) + "\n") + + entry = CopilotParser(base_path=tmp_path).parse_session_dir(session_dir) + assert entry is not None + assert entry.source == "copilot" + assert entry.session_id == "copilot_abc-session" + assert entry.project_path == "C:\\repo" + assert entry.model == "gpt-5.6-sol" + assert entry.timestamp == datetime(2026, 8, 10, 11, 0, 0, tzinfo=timezone.utc) + assert [msg.role for msg in entry.chat_history] == ["user", "assistant"] + tool = entry.chat_history[1].tools[0] + assert tool.tool_name == "powershell" + assert tool.result == "file1\nfile2" + assert tool.status == "success" + assert entry.session_context["workspace_metadata"]["updated_at"] == "2026-08-10T11:00:00.000Z" + assert entry.session_context["transformed_user_messages"][0]["sequence_id"] == "user-1" + def _build_warp_db(db_path: Path, conversations: list) -> None: """Create a synthetic warp.sqlite matching the schema WarpParser queries. From 5fd47583a97e9f57bccecfbba388108f43e67e1f Mon Sep 17 00:00:00 2001 From: "Li Zexian (External - Orange CyberDefense Sweden AB)" Date: Wed, 12 Aug 2026 12:14:18 +0800 Subject: [PATCH 2/3] Add GitHub Copilot parser and fix incremental timestamps for resumed sessions --- Sensor/adr_sensor/parsers/codex_parser.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/Sensor/adr_sensor/parsers/codex_parser.py b/Sensor/adr_sensor/parsers/codex_parser.py index 42c430f..5d0958a 100644 --- a/Sensor/adr_sensor/parsers/codex_parser.py +++ b/Sensor/adr_sensor/parsers/codex_parser.py @@ -177,10 +177,13 @@ def _process_event(self, event: Dict[str, Any], session_data: Dict[str, Any]): event_timestamp = event.get("timestamp") if event_timestamp: - normalized = normalize_timestamp(event_timestamp) - current = session_data.get("last_event_timestamp") - if current is None or normalized > current: - session_data["last_event_timestamp"] = normalized + try: + normalized = normalize_timestamp(event_timestamp) + current = session_data.get("last_event_timestamp") + if current is None or normalized > current: + session_data["last_event_timestamp"] = normalized + except Exception: + pass if evt_type == "session_meta": session_data["id"] = payload.get("id") From 30d2982d851f2d07123c011e44e753dd283024b1 Mon Sep 17 00:00:00 2001 From: "Li Zexian (External - Orange CyberDefense Sweden AB)" Date: Tue, 1 Sep 2026 18:15:35 +0800 Subject: [PATCH 3/3] Fix incremental exports for Codex and Copilot --- Sensor/adr_sensor/observer.py | 534 +++++++++++++++++++- Sensor/adr_sensor/parsers/claude_parser.py | 2 +- Sensor/adr_sensor/parsers/codex_parser.py | 21 +- Sensor/adr_sensor/parsers/copilot_parser.py | 23 +- Sensor/tests/test_observer.py | 417 ++++++++++++++- Sensor/tests/test_parsers.py | 87 ++-- 6 files changed, 1027 insertions(+), 57 deletions(-) diff --git a/Sensor/adr_sensor/observer.py b/Sensor/adr_sensor/observer.py index 514f22b..962ca10 100644 --- a/Sensor/adr_sensor/observer.py +++ b/Sensor/adr_sensor/observer.py @@ -5,10 +5,16 @@ the ingestion, display, and export of agent telemetry data. """ +import errno +import hashlib import json import os import platform +import re +import secrets +import stat import sys +import time import traceback from datetime import datetime from pathlib import Path @@ -19,8 +25,8 @@ from .parsers.claude_desktop_parser import ClaudeDesktopParser from .parsers.claude_parser import ClaudeParser from .parsers.cline_parser import ClineParser -from .parsers.copilot_parser import CopilotParser from .parsers.codex_parser import CodexParser +from .parsers.copilot_parser import CopilotParser from .parsers.cursor_parser import CursorParser from .parsers.opencode_parser import OpencodeParser from .parsers.warp_parser import WarpParser @@ -28,6 +34,8 @@ from .schemas.system_config_schema import SystemConfiguration from .utils.timestamp_utils import format_timestamp_for_filename, normalize_timestamp, parse_timestamp_from_filename +_COLLISION_SUFFIX_PATTERN = re.compile(r"_([0-9a-f]{64})(?:_(\d+))?$") + class AgentObserver: """Main class for observing and analyzing AI agent interactions. @@ -61,6 +69,8 @@ class AgentObserver: "claude_desktop": ("Darwin", "Windows"), } + CONTENT_AWARE_INCREMENTAL_SOURCES = frozenset({"codex", "copilot"}) + def __init__(self, output_dir: Optional[Path] = None, max_age_days: Optional[int] = None): """Initialize the AgentObserver. @@ -282,20 +292,85 @@ def save_sessions_to_individual_files( output_dir.mkdir(parents=True, exist_ok=True) saved_files = [] + session_file_index = ( + self._build_session_file_index(output_dir) + if any(entry.source in self.CONTENT_AWARE_INCREMENTAL_SOURCES for entry in entries) + else {} + ) for entry in entries: timestamp_str = format_timestamp_for_filename(entry.timestamp) - clean_session_id = self._clean_filename(entry.session_id) - filename = f"adr.{clean_session_id}.{timestamp_str}.json" + filename_session_id = ( + self._session_filename_id(entry.session_id) + if entry.source in self.CONTENT_AWARE_INCREMENTAL_SOURCES + else self._clean_filename(entry.session_id) + ) + filename = f"adr.{filename_session_id}.{timestamp_str}.json" file_path = output_dir / filename + temp_path: Optional[Path] = None + lock_path: Optional[Path] = None + lock_fd: Optional[int] = None try: - with open(file_path, "w", encoding="utf-8") as f: - json.dump(entry.get_non_null_fields(), f, indent=2, ensure_ascii=False) + if entry.source in self.CONTENT_AWARE_INCREMENTAL_SOURCES: + lock_path = output_dir / f".adr.{filename_session_id}.lock" + lock_fd = self._acquire_session_lock(lock_path) + existing_info = self._find_session_file(entry, session_file_index) + file_path = self._resolve_session_file_path( + output_dir, entry, filename_session_id, timestamp_str, existing_info + ) + filename = file_path.name + fresh_target = self._session_file_info(file_path) + if ( + fresh_target is not None + and fresh_target["data"].get("session_id") == entry.session_id + ): + existing_info = self._newer_session_file(existing_info, fresh_target) + if self._session_revision_regresses(entry, existing_info): + print(f"Skipped stale session: {filename}") + continue + + entry_data = entry.get_non_null_fields() + temp_path, temp_fd = self._create_session_temp(output_dir, filename, file_path) + with os.fdopen(temp_fd, mode="w", encoding="utf-8") as f: + json.dump(entry_data, f, indent=2, ensure_ascii=False) + f.flush() + os.fsync(f.fileno()) + os.replace(temp_path, file_path) + temp_path = None + self._sync_session_directory(output_dir) + + if entry.source in self.CONTENT_AWARE_INCREMENTAL_SOURCES: + removed_stale_files = self._remove_stale_session_files( + entry.session_id, + file_path, + self._session_file_candidates(entry, session_file_index), + ) + if removed_stale_files: + self._sync_session_directory(output_dir) + self._index_session_file(session_file_index, file_path, entry_data) + saved_files.append(file_path) print(f"Saved session: {filename}") except Exception as e: print(f"Error saving session {filename}: {e}") + self._emit_error( + { + "source": entry.source, + "stage": "save_session", + "error_type": e.__class__.__name__, + "message": str(e), + "session_id": entry.session_id, + } + ) + finally: + if temp_path is not None and temp_path.exists(): + try: + temp_path.unlink() + except OSError as cleanup_error: + print(f"Error removing temporary session file {temp_path.name}: {cleanup_error}") + if lock_fd is not None and lock_path is not None: + self._release_session_lock(lock_fd) print(f"\nSaved {len(saved_files)} sessions to: {output_dir}") return saved_files @@ -305,15 +380,36 @@ def filter_entries_by_existing_files( ) -> List[AgentEvent]: """Filter out entries that haven't changed since last processing.""" existing_files = self._get_existing_session_files(output_dir) + target_dir = Path(output_dir) if output_dir is not None else self._get_default_session_dir() + session_file_index = ( + self._build_session_file_index(target_dir) + if any(entry.source in self.CONTENT_AWARE_INCREMENTAL_SOURCES for entry in entries) + else {} + ) filtered_entries = [] for entry in entries: - session_id = entry.session_id - if session_id not in existing_files: + filename_session_id = ( + self._session_filename_id(entry.session_id) + if entry.source in self.CONTENT_AWARE_INCREMENTAL_SOURCES + else self._clean_filename(entry.session_id) + ) + existing_info = existing_files.get(filename_session_id) + if entry.source in self.CONTENT_AWARE_INCREMENTAL_SOURCES: + existing_info = self._find_session_file(entry, session_file_index) + + if existing_info is None: filtered_entries.append(entry) continue - existing_info = existing_files[session_id] + if entry.source in self.CONTENT_AWARE_INCREMENTAL_SOURCES: + if self._session_content_changed(entry, existing_info): + existing_revision = self._session_file_revision(existing_info) + current_revision = self._entry_session_revision(entry) + if existing_revision is None or current_revision is None or current_revision >= existing_revision: + filtered_entries.append(entry) + continue + existing_ts = normalize_timestamp(existing_info["timestamp"]).replace(microsecond=0) entry_ts = normalize_timestamp(entry.timestamp).replace(microsecond=0) @@ -322,6 +418,401 @@ def filter_entries_by_existing_files( return filtered_entries + def _session_content_changed(self, entry: AgentEvent, existing_info: Dict[str, Any]) -> bool: + """Compare complete exported content for sources whose sessions can resume.""" + existing_path = existing_info["file_path"] + try: + existing_data = existing_info.get("data") + if existing_data is None: + with open(existing_path, encoding="utf-8") as handle: + existing_data = json.load(handle) + if not isinstance(existing_data, dict): + raise ValueError("existing session file must contain a JSON object") + current_content = self._session_export_content(entry.get_non_null_fields()) + existing_content = self._session_export_content(existing_data) + if entry.source == "codex" and "session_context" not in existing_content: + current_context = current_content.get("session_context") + if isinstance(current_context, dict) and set(current_context) == {"last_event_at"}: + current_content.pop("session_context") + return current_content != existing_content + except (OSError, UnicodeError, json.JSONDecodeError, TypeError, ValueError) as exc: + print(f"Error comparing existing session {existing_path.name}: {exc}") + self._emit_error( + { + "source": entry.source, + "stage": "compare_session", + "error_type": exc.__class__.__name__, + "message": str(exc), + "session_id": entry.session_id, + } + ) + return True + + @staticmethod + def _session_export_content(event_data: Dict[str, Any]) -> Dict[str, Any]: + """Ignore fields that change only because a snapshot is rewritten.""" + return {key: value for key, value in event_data.items() if key not in {"timestamp", "uuid"}} + + def _find_session_file( + self, + entry: AgentEvent, + session_file_index: Dict[str, List[Dict[str, Any]]], + ) -> Optional[Dict[str, Any]]: + """Find the latest snapshot whose stored session ID matches exactly.""" + latest: Optional[Dict[str, Any]] = None + for candidate_info in self._session_file_candidates(entry, session_file_index): + candidate = candidate_info["file_path"] + if not candidate.exists(): + continue + data = candidate_info.get("data") or self._load_session_file(candidate) + if data is None or data.get("session_id") != entry.session_id: + continue + candidate_info = dict(candidate_info) + candidate_info["data"] = data + candidate_info["revision"] = self._session_file_revision(candidate_info) + latest = self._newer_session_file(latest, candidate_info) + return latest + + def _session_file_info(self, file_path: Path) -> Optional[Dict[str, Any]]: + if not file_path.exists(): + return None + file_timestamp = parse_timestamp_from_filename(file_path.name) + data = self._load_session_file(file_path) + if file_timestamp is None or data is None: + return None + info = { + "file_path": file_path, + "timestamp": file_timestamp, + "filename": file_path.name, + "data": data, + } + info["revision"] = self._session_file_revision(info) + return info + + @staticmethod + def _newer_session_file( + current: Optional[Dict[str, Any]], + candidate: Dict[str, Any], + ) -> Dict[str, Any]: + if current is None: + return candidate + candidate_revision = candidate.get("revision") + current_revision = current.get("revision") + candidate_event_count = AgentObserver._session_file_event_count(candidate) + current_event_count = AgentObserver._session_file_event_count(current) + if ( + candidate_revision is not None + and (current_revision is None or candidate_revision > current_revision) + ) or ( + candidate_revision == current_revision + and ( + candidate_event_count is not None + and (current_event_count is None or candidate_event_count > current_event_count) + ) + ) or ( + candidate_revision == current_revision + and candidate_event_count == current_event_count + and candidate["timestamp"] > current["timestamp"] + ): + return candidate + return current + + @staticmethod + def _create_session_temp(output_dir: Path, filename: str, target_path: Path) -> Tuple[Path, int]: + """Create a same-directory temporary file with compatible permissions.""" + for _ in range(10): + temp_path = output_dir / f".{filename}.{secrets.token_hex(8)}.tmp" + try: + fd = os.open(temp_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o666) + except FileExistsError: + continue + + try: + if target_path.exists(): + os.chmod(temp_path, stat.S_IMODE(target_path.stat().st_mode)) + except OSError: + os.close(fd) + temp_path.unlink(missing_ok=True) + raise + return temp_path, fd + raise FileExistsError(f"unable to allocate temporary file for {filename}") + + def _build_session_file_index(self, output_dir: Path) -> Dict[str, List[Dict[str, Any]]]: + """Index filenames once so batch incremental processing stays linear.""" + index: Dict[str, List[Dict[str, Any]]] = {} + if not output_dir.exists(): + return index + + for file_path in output_dir.glob("adr.*.json"): + file_timestamp = parse_timestamp_from_filename(file_path.name) + if file_timestamp is None: + continue + filename_session_id = file_path.name[4:-5].rsplit(".", 1)[0] + info = { + "file_path": file_path, + "timestamp": file_timestamp, + "filename": file_path.name, + } + index.setdefault(filename_session_id, []).append(info) + match = _COLLISION_SUFFIX_PATTERN.search(filename_session_id) + if match: + index.setdefault(f"#sha256:{match.group(1)}", []).append(info) + return index + + def _session_file_candidates( + self, + entry: AgentEvent, + session_file_index: Dict[str, List[Dict[str, Any]]], + ) -> List[Dict[str, Any]]: + filename_session_id = self._session_filename_id(entry.session_id) + legacy_session_id = self._clean_filename(entry.session_id) + digest = hashlib.sha256(entry.session_id.encode("utf-8")).hexdigest() + candidates: List[Dict[str, Any]] = [] + seen_paths = set() + for key in (filename_session_id, legacy_session_id, f"#sha256:{digest}"): + for info in session_file_index.get(key, []): + path = info["file_path"] + if path in seen_paths: + continue + candidate_session_id = path.name[4:-5].rsplit(".", 1)[0] + if self._filename_id_matches_session(candidate_session_id, entry.session_id): + candidates.append(info) + seen_paths.add(path) + return candidates + + def _index_session_file( + self, + session_file_index: Dict[str, List[Dict[str, Any]]], + file_path: Path, + data: Dict[str, Any], + ) -> None: + filename_session_id = file_path.name[4:-5].rsplit(".", 1)[0] + keys = [filename_session_id] + match = _COLLISION_SUFFIX_PATTERN.search(filename_session_id) + if match: + keys.append(f"#sha256:{match.group(1)}") + for key in keys: + session_file_index[key] = [ + info for info in session_file_index.get(key, []) if info["file_path"] != file_path + ] + + info = { + "file_path": file_path, + "timestamp": parse_timestamp_from_filename(file_path.name), + "filename": file_path.name, + "data": data, + } + session_file_index.setdefault(filename_session_id, []).append(info) + if match: + session_file_index.setdefault(f"#sha256:{match.group(1)}", []).append(info) + + def _resolve_session_file_path( + self, + output_dir: Path, + entry: AgentEvent, + filename_session_id: str, + timestamp_str: str, + existing_info: Optional[Dict[str, Any]], + ) -> Path: + """Resolve a stable path without overwriting a colliding session ID.""" + preferred = output_dir / f"adr.{filename_session_id}.{timestamp_str}.json" + if existing_info is not None and format_timestamp_for_filename(existing_info["timestamp"]) == timestamp_str: + existing_session_part = existing_info["file_path"].name[4:-5].rsplit(".", 1)[0] + if ( + existing_session_part == filename_session_id + or existing_session_part.startswith(f"{filename_session_id}_") + ): + return existing_info["file_path"] + + if not preferred.exists(): + return preferred + if self._session_file_has_id(preferred, entry.session_id): + return preferred + + digest = hashlib.sha256(entry.session_id.encode("utf-8")).hexdigest() + counter = 0 + while True: + suffix = digest if counter == 0 else f"{digest}_{counter}" + prefix = filename_session_id[: 200 - len(suffix) - 1] + alternate_id = f"{prefix}_{suffix}".strip("_") + alternate = output_dir / f"adr.{alternate_id}.{timestamp_str}.json" + if not alternate.exists() or self._session_file_has_id(alternate, entry.session_id): + return alternate + counter += 1 + + def _session_revision_regresses( + self, entry: AgentEvent, existing_info: Optional[Dict[str, Any]] + ) -> bool: + """Prevent an older concurrent parse from replacing a newer snapshot.""" + if existing_info is None: + return False + existing_revision = self._session_file_revision(existing_info) + current_revision = self._entry_session_revision(entry) + existing_event_count = self._session_file_event_count(existing_info) + current_event_count = self._entry_session_event_count(entry) + return ( + existing_revision is not None + and current_revision is not None + and ( + current_revision < existing_revision + or ( + current_revision == existing_revision + and existing_event_count is not None + and current_event_count is not None + and current_event_count < existing_event_count + ) + ) + ) + + @staticmethod + def _entry_session_revision(entry: AgentEvent) -> Optional[datetime]: + context = entry.session_context or {} + last_event_at = context.get("last_event_at") if isinstance(context, dict) else None + if last_event_at is None: + return None + try: + return normalize_timestamp(last_event_at) + except (TypeError, ValueError): + return None + + @staticmethod + def _entry_session_event_count(entry: AgentEvent) -> Optional[int]: + context = entry.session_context or {} + event_count = context.get("event_count") if isinstance(context, dict) else None + return event_count if isinstance(event_count, int) and event_count >= 0 else None + + @staticmethod + def _session_file_event_count(existing_info: Dict[str, Any]) -> Optional[int]: + data = existing_info.get("data") + if data is None: + data = AgentObserver._load_session_file(existing_info["file_path"]) + context = data.get("session_context") if isinstance(data, dict) else None + event_count = context.get("event_count") if isinstance(context, dict) else None + return event_count if isinstance(event_count, int) and event_count >= 0 else None + + @staticmethod + def _session_file_revision(existing_info: Dict[str, Any]) -> Optional[datetime]: + if "revision" in existing_info: + revision = existing_info["revision"] + return revision if isinstance(revision, datetime) else None + try: + data = existing_info.get("data") + if data is None: + with open(existing_info["file_path"], encoding="utf-8") as handle: + data = json.load(handle) + if not isinstance(data, dict): + return None + context = data.get("session_context") + last_event_at = context.get("last_event_at") if isinstance(context, dict) else None + return normalize_timestamp(last_event_at or data.get("timestamp") or existing_info["timestamp"]) + except (OSError, UnicodeError, json.JSONDecodeError, TypeError, ValueError): + return None + + @staticmethod + def _acquire_session_lock(lock_path: Path, timeout_seconds: float = 30.0) -> int: + """Acquire an OS-managed cross-process lock for one session.""" + fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600) + try: + if os.fstat(fd).st_size == 0: + os.write(fd, b"0") + except Exception: + os.close(fd) + raise + + deadline = time.monotonic() + timeout_seconds + while True: + try: + os.lseek(fd, 0, os.SEEK_SET) + if os.name == "nt": + import msvcrt + + msvcrt.locking(fd, msvcrt.LK_NBLCK, 1) + else: + import fcntl + + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + return fd + except OSError as exc: + if exc.errno not in {errno.EACCES, errno.EAGAIN}: + os.close(fd) + raise + if time.monotonic() >= deadline: + os.close(fd) + raise TimeoutError(f"timed out waiting for session lock {lock_path.name}") + time.sleep(0.05) + + @staticmethod + def _release_session_lock(lock_fd: int) -> None: + try: + os.lseek(lock_fd, 0, os.SEEK_SET) + if os.name == "nt": + import msvcrt + + msvcrt.locking(lock_fd, msvcrt.LK_UNLCK, 1) + else: + import fcntl + + fcntl.flock(lock_fd, fcntl.LOCK_UN) + finally: + os.close(lock_fd) + + @staticmethod + def _sync_session_directory(output_dir: Path) -> None: + """Persist directory entries after replacing or removing snapshots.""" + if os.name == "nt": + return + directory_fd = os.open(output_dir, os.O_RDONLY) + try: + os.fsync(directory_fd) + finally: + os.close(directory_fd) + + def _remove_stale_session_files( + self, + session_id: str, + keep_path: Path, + candidates: List[Dict[str, Any]], + ) -> bool: + """Remove superseded moving-timestamp snapshots after the replacement succeeds.""" + removed_files = False + for info in candidates: + candidate = info["file_path"] + if candidate == keep_path or not candidate.exists(): + continue + # Reload immediately before removal: indexed data can be stale after a writer updates a path. + data = self._load_session_file(candidate) + if data is None or data.get("session_id") != session_id: + continue + try: + candidate.unlink() + removed_files = True + except OSError as exc: + print(f"Error removing stale session {candidate.name}: {exc}") + self._emit_error( + { + "stage": "remove_stale_session", + "error_type": exc.__class__.__name__, + "message": str(exc), + "session_id": session_id, + } + ) + return removed_files + + @staticmethod + def _session_file_has_id(file_path: Path, session_id: str) -> bool: + """Verify ownership before migrating or deleting a legacy filename.""" + data = AgentObserver._load_session_file(file_path) + return data is not None and data.get("session_id") == session_id + + @staticmethod + def _load_session_file(file_path: Path) -> Optional[Dict[str, Any]]: + try: + with open(file_path, encoding="utf-8") as handle: + data = json.load(handle) + return data if isinstance(data, dict) else None + except (OSError, UnicodeError, json.JSONDecodeError, TypeError): + return None + def _get_existing_session_files(self, output_dir: Optional[Path] = None) -> Dict[str, Dict[str, Any]]: """Get a mapping of session_id to file info for existing session files.""" if output_dir is None: @@ -373,3 +864,30 @@ def _clean_filename(self, session_id: str) -> str: clean_id = clean_id[:max_session_length] return clean_id + + def _session_filename_id(self, session_id: str) -> str: + """Preserve existing safe names while disambiguating lossy sanitization.""" + clean_id = self._clean_filename(session_id) + if clean_id == session_id: + return clean_id + + digest = hashlib.sha256(session_id.encode("utf-8")).hexdigest()[:12] + clean_id = clean_id[: 200 - len(digest) - 1] + return f"{clean_id}_{digest}".strip("_") + + def _filename_id_matches_session(self, filename_session_id: str, session_id: str) -> bool: + current_id = self._session_filename_id(session_id) + if filename_session_id in {current_id, self._clean_filename(session_id)}: + return True + + match = _COLLISION_SUFFIX_PATTERN.search(filename_session_id) + if not match: + return False + digest = hashlib.sha256(session_id.encode("utf-8")).hexdigest() + if match.group(1) != digest: + return False + + counter = int(match.group(2)) if match.group(2) is not None else 0 + suffix = digest if counter == 0 else f"{digest}_{counter}" + prefix = current_id[: 200 - len(suffix) - 1] + return filename_session_id == f"{prefix}_{suffix}".strip("_") diff --git a/Sensor/adr_sensor/parsers/claude_parser.py b/Sensor/adr_sensor/parsers/claude_parser.py index 05000b7..856de38 100644 --- a/Sensor/adr_sensor/parsers/claude_parser.py +++ b/Sensor/adr_sensor/parsers/claude_parser.py @@ -127,7 +127,7 @@ def parse_jsonl_file(self, file_path: Path) -> List[AgentEvent]: if "timestamp" in obj: try: ts = normalize_timestamp(obj["timestamp"]) - if sessions[session_id]["timestamp"] is None or ts > sessions[session_id]["timestamp"]: + if sessions[session_id]["timestamp"] is None or ts < sessions[session_id]["timestamp"]: sessions[session_id]["timestamp"] = ts except Exception: pass diff --git a/Sensor/adr_sensor/parsers/codex_parser.py b/Sensor/adr_sensor/parsers/codex_parser.py index 5d0958a..7c3e490 100644 --- a/Sensor/adr_sensor/parsers/codex_parser.py +++ b/Sensor/adr_sensor/parsers/codex_parser.py @@ -48,7 +48,9 @@ def parse_jsonl_file(self, file_path: Path) -> Optional[AgentEvent]: session_data: Dict[str, Any] = { "id": None, "timestamp": None, + "first_event_timestamp": None, "last_event_timestamp": None, + "event_count": 0, "cwd": None, "model": None, "messages": [], @@ -62,6 +64,7 @@ def parse_jsonl_file(self, file_path: Path) -> Optional[AgentEvent]: continue try: event = json.loads(line) + session_data["event_count"] += 1 self._process_event(event, session_data) except json.JSONDecodeError: continue @@ -95,13 +98,26 @@ def parse_jsonl_file(self, file_path: Path) -> Optional[AgentEvent]: ) return AgentEvent( - timestamp=session_data["last_event_timestamp"] or session_data["timestamp"] or datetime.now(timezone.utc), + # Keep the filename identity stable; resumed content is detected by the exporter. + timestamp=( + session_data["timestamp"] + or session_data["first_event_timestamp"] + or datetime.now(timezone.utc) + ), source="codex", session_id=f"codex_{session_data['id']}", project_path=session_data["cwd"], model=session_data["model"], chat_history=chat_history, raw_log_path=str(file_path), + session_context=( + { + "last_event_at": session_data["last_event_timestamp"].isoformat(), + "event_count": session_data["event_count"], + } + if session_data["last_event_timestamp"] + else {"event_count": session_data["event_count"]} + ), ) except Exception as e: @@ -179,6 +195,9 @@ def _process_event(self, event: Dict[str, Any], session_data: Dict[str, Any]): if event_timestamp: try: normalized = normalize_timestamp(event_timestamp) + current = session_data.get("first_event_timestamp") + if current is None or normalized < current: + session_data["first_event_timestamp"] = normalized current = session_data.get("last_event_timestamp") if current is None or normalized > current: session_data["last_event_timestamp"] = normalized diff --git a/Sensor/adr_sensor/parsers/copilot_parser.py b/Sensor/adr_sensor/parsers/copilot_parser.py index a39845d..933ac4a 100644 --- a/Sensor/adr_sensor/parsers/copilot_parser.py +++ b/Sensor/adr_sensor/parsers/copilot_parser.py @@ -132,10 +132,8 @@ def parse_session_dir(self, session_dir: Path) -> Optional[AgentEvent]: return None timestamp = ( - self._normalize_optional_timestamp(workspace_meta.get("updated_at")) - or self._normalize_optional_timestamp(vscode_meta.get("modified")) - or self._normalize_optional_timestamp(session_data["last_event_at"]) - or session_data["timestamp"] + # Updated/modified sidecars can lag resumed events and must not define file identity. + session_data["timestamp"] or self._normalize_optional_timestamp(workspace_meta.get("created_at")) or self._normalize_optional_timestamp(vscode_meta.get("created")) or self._normalize_optional_timestamp(session_data["first_event_at"]) @@ -165,10 +163,13 @@ def _process_event(self, event: Dict[str, Any], session_data: Dict[str, Any]): event_time = self._normalize_optional_timestamp(event.get("timestamp")) session_data["event_counts"][event_type] += 1 - if event_time and session_data["first_event_at"] is None: - session_data["first_event_at"] = event_time if event_time: - session_data["last_event_at"] = event_time + first_event_at = session_data["first_event_at"] + last_event_at = session_data["last_event_at"] + if first_event_at is None or event_time < first_event_at: + session_data["first_event_at"] = event_time + if last_event_at is None or event_time > last_event_at: + session_data["last_event_at"] = event_time if event_type == "session.start": session_id = data.get("sessionId") @@ -259,10 +260,11 @@ def _process_event(self, event: Dict[str, Any], session_data: Dict[str, Any]): if tool is not None: success = bool(data.get("success")) result_text = self._normalize_tool_result(data.get("result")) + error_text = self._normalize_tool_result(data.get("error")) tool["result"] = result_text tool["status"] = "success" if success else "error" - if not success and result_text: - tool["error"] = result_text + if not success: + tool["error"] = error_text or result_text session_data["model"] = data.get("model") or session_data["model"] return @@ -427,6 +429,7 @@ def _build_session_context(self, session_data: Dict[str, Any]) -> Dict[str, Any] "first_event_at": self._format_datetime(session_data["first_event_at"]), "last_event_at": self._format_datetime(session_data["last_event_at"]), "event_counts": dict(session_data["event_counts"]), + "event_count": sum(session_data["event_counts"].values()), "model_changes": session_data["model_changes"], "usage_checkpoints": session_data["usage_checkpoints"], "permissions": session_data["permissions"], @@ -450,6 +453,8 @@ def _build_session_context(self, session_data: Dict[str, Any]) -> Dict[str, Any] filtered[key] = value elif isinstance(value, str) and value: filtered[key] = value + elif isinstance(value, int) and value >= 0: + filtered[key] = value return filtered def _normalize_tool_result(self, result: Any) -> Optional[str]: diff --git a/Sensor/tests/test_observer.py b/Sensor/tests/test_observer.py index 6802cef..b8f9e87 100644 --- a/Sensor/tests/test_observer.py +++ b/Sensor/tests/test_observer.py @@ -1,8 +1,9 @@ """Tests for the AgentObserver.""" import json +import os +import stat from datetime import datetime, timezone -from pathlib import Path from unittest.mock import MagicMock, patch import pytest @@ -126,6 +127,8 @@ def test_clean_filename(self, tmp_path): assert observer._clean_filename("simple_id") == "simple_id" assert observer._clean_filename("path/with:special*chars") == "path_with_special_chars" assert observer._clean_filename(" spaces ") == "spaces" + assert observer._session_filename_id("simple_id") == "simple_id" + assert observer._session_filename_id("path/with") != observer._session_filename_id("path:with") def test_filter_entries_by_existing_files(self, tmp_path): """Test incremental filtering.""" @@ -159,6 +162,418 @@ def test_filter_entries_by_existing_files(self, tmp_path): assert len(filtered) == 1 assert filtered[0].session_id == "claude_session2" + def test_content_filter_ignores_timestamp_identity_migration(self, tmp_path): + """Changing from activity time to start time must not re-export unchanged history.""" + observer = AgentObserver(output_dir=tmp_path) + entry = AgentEvent( + timestamp=datetime(2025, 6, 15, 10, 0, 0, tzinfo=timezone.utc), + source="codex", + session_id="codex_session1", + hostname="host", + username="user", + chat_history=[ChatMessage(role="user", content="unchanged")], + session_context={"last_event_at": "2025-06-16T12:30:00+00:00"}, + ) + legacy_data = entry.get_non_null_fields() + legacy_data["timestamp"] = "2025-06-16T12:30:00+00:00" + legacy_data["uuid"] = "legacy-moving-timestamp-uuid" + legacy_data.pop("session_context") + legacy_file = tmp_path / "adr.codex_session1.20250616_123000.json" + legacy_file.write_text(json.dumps(legacy_data), encoding="utf-8") + + assert observer.filter_entries_by_existing_files([entry], output_dir=tmp_path) == [] + assert legacy_file.exists() + + def test_content_filter_detects_same_second_tool_result(self, tmp_path): + """A result appended in the filename's second must still replace the snapshot.""" + observer = AgentObserver(output_dir=tmp_path) + timestamp = datetime(2025, 6, 15, 10, 0, 0, 100000, tzinfo=timezone.utc) + pending = AgentEvent( + timestamp=timestamp, + source="codex", + session_id="codex_session1", + hostname="host", + username="user", + chat_history=[ + ChatMessage( + role="assistant", + content="", + tools=[ToolUsage(tool_name="shell", tool_type="function_call", status="pending")], + ) + ], + ) + completed = AgentEvent( + timestamp=timestamp.replace(microsecond=900000), + source="codex", + session_id="codex_session1", + hostname="host", + username="user", + chat_history=[ + ChatMessage( + role="assistant", + content="", + tools=[ + ToolUsage( + tool_name="shell", + tool_type="function_call", + result="done", + status="success", + ) + ], + ) + ], + ) + + first_saved = observer.save_sessions_to_individual_files([pending], output_dir=tmp_path) + assert observer.filter_entries_by_existing_files([completed], output_dir=tmp_path) == [completed] + second_saved = observer.save_sessions_to_individual_files([completed], output_dir=tmp_path) + + assert first_saved == second_saved + session_files = list(tmp_path.glob("adr.codex_session1.*.json")) + assert session_files == second_saved + saved_data = json.loads(session_files[0].read_text(encoding="utf-8")) + assert saved_data["chat_history"][0]["tools"][0]["result"] == "done" + + def test_content_update_replaces_legacy_snapshot_with_stable_file(self, tmp_path): + """A changed legacy snapshot should converge to one start-time file.""" + observer = AgentObserver(output_dir=tmp_path) + legacy_file = tmp_path / "adr.copilot_session1.20250616_123000.json" + backup_file = tmp_path / "adr.copilot_session1.backup.json" + backup_file.write_text("keep", encoding="utf-8") + legacy_file.write_text( + json.dumps( + { + "timestamp": "2025-06-16T12:30:00+00:00", + "source": "copilot", + "session_id": "copilot_session1", + "chat_history": [{"role": "user", "content": "old", "tools": []}], + "uuid": "legacy", + } + ), + encoding="utf-8", + ) + entry = AgentEvent( + timestamp=datetime(2025, 6, 15, 10, 0, 0, tzinfo=timezone.utc), + source="copilot", + session_id="copilot_session1", + hostname="host", + username="user", + chat_history=[ChatMessage(role="user", content="updated")], + ) + + assert observer.filter_entries_by_existing_files([entry], output_dir=tmp_path) == [entry] + saved = observer.save_sessions_to_individual_files([entry], output_dir=tmp_path) + + assert [path.name for path in saved] == ["adr.copilot_session1.20250615_100000.json"] + assert saved[0].exists() + assert not legacy_file.exists() + assert backup_file.read_text(encoding="utf-8") == "keep" + + def test_cleanup_rechecks_session_ownership_before_deleting(self, tmp_path): + """A stale in-memory index must not authorize deleting a replaced path.""" + observer = AgentObserver(output_dir=tmp_path) + candidate = tmp_path / "adr.codex_session1.20250616_123000.json" + candidate.write_text(json.dumps({"session_id": "other_session"}), encoding="utf-8") + + removed = observer._remove_stale_session_files( + "codex_session1", + tmp_path / "adr.codex_session1.20250615_100000.json", + [{"file_path": candidate, "data": {"session_id": "codex_session1"}}], + ) + + assert not removed + assert candidate.exists() + + def test_directory_sync_failure_preserves_legacy_snapshot(self, tmp_path): + """Do not remove the only prior snapshot until the replacement entry is durable.""" + observer = AgentObserver(output_dir=tmp_path) + legacy_file = tmp_path / "adr.codex_session1.20250616_123000.json" + legacy_file.write_text(json.dumps({"session_id": "codex_session1"}), encoding="utf-8") + entry = AgentEvent( + timestamp=datetime(2025, 6, 15, 10, 0, 0, tzinfo=timezone.utc), + source="codex", + session_id="codex_session1", + hostname="host", + username="user", + chat_history=[ChatMessage(role="user", content="updated")], + ) + + with patch.object(observer, "_sync_session_directory", side_effect=OSError("sync failed")): + assert observer.save_sessions_to_individual_files([entry], output_dir=tmp_path) == [] + + assert legacy_file.exists() + + def test_atomic_save_failure_preserves_existing_snapshot(self, tmp_path): + """A failed replacement must leave the previous complete JSON intact.""" + observer = AgentObserver(output_dir=tmp_path) + old_entry = AgentEvent( + timestamp=datetime(2025, 6, 15, 10, 0, 0, tzinfo=timezone.utc), + source="codex", + session_id="codex_session1", + hostname="host", + username="user", + chat_history=[ChatMessage(role="user", content="old")], + ) + new_entry = AgentEvent( + timestamp=old_entry.timestamp, + source="codex", + session_id="codex_session1", + hostname="host", + username="user", + chat_history=[ChatMessage(role="user", content="new")], + ) + existing_path = observer.save_sessions_to_individual_files([old_entry], output_dir=tmp_path)[0] + original_data = existing_path.read_text(encoding="utf-8") + + with patch("adr_sensor.observer.os.replace", side_effect=OSError("replace failed")): + saved = observer.save_sessions_to_individual_files([new_entry], output_dir=tmp_path) + + assert saved == [] + assert existing_path.read_text(encoding="utf-8") == original_data + assert list(tmp_path.glob("*.tmp")) == [] + + @pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits are not preserved on Windows") + def test_atomic_replacement_preserves_existing_permissions(self, tmp_path): + observer = AgentObserver(output_dir=tmp_path) + entry = AgentEvent( + timestamp=datetime(2025, 6, 15, 10, 0, 0, tzinfo=timezone.utc), + source="codex", + session_id="codex_session1", + hostname="host", + username="user", + chat_history=[ChatMessage(role="user", content="old")], + ) + path = observer.save_sessions_to_individual_files([entry], output_dir=tmp_path)[0] + path.chmod(0o640) + updated = AgentEvent( + timestamp=entry.timestamp, + source=entry.source, + session_id=entry.session_id, + hostname=entry.hostname, + username=entry.username, + chat_history=[ChatMessage(role="user", content="updated")], + ) + + observer.save_sessions_to_individual_files([updated], output_dir=tmp_path) + + assert stat.S_IMODE(path.stat().st_mode) == 0o640 + + def test_lossy_session_ids_do_not_overwrite_or_delete_each_other(self, tmp_path): + """Sanitized filename collisions must retain both sessions.""" + observer = AgentObserver(output_dir=tmp_path) + timestamp = datetime(2025, 6, 15, 10, 0, 0, tzinfo=timezone.utc) + slash_entry = AgentEvent( + timestamp=timestamp, + source="codex", + session_id="codex_a/b", + hostname="host", + username="user", + chat_history=[ChatMessage(role="user", content="slash")], + ) + colon_entry = AgentEvent( + timestamp=timestamp, + source="codex", + session_id="codex_a:b", + hostname="host", + username="user", + chat_history=[ChatMessage(role="user", content="colon")], + ) + + saved = observer.save_sessions_to_individual_files([slash_entry, colon_entry], output_dir=tmp_path) + + assert len(saved) == 2 + assert saved[0] != saved[1] + assert all(path.exists() for path in saved) + assert observer.filter_entries_by_existing_files([slash_entry, colon_entry], output_dir=tmp_path) == [] + + colliding_legacy = tmp_path / "adr.codex_a_b.20250616_100000.json" + colliding_legacy.write_text(json.dumps(colon_entry.get_non_null_fields()), encoding="utf-8") + updated_slash = AgentEvent( + timestamp=timestamp, + source="codex", + session_id=slash_entry.session_id, + hostname="host", + username="user", + chat_history=[ChatMessage(role="user", content="updated slash")], + ) + observer.save_sessions_to_individual_files([updated_slash], output_dir=tmp_path) + assert colliding_legacy.exists() + assert json.loads(colliding_legacy.read_text(encoding="utf-8"))["session_id"] == colon_entry.session_id + + def test_generated_filename_does_not_overwrite_matching_literal_session_id(self, tmp_path): + """A hash-suffixed generated name may also be another session's literal ID.""" + observer = AgentObserver(output_dir=tmp_path) + timestamp = datetime(2025, 6, 15, 10, 0, 0, tzinfo=timezone.utc) + lossy_id = "codex_a/b" + literal_id = observer._session_filename_id(lossy_id) + lossy_entry = AgentEvent( + timestamp=timestamp, + source="codex", + session_id=lossy_id, + hostname="host", + username="user", + chat_history=[ChatMessage(role="user", content="lossy")], + ) + literal_entry = AgentEvent( + timestamp=timestamp, + source="codex", + session_id=literal_id, + hostname="host", + username="user", + chat_history=[ChatMessage(role="user", content="literal")], + ) + + saved = observer.save_sessions_to_individual_files([lossy_entry, literal_entry], output_dir=tmp_path) + + assert len(saved) == 2 + assert saved[0] != saved[1] + assert {json.loads(path.read_text(encoding="utf-8"))["session_id"] for path in saved} == { + lossy_id, + literal_id, + } + assert observer.filter_entries_by_existing_files([lossy_entry, literal_entry], output_dir=tmp_path) == [] + + def test_older_revision_cannot_replace_newer_snapshot(self, tmp_path): + """A delayed writer must not regress an already exported session.""" + observer = AgentObserver(output_dir=tmp_path) + timestamp = datetime(2025, 6, 15, 10, 0, 0, tzinfo=timezone.utc) + newer = AgentEvent( + timestamp=timestamp, + source="copilot", + session_id="copilot_session1", + hostname="host", + username="user", + chat_history=[ChatMessage(role="user", content="newer")], + session_context={"last_event_at": "2025-06-16T12:30:01.000Z"}, + ) + older = AgentEvent( + timestamp=timestamp, + source="copilot", + session_id="copilot_session1", + hostname="host", + username="user", + chat_history=[ChatMessage(role="user", content="older")], + session_context={"last_event_at": "2025-06-16T12:30:00.000Z"}, + ) + + saved_path = observer.save_sessions_to_individual_files([newer], output_dir=tmp_path)[0] + assert observer.filter_entries_by_existing_files([older], output_dir=tmp_path) == [] + assert observer.save_sessions_to_individual_files([older], output_dir=tmp_path) == [] + assert json.loads(saved_path.read_text(encoding="utf-8"))["chat_history"][0]["content"] == "newer" + + def test_equal_event_timestamp_uses_event_count_to_prevent_stale_replacement(self, tmp_path): + """An older parse cannot overwrite a same-timestamp append-only snapshot.""" + observer = AgentObserver(output_dir=tmp_path) + timestamp = datetime(2025, 6, 15, 10, 0, 0, tzinfo=timezone.utc) + newer = AgentEvent( + timestamp=timestamp, + source="codex", + session_id="codex_session1", + hostname="host", + username="user", + chat_history=[ChatMessage(role="user", content="newer")], + session_context={"last_event_at": "2025-06-16T12:30:00.000Z", "event_count": 3}, + ) + stale = AgentEvent( + timestamp=timestamp, + source="codex", + session_id="codex_session1", + hostname="host", + username="user", + chat_history=[ChatMessage(role="user", content="stale")], + session_context={"last_event_at": "2025-06-16T12:30:00.000Z", "event_count": 2}, + ) + + saved_path = observer.save_sessions_to_individual_files([newer], output_dir=tmp_path)[0] + assert observer.save_sessions_to_individual_files([stale], output_dir=tmp_path) == [] + assert json.loads(saved_path.read_text(encoding="utf-8"))["chat_history"][0]["content"] == "newer" + + def test_leftover_legacy_file_cannot_hide_newer_stable_revision(self, tmp_path): + """Regression checks must use stored revisions, not moving filenames.""" + observer = AgentObserver(output_dir=tmp_path) + timestamp = datetime(2025, 6, 15, 10, 0, 0, tzinfo=timezone.utc) + newer = AgentEvent( + timestamp=timestamp, + source="copilot", + session_id="copilot_session1", + hostname="host", + username="user", + chat_history=[ChatMessage(role="user", content="newest")], + session_context={"last_event_at": "2025-06-18T12:30:00.000Z"}, + ) + stale = AgentEvent( + timestamp=timestamp, + source="copilot", + session_id="copilot_session1", + hostname="host", + username="user", + chat_history=[ChatMessage(role="user", content="stale")], + session_context={"last_event_at": "2025-06-17T12:30:00.000Z"}, + ) + stable_path = observer.save_sessions_to_individual_files([newer], output_dir=tmp_path)[0] + legacy_data = stale.get_non_null_fields() + legacy_data["timestamp"] = "2025-06-16T12:30:00+00:00" + legacy_data["session_context"]["last_event_at"] = "2025-06-16T12:30:00.000Z" + legacy_path = tmp_path / "adr.copilot_session1.20250620_123000.json" + legacy_path.write_text(json.dumps(legacy_data), encoding="utf-8") + + assert observer.filter_entries_by_existing_files([stale], output_dir=tmp_path) == [] + assert observer.save_sessions_to_individual_files([stale], output_dir=tmp_path) == [] + assert json.loads(stable_path.read_text(encoding="utf-8"))["chat_history"][0]["content"] == "newest" + assert legacy_path.exists() + + def test_session_lock_blocks_competing_writer_and_recovers(self, tmp_path): + """The OS releases a session lock cleanly for the next writer.""" + lock_path = tmp_path / ".adr.codex_session1.lock" + first_fd = AgentObserver._acquire_session_lock(lock_path) + try: + with pytest.raises(TimeoutError): + AgentObserver._acquire_session_lock(lock_path, timeout_seconds=0) + finally: + AgentObserver._release_session_lock(first_fd) + + second_fd = AgentObserver._acquire_session_lock(lock_path, timeout_seconds=0) + AgentObserver._release_session_lock(second_fd) + + def test_waiting_stale_writer_refreshes_target_after_lock(self, tmp_path): + """A writer must re-read output created after its initial directory scan.""" + observer = AgentObserver(output_dir=tmp_path) + competing_observer = AgentObserver(output_dir=tmp_path) + timestamp = datetime(2025, 6, 15, 10, 0, 0, tzinfo=timezone.utc) + stale = AgentEvent( + timestamp=timestamp, + source="codex", + session_id="codex_session1", + hostname="host", + username="user", + chat_history=[ChatMessage(role="user", content="stale")], + session_context={"last_event_at": "2025-06-16T12:30:00.000Z"}, + ) + newer = AgentEvent( + timestamp=timestamp, + source="codex", + session_id="codex_session1", + hostname="host", + username="user", + chat_history=[ChatMessage(role="user", content="newer")], + session_context={"last_event_at": "2025-06-16T12:30:01.000Z"}, + ) + acquire_lock = AgentObserver._acquire_session_lock + injected = False + + def inject_competing_write(lock_path): + nonlocal injected + if not injected: + injected = True + competing_observer.save_sessions_to_individual_files([newer], output_dir=tmp_path) + return acquire_lock(lock_path) + + with patch.object(observer, "_acquire_session_lock", side_effect=inject_competing_write): + assert observer.save_sessions_to_individual_files([stale], output_dir=tmp_path) == [] + + saved_path = next(tmp_path.glob("adr.codex_session1.*.json")) + assert json.loads(saved_path.read_text(encoding="utf-8"))["chat_history"][0]["content"] == "newer" + @patch("adr_sensor.observer.ClaudeParser") def test_ingest_all_handles_parser_errors(self, mock_claude_cls, tmp_path): """Test that ingest_all handles parser errors gracefully.""" diff --git a/Sensor/tests/test_parsers.py b/Sensor/tests/test_parsers.py index 816336f..139b684 100644 --- a/Sensor/tests/test_parsers.py +++ b/Sensor/tests/test_parsers.py @@ -12,8 +12,8 @@ from adr_sensor.parsers.claude_desktop_parser import ClaudeDesktopParser from adr_sensor.parsers.claude_parser import ClaudeParser from adr_sensor.parsers.cline_parser import ClineParser -from adr_sensor.parsers.copilot_parser import CopilotParser from adr_sensor.parsers.codex_parser import CodexParser +from adr_sensor.parsers.copilot_parser import CopilotParser from adr_sensor.parsers.cursor_parser import CursorParser from adr_sensor.parsers.opencode_parser import OpencodeParser from adr_sensor.parsers.warp_parser import WarpParser @@ -102,37 +102,6 @@ def test_truncate_large_arguments(self): assert len(result["long"]) < 2000 assert "[truncated" in result["long"] - def test_uses_latest_timestamp_for_resumed_session(self, tmp_path): - """Incremental export should see a resumed session as updated.""" - jsonl_file = tmp_path / "resumed.jsonl" - messages = [ - { - "type": "user", - "sessionId": "session1", - "timestamp": "2025-06-15T10:00:00Z", - "message": {"content": "first prompt"}, - }, - { - "type": "assistant", - "sessionId": "session1", - "timestamp": "2025-06-15T10:00:01Z", - "message": {"content": [{"type": "text", "text": "first reply"}]}, - }, - { - "type": "user", - "sessionId": "session1", - "timestamp": "2025-06-16T12:30:00Z", - "message": {"content": "continued next day"}, - }, - ] - with open(jsonl_file, "w", encoding="utf-8") as f: - for msg in messages: - f.write(json.dumps(msg) + "\n") - - entries = ClaudeParser().parse_jsonl_file(jsonl_file) - assert len(entries) == 1 - assert entries[0].timestamp == datetime(2025, 6, 16, 12, 30, 0, tzinfo=timezone.utc) - class TestClineParser: def test_parse_cline_log(self, tmp_path): @@ -466,8 +435,8 @@ def test_parse_no_directory(self): entries = parser.parse_all() assert entries == [] - def test_uses_latest_event_timestamp_for_resumed_session(self, tmp_path): - """Incremental export should use the last event, not only session_meta.""" + def test_keeps_session_start_timestamp_for_resumed_session(self, tmp_path): + """Resumed content must not change the timestamp used for file identity.""" jsonl_file = tmp_path / "resumed-codex.jsonl" events = [ { @@ -500,7 +469,10 @@ def test_uses_latest_event_timestamp_for_resumed_session(self, tmp_path): entry = CodexParser().parse_jsonl_file(jsonl_file) assert entry is not None - assert entry.timestamp == datetime(2025, 6, 16, 12, 30, 0, tzinfo=timezone.utc) + assert entry.timestamp == datetime(2025, 6, 15, 10, 0, 0, tzinfo=timezone.utc) + assert entry.chat_history[-1].content == "later follow-up" + assert entry.session_context["last_event_at"] == "2025-06-16T12:30:00+00:00" + assert entry.session_context["event_count"] == len(events) class TestCopilotParser: @@ -549,7 +521,10 @@ def test_parse_session_dir(self, tmp_path): "type": "user.message", "id": "user-1", "timestamp": "2026-08-10T10:00:05.000Z", - "data": {"content": "inspect the repo", "transformedContent": "...\ninspect the repo"}, + "data": { + "content": "inspect the repo", + "transformedContent": "...\ninspect the repo", + }, }, { "type": "assistant.message", @@ -590,15 +565,53 @@ def test_parse_session_dir(self, tmp_path): assert entry.session_id == "copilot_abc-session" assert entry.project_path == "C:\\repo" assert entry.model == "gpt-5.6-sol" - assert entry.timestamp == datetime(2026, 8, 10, 11, 0, 0, tzinfo=timezone.utc) + assert entry.timestamp == datetime(2026, 8, 10, 10, 0, 0, tzinfo=timezone.utc) assert [msg.role for msg in entry.chat_history] == ["user", "assistant"] tool = entry.chat_history[1].tools[0] assert tool.tool_name == "powershell" assert tool.result == "file1\nfile2" assert tool.status == "success" assert entry.session_context["workspace_metadata"]["updated_at"] == "2026-08-10T11:00:00.000Z" + assert entry.session_context["last_event_at"] == "2026-08-10T11:00:00+00:00" + assert entry.session_context["event_count"] == len(events) assert entry.session_context["transformed_user_messages"][0]["sequence_id"] == "user-1" + def test_tool_failure_uses_error_field_when_result_is_absent(self, tmp_path): + """Copilot records some failed tools with error but no result.""" + session_dir = tmp_path / "failed-tool" + session_dir.mkdir() + events = [ + { + "type": "session.start", + "timestamp": "2026-08-10T10:00:00.000Z", + "data": {"sessionId": "failed-tool", "startTime": "2026-08-10T10:00:00.000Z"}, + }, + { + "type": "user.message", + "timestamp": "2026-08-10T10:00:01.000Z", + "data": {"content": "read missing file"}, + }, + { + "type": "tool.execution_start", + "timestamp": "2026-08-10T10:00:02.000Z", + "data": {"toolCallId": "call-1", "toolName": "read_file", "arguments": {"path": "missing"}}, + }, + { + "type": "tool.execution_complete", + "timestamp": "2026-08-10T10:00:02.001Z", + "data": {"toolCallId": "call-1", "success": False, "error": "File not found"}, + }, + ] + with open(session_dir / "events.jsonl", "w", encoding="utf-8") as handle: + for event in events: + handle.write(json.dumps(event) + "\n") + + entry = CopilotParser(base_path=tmp_path).parse_session_dir(session_dir) + tool = [tool for message in entry.chat_history for tool in message.tools][0] + assert tool.status == "error" + assert tool.result is None + assert tool.error == "File not found" + def _build_warp_db(db_path: Path, conversations: list) -> None: """Create a synthetic warp.sqlite matching the schema WarpParser queries.