diff --git a/Sensor/README.md b/Sensor/README.md index cc6e5e1..7e3e92c 100644 --- a/Sensor/README.md +++ b/Sensor/README.md @@ -23,6 +23,32 @@ ADR Sensor is a Python library that collects telemetry from AI coding agents to | **opencode** | `opencode` | SQLite (`opencode.db`) or JSON tree | macOS, Linux | | **Gemini CLI** | `gemini` | JSONL journals + legacy JSON chats | macOS, Linux, Windows | +### Claude Code + +The `claude` source reads transcripts recursively under `~/.claude/projects/`, +including [subagent transcripts](https://code.claude.com/docs/en/sub-agents#resume-subagents) +under `//subagents/agent-.jsonl` and nested workflow +directories. Main sessions keep the `claude_` identity; subagents use +`claude__agent_` and include `parent_session_id` and `agent_id` +in `session_context` so their exports do not overwrite the parent conversation. + +String and text-block messages are retained, including user text accompanying +tool results. Results are matched by tool-call ID. Malformed records are skipped +without discarding surrounding messages. Complete JSON objects concatenated on +one physical line and NUL padding between objects are accepted; incomplete tails +are skipped without joining physical lines or repairing text inside a message. + +Each event includes `raw_log_path`, a stable conversation-start `timestamp`, and +`session_context.last_event_at` and `event_count` for incremental updates. A file +without any valid timestamp uses its modification time. `--save-sessions` updates +the saved snapshot when a tool completes or a conversation resumes, even within +the same timestamp second. All recorded branches are retained in file order. + +The default lookback is 14 days by file modification time. Existing limits still +apply: top-level tool argument strings and tool results are truncated at 1,000 +characters; non-text content blocks and separately spilled tool-output files are +not imported. Contract tests use synthetic transcripts and do not launch Claude. + ### Claude Desktop Agent Mode The `claude_desktop` source covers Claude Desktop's local agent mode (released as diff --git a/Sensor/adr_sensor/observer.py b/Sensor/adr_sensor/observer.py index b8546b0..f8fbdb3 100644 --- a/Sensor/adr_sensor/observer.py +++ b/Sensor/adr_sensor/observer.py @@ -73,7 +73,7 @@ class AgentObserver: "claude_desktop": ("Darwin", "Windows"), } - CONTENT_AWARE_INCREMENTAL_SOURCES = frozenset({"codex", "copilot", "dsh", "gemini"}) + CONTENT_AWARE_INCREMENTAL_SOURCES = frozenset({"claude", "codex", "copilot", "dsh", "gemini"}) def __init__(self, output_dir: Optional[Path] = None, max_age_days: Optional[int] = None): """Initialize the AgentObserver. diff --git a/Sensor/adr_sensor/parsers/claude_parser.py b/Sensor/adr_sensor/parsers/claude_parser.py index 856de38..32c9a93 100644 --- a/Sensor/adr_sensor/parsers/claude_parser.py +++ b/Sensor/adr_sensor/parsers/claude_parser.py @@ -9,9 +9,10 @@ """ import json +from dataclasses import replace from datetime import datetime, timedelta, timezone from pathlib import Path -from typing import Any, Dict, List, Optional +from typing import Any, Dict, Iterator, List, Optional, Tuple from ..schemas.agent_event_schema import AgentEvent, ChatMessage, ToolUsage from ..utils.string_utils import truncate_middle @@ -76,7 +77,7 @@ def _normalize_result_content(self, result_content: Any) -> str: text_parts = [] for item in result_content: if isinstance(item, dict): - if item.get("type") == "text" and "text" in item: + if item.get("type") == "text" and isinstance(item.get("text"), str): text_parts.append(item["text"]) return "\n".join(text_parts) @@ -96,61 +97,97 @@ def _truncate_large_arguments(self, arguments: Dict[str, Any]) -> Dict[str, Any] return truncated + @staticmethod + def _decode_jsonl_line(line: str) -> Iterator[Any]: + """Decode complete values on one physical line, retaining a valid prefix. + + NUL padding is accepted only between values, never inside JSON strings. + Stop at the first damaged value rather than searching its text for another + object or joining it to the next line of the transcript. + """ + decoder = json.JSONDecoder() + offset = 0 + while offset < len(line): + while offset < len(line) and line[offset] in " \t\r\n\0": + offset += 1 + if offset == len(line): + return + try: + value, offset = decoder.raw_decode(line, offset) + except (ValueError, RecursionError): + return + yield value + + @staticmethod + def _agent_id(obj: Dict[str, Any], file_path: Path) -> Optional[str]: + """Identify documented subagent paths, including nested workflow logs.""" + if file_path.stem.startswith("agent-") and any(parent.name == "subagents" for parent in file_path.parents): + return file_path.stem[len("agent-") :] or None + agent_id = obj.get("agentId") + return agent_id if isinstance(agent_id, str) and agent_id else None + def parse_jsonl_file(self, file_path: Path) -> List[AgentEvent]: - """Parse a single JSONL file.""" + """Parse a transcript without letting a malformed record discard its peers.""" entries = [] - sessions: Dict[str, Dict[str, Any]] = {} + sessions: Dict[Tuple[str, Optional[str]], Dict[str, Any]] = {} try: with open(file_path, encoding="utf-8") as file: - for line_num, line in enumerate(file): - line = line.strip() - if not line: - continue - try: - obj = json.loads(line) - except json.JSONDecodeError: - continue - - session_id = obj.get("sessionId") - if not session_id: - continue - - if session_id not in sessions: - sessions[session_id] = { - "messages": [], - "timestamp": None, - "project_path": obj.get("cwd"), - "model": None, - } - - if "timestamp" in obj: - try: - ts = normalize_timestamp(obj["timestamp"]) - if sessions[session_id]["timestamp"] is None or ts < sessions[session_id]["timestamp"]: - sessions[session_id]["timestamp"] = ts - except Exception: - pass - - if obj.get("type") == "assistant" and "message" in obj: - msg = obj["message"] - if "model" in msg: - sessions[session_id]["model"] = msg["model"] - - extracted_msg = self._extract_message_data(obj) - if extracted_msg: - sessions[session_id]["messages"].append(extracted_msg) - - del obj - - for session_id, session_data in sessions.items(): - entry = self._create_entry_from_extracted_session(session_id, session_data, file_path) - if entry and entry.has_meaningful_content(): - entries.append(entry) - - except Exception as e: + for line in file: + for obj in self._decode_jsonl_line(line): + if not isinstance(obj, dict): + continue + session_id = obj.get("sessionId") + if not isinstance(session_id, str) or not session_id: + continue + msg_type = obj.get("type") + if not isinstance(msg_type, str): + continue + if msg_type in ("user", "assistant"): + message = obj.get("message") + if not isinstance(message, dict) or not isinstance(message.get("content", ""), (str, list)): + continue + + agent_id = self._agent_id(obj, file_path) + session_key = (session_id, agent_id) + if session_key not in sessions: + sessions[session_key] = { + "messages": [], + "timestamp": None, + "last_event_at": None, + "event_count": 0, + "project_path": None, + "model": None, + "agent_id": agent_id, + } + session = sessions[session_key] + session["event_count"] += 1 + if isinstance(obj.get("cwd"), str) and not session["project_path"]: + session["project_path"] = obj["cwd"] + + if "timestamp" in obj and not isinstance(obj["timestamp"], bool): + try: + ts = normalize_timestamp(obj["timestamp"]) + session["timestamp"] = min(session["timestamp"] or ts, ts) + session["last_event_at"] = max(session["last_event_at"] or ts, ts) + except (TypeError, ValueError, OverflowError, OSError): + pass + + if msg_type == "assistant" and isinstance(obj["message"].get("model"), str): + session["model"] = obj["message"]["model"] + + extracted_msg = self._extract_message_data(obj) + if extracted_msg: + session["messages"].append(extracted_msg) + + except (OSError, UnicodeError) as e: print(f"[CLAUDE] Error reading {file_path}: {e}") + for (session_id, _), session_data in sessions.items(): + entry = self._create_entry_from_extracted_session(session_id, session_data, file_path) + if entry and entry.has_meaningful_content(): + entries.append(entry) + return entries def _extract_message_data(self, obj: Dict[str, Any]) -> Optional[Dict[str, Any]]: @@ -159,25 +196,34 @@ def _extract_message_data(self, obj: Dict[str, Any]) -> Optional[Dict[str, Any]] if msg_type not in ("user", "assistant"): return None + message = obj.get("message") + if not isinstance(message, dict): + return None + extracted: Dict[str, Any] = { "type": msg_type, - "uuid": obj.get("uuid"), - "parent_uuid": obj.get("parentUuid"), + "uuid": obj.get("uuid") if isinstance(obj.get("uuid"), str) else None, } - - if "message" not in obj: - return None - - message = obj["message"] + content = message.get("content", "") + text_parts = [] + if isinstance(content, str): + text_parts.append(content) + elif isinstance(content, list): + text_parts.extend( + item["text"] + for item in content + if isinstance(item, dict) and item.get("type") == "text" and isinstance(item.get("text"), str) + ) + extracted["content"] = "".join(text_parts) if msg_type == "user": - content = message.get("content", "") - tool_results = [] if isinstance(content, list): for item in content: if isinstance(item, dict) and item.get("type") == "tool_result": tool_use_id = item.get("tool_use_id") + if not isinstance(tool_use_id, str) or not tool_use_id: + continue result_content = item.get("content", "") if "toolUseResult" in obj and isinstance(obj["toolUseResult"], dict): result_content = obj["toolUseResult"].get("result", result_content) @@ -188,34 +234,26 @@ def _extract_message_data(self, obj: Dict[str, Any]) -> Optional[Dict[str, Any]] result_content = truncate_middle(result_content, max_length=1000, edge_chars=400) tool_results.append({"tool_use_id": tool_use_id, "result": result_content}) - if tool_results: - extracted["tool_results"] = tool_results - extracted["content"] = "" - elif isinstance(content, str): - extracted["content"] = content - else: - extracted["content"] = "" + extracted["tool_results"] = tool_results elif msg_type == "assistant": - content_items = message.get("content", []) - text_parts = [] tools = [] - - if isinstance(content_items, list): - for item in content_items: - if isinstance(item, dict): - if item.get("type") == "text": - text_parts.append(item.get("text", "")) - elif item.get("type") == "tool_use": - raw_input = item.get("input", {}) - truncated_input = self._truncate_large_arguments(raw_input) - tools.append({ - "id": item.get("id"), - "name": item.get("name", "unknown"), - "input": truncated_input, - }) - - extracted["content"] = "".join(text_parts) + if isinstance(content, list): + for item in content: + if not isinstance(item, dict) or item.get("type") != "tool_use": + continue + raw_input = item.get("input", {}) + name = item.get("name", "unknown") + if not isinstance(raw_input, dict) or not isinstance(name, str): + continue + tool_id = item.get("id") + tools.append( + { + "id": tool_id if isinstance(tool_id, str) else None, + "name": name, + "input": self._truncate_large_arguments(raw_input), + } + ) extracted["tools"] = tools return extracted @@ -225,15 +263,9 @@ def _create_entry_from_extracted_session( ) -> Optional[AgentEvent]: """Create an AgentEvent from pre-extracted session data.""" try: - entry = AgentEvent( - timestamp=session_data["timestamp"] or datetime.now(timezone.utc), - source="claude", - session_id=f"claude_{session_id}", - project_path=session_data["project_path"], - model=session_data["model"], - ) - - pending_tools: Dict[str, ToolUsage] = {} + chat_history: List[ChatMessage] = [] + # Store exact locations: distinct invocations can have equal fields. + pending_tools: Dict[str, Tuple[int, int]] = {} for i, msg_data in enumerate(session_data["messages"]): msg_type = msg_data["type"] @@ -246,28 +278,20 @@ def _create_entry_from_extracted_session( tool_use_id = tool_result.get("tool_use_id") result = tool_result.get("result") if tool_use_id in pending_tools: - old_tool = pending_tools[tool_use_id] - updated_tool = ToolUsage( - tool_name=old_tool.tool_name, - tool_type=old_tool.tool_type, - arguments=old_tool.arguments, + message_index, tool_index = pending_tools[tool_use_id] + old_message = chat_history[message_index] + new_tools = list(old_message.tools) + new_tools[tool_index] = replace( + new_tools[tool_index], result=result, status="success" if result else "unknown", ) - for msg in entry.chat_history: - if msg.role == "assistant": - for idx, t in enumerate(msg.tools): - if t == old_tool: - new_tools = list(msg.tools) - new_tools[idx] = updated_tool - object.__setattr__(msg, "tools", new_tools) - break - continue + chat_history[message_index] = replace(old_message, tools=new_tools) content = msg_data.get("content", "") if content: msg = ChatMessage(role="user", content=content, tools=[], sequence_id=sequence_id) - entry.chat_history.append(msg) + chat_history.append(msg) elif msg_type == "assistant": content = msg_data.get("content", "") @@ -283,7 +307,7 @@ def _create_entry_from_extracted_session( tools.append(tool) tool_id = tool_data.get("id") if tool_id: - pending_tools[tool_id] = tool + pending_tools[tool_id] = (len(chat_history), len(tools) - 1) if content or tools: msg = ChatMessage( @@ -292,9 +316,32 @@ def _create_entry_from_extracted_session( tools=tools, sequence_id=sequence_id, ) - entry.chat_history.append(msg) - - return entry + chat_history.append(msg) + + timestamp = session_data["timestamp"] + if timestamp is None: + timestamp = datetime.fromtimestamp(file_path.stat().st_mtime, tz=timezone.utc) + context = { + "last_event_at": (session_data["last_event_at"] or timestamp).isoformat(), + "event_count": session_data["event_count"], + } + entry_session_id = f"claude_{session_id}" + agent_id = session_data["agent_id"] + if agent_id: + context["parent_session_id"] = entry_session_id + context["agent_id"] = agent_id + entry_session_id += f"_agent_{agent_id}" + + return AgentEvent( + timestamp=timestamp, + source="claude", + session_id=entry_session_id, + chat_history=chat_history, + project_path=session_data["project_path"], + model=session_data["model"], + raw_log_path=str(file_path), + session_context=context, + ) except Exception as e: print(f"[CLAUDE] Error creating entry for session {session_id}: {e}") diff --git a/Sensor/tests/test_claude_parser.py b/Sensor/tests/test_claude_parser.py new file mode 100644 index 0000000..571d7f8 --- /dev/null +++ b/Sensor/tests/test_claude_parser.py @@ -0,0 +1,378 @@ +"""Regression tests for public Claude Code transcript formats using synthetic data.""" + +import json +from dataclasses import replace +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +import pytest + +from adr_sensor.parsers.claude_parser import ClaudeParser + + +def _record( + role: str, + content: Any, + *, + session_id: str = "session-1", + timestamp: str = "2026-09-01T10:00:00Z", + **fields: Any, +) -> dict: + return { + "type": role, + "sessionId": session_id, + "timestamp": timestamp, + "cwd": "/synthetic/project", + "message": {"content": content}, + **fields, + } + + +def _write_records(path: Path, records: list) -> Path: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("\n".join(json.dumps(record) for record in records) + "\n", encoding="utf-8") + return path + + +def _contents(entries: list) -> list: + return [message.content for entry in entries for message in entry.chat_history] + + +@pytest.mark.parametrize("malformed", [None, True, 42, "not an object", []]) +def test_nonobject_records_do_not_discard_surrounding_messages(tmp_path, malformed): + path = _write_records( + tmp_path / "session.jsonl", + [_record("user", "Before malformed record"), malformed, _record("assistant", "After malformed record")], + ) + + entries = ClaudeParser().parse_jsonl_file(path) + + assert len(entries) == 1 + assert _contents(entries) == ["Before malformed record", "After malformed record"] + + +@pytest.mark.parametrize("role", ["user", "assistant"]) +@pytest.mark.parametrize("message", [None, [], "not an object", 42]) +def test_malformed_message_objects_do_not_discard_surrounding_messages(tmp_path, role, message): + path = _write_records( + tmp_path / "session.jsonl", + [ + _record("user", "Before malformed message"), + _record(role, "unused", message=message), + _record("assistant", "After malformed message"), + ], + ) + + entries = ClaudeParser().parse_jsonl_file(path) + + assert len(entries) == 1 + assert _contents(entries) == ["Before malformed message", "After malformed message"] + + +@pytest.mark.parametrize("session_id", [None, "", [], {}, 42, True]) +def test_invalid_session_identifiers_are_isolated(tmp_path, session_id): + path = _write_records( + tmp_path / "session.jsonl", + [ + _record("user", "Before invalid identifier"), + _record("user", "Invalid session must be ignored", session_id=session_id), + _record("assistant", "After invalid identifier"), + ], + ) + + entries = ClaudeParser().parse_jsonl_file(path) + + assert [entry.session_id for entry in entries] == ["claude_session-1"] + assert _contents(entries) == ["Before invalid identifier", "After invalid identifier"] + + +@pytest.mark.parametrize("role", ["user", "assistant"]) +@pytest.mark.parametrize( + "content", + [ + "First paragraph.\nSecond paragraph.", + [{"type": "text", "text": "First paragraph.\n"}, {"type": "text", "text": "Second paragraph."}], + ], + ids=["string", "text-blocks"], +) +def test_preserves_string_and_text_block_messages(tmp_path, role, content): + path = _write_records(tmp_path / "session.jsonl", [_record(role, content, uuid="message-uuid")]) + + entries = ClaudeParser().parse_jsonl_file(path) + + assert len(entries) == 1 + assert len(entries[0].chat_history) == 1 + message = entries[0].chat_history[0] + assert message.role == role + assert message.content == "First paragraph.\nSecond paragraph." + assert message.sequence_id == "message-uuid" + + +@pytest.mark.parametrize("role", ["user", "assistant"]) +def test_malformed_content_blocks_do_not_hide_valid_text(tmp_path, role): + content = [ + None, + 42, + "not a block", + {"type": "text", "text": None}, + {"type": "text", "text": ["not text"]}, + {"type": "text", "text": "Visible text survives."}, + {"type": "image", "source": {"type": "base64", "data": "synthetic"}}, + ] + path = _write_records(tmp_path / "session.jsonl", [_record(role, content)]) + + entries = ClaudeParser().parse_jsonl_file(path) + + assert _contents(entries) == ["Visible text survives."] + + +def test_user_text_is_preserved_alongside_tool_results(tmp_path): + path = _write_records( + tmp_path / "session.jsonl", + [ + _record( + "assistant", + [{"type": "tool_use", "id": "read-1", "name": "Read", "input": {"file_path": "file.txt"}}], + ), + _record( + "user", + [ + {"type": "text", "text": "The file is ready. "}, + { + "type": "tool_result", + "tool_use_id": "read-1", + "content": [{"type": "text", "text": "first line"}, {"type": "text", "text": "second line"}], + }, + {"type": "text", "text": "Please continue."}, + ], + uuid="user-with-result", + ), + ], + ) + + entries = ClaudeParser().parse_jsonl_file(path) + + assert len(entries) == 1 + assistant, user = entries[0].chat_history + assert assistant.tools[0].result == "first line\nsecond line" + assert user.role == "user" + assert user.content == "The file is ready. Please continue." + assert user.sequence_id == "user-with-result" + + +@pytest.mark.parametrize("same_message", [True, False], ids=["same-assistant-message", "separate-assistant-messages"]) +def test_identical_tool_calls_keep_results_with_their_exact_call(tmp_path, same_message): + calls = [ + {"type": "tool_use", "id": tool_id, "name": "Read", "input": {"file_path": "same.txt"}} + for tool_id in ("first-call", "second-call") + ] + records = [_record("assistant", calls)] if same_message else [_record("assistant", [call]) for call in calls] + records.append( + _record( + "user", + [ + {"type": "tool_result", "tool_use_id": "second-call", "content": "Second call output"}, + {"type": "tool_result", "tool_use_id": "first-call", "content": "First call output"}, + ], + ) + ) + path = _write_records(tmp_path / "session.jsonl", records) + + entries = ClaudeParser().parse_jsonl_file(path) + + assert len(entries) == 1 + tools = [tool for message in entries[0].chat_history for tool in message.tools] + assert [tool.result for tool in tools] == ["First call output", "Second call output"] + assert [message.role for message in entries[0].chat_history] == ["assistant"] * (1 if same_message else 2) + + +def test_malformed_tool_result_blocks_do_not_discard_valid_results(tmp_path): + path = _write_records( + tmp_path / "session.jsonl", + [ + _record("assistant", [{"type": "tool_use", "id": "read-1", "name": "Read", "input": {}}]), + _record( + "user", + [ + {"type": "tool_result", "tool_use_id": [], "content": "Malformed identifier"}, + { + "type": "tool_result", + "tool_use_id": "read-1", + "content": [None, {"type": "text", "text": None}, {"type": "text", "text": "Valid result"}], + }, + {"type": "text", "text": "Visible user follow-up"}, + ], + ), + ], + ) + + entries = ClaudeParser().parse_jsonl_file(path) + + assert len(entries) == 1 + assert entries[0].chat_history[0].tools[0].result == "Valid result" + assert entries[0].chat_history[1].content == "Visible user follow-up" + + +def test_large_tool_arguments_and_results_remain_truncated(tmp_path): + long_text = "start-" + "x" * 2000 + "-finish" + path = _write_records( + tmp_path / "session.jsonl", + [ + _record( + "assistant", + [{"type": "tool_use", "id": "write-1", "name": "Write", "input": {"content": long_text}}], + ), + _record("user", [{"type": "tool_result", "tool_use_id": "write-1", "content": long_text}]), + ], + ) + + entries = ClaudeParser().parse_jsonl_file(path) + + tool = entries[0].chat_history[0].tools[0] + for text in (tool.arguments["content"], tool.result): + assert len(text) < len(long_text) + assert "[truncated" in text + assert text.startswith("start-") + assert text.endswith("-finish") + + +def test_parent_direct_child_and_nested_workflow_have_distinct_identities(tmp_path): + paths = { + "claude_parent": tmp_path / "project" / "parent.jsonl", + "claude_parent_agent_direct": tmp_path / "project" / "parent" / "subagents" / "agent-direct.jsonl", + "claude_parent_agent_nested": ( + tmp_path / "project" / "parent" / "subagents" / "workflows" / "run-1" / "agent-nested.jsonl" + ), + } + _write_records(paths["claude_parent"], [_record("user", "Parent conversation", session_id="parent")]) + _write_records( + paths["claude_parent_agent_direct"], + [_record("assistant", "Direct child conversation", session_id="parent", agentId="direct", isSidechain=True)], + ) + _write_records( + paths["claude_parent_agent_nested"], + [_record("assistant", "Nested workflow conversation", session_id="parent", isSidechain=True)], + ) + parser = ClaudeParser() + parser.base_path = tmp_path + + entries = parser.parse_all() + + assert {entry.session_id for entry in entries} == set(paths) + assert len({entry.uuid for entry in entries}) == 3 + for entry in entries: + assert entry.raw_log_path == str(paths[entry.session_id]) + if entry.session_id != "claude_parent": + assert entry.session_context["parent_session_id"] == "claude_parent" + assert entry.session_context["agent_id"] == entry.session_id.rsplit("_", 1)[-1] + + +def test_session_metadata_uses_earliest_and_latest_valid_timestamps(tmp_path): + path = _write_records( + tmp_path / "session.jsonl", + [ + _record("assistant", "First physical record", timestamp="2026-09-01T10:05:00Z"), + _record("user", "Earlier session start", timestamp="2026-09-01T10:00:00Z"), + { + "type": "system", + "sessionId": "session-1", + "timestamp": "2026-09-01T10:10:00Z", + "subtype": "turn_duration", + }, + _record("assistant", "Bad timestamp is isolated", timestamp="not-a-timestamp"), + ], + ) + + entries = ClaudeParser().parse_jsonl_file(path) + + assert len(entries) == 1 + entry = entries[0] + assert entry.timestamp == datetime(2026, 9, 1, 10, tzinfo=timezone.utc) + assert entry.session_context["last_event_at"] == "2026-09-01T10:10:00+00:00" + assert entry.session_context["event_count"] == 4 + assert entry.raw_log_path == str(path) + assert entry.project_path == "/synthetic/project" + + +def test_growing_session_keeps_start_and_updates_latest_event_metadata(tmp_path): + path = tmp_path / "session.jsonl" + records = [_record("user", "Initial user message")] + _write_records(path, records) + parser = ClaudeParser() + first = parser.parse_jsonl_file(path)[0] + records.append(_record("assistant", "Later assistant response", timestamp="2026-09-01T10:20:00Z")) + _write_records(path, records) + + grown = parser.parse_jsonl_file(path)[0] + + assert grown.timestamp == first.timestamp + assert grown.session_id == first.session_id + assert first.session_context["last_event_at"] == "2026-09-01T10:00:00+00:00" + assert grown.session_context["last_event_at"] == "2026-09-01T10:20:00+00:00" + assert first.session_context["event_count"] == 1 + assert grown.session_context["event_count"] == 2 + assert _contents([grown]) == ["Initial user message", "Later assistant response"] + + +def test_event_identity_is_built_from_completed_chat_history(tmp_path): + path = _write_records( + tmp_path / "session.jsonl", + [ + _record("user", "Inspect the project"), + _record("assistant", [{"type": "tool_use", "id": "call", "name": "Read", "input": {}}]), + _record("user", [{"type": "tool_result", "tool_use_id": "call", "content": "Tool output"}]), + ], + ) + entry = ClaudeParser().parse_jsonl_file(path)[0] + + assert entry.uuid == replace(entry).uuid + assert entry.uuid != replace(entry, chat_history=[]).uuid + + +@pytest.mark.parametrize("separator", ["", " ", "\x00", "\x00 \x00"]) +def test_decodes_concatenated_objects_and_nul_padding_on_one_physical_line(tmp_path, separator): + path = tmp_path / "session.jsonl" + records = [ + _record("user", 'A message with braces { } and an escaped quote: "hello"'), + _record("assistant", "Second concatenated message"), + ] + path.write_text("\x00 " + separator.join(json.dumps(record) for record in records) + " \x00\n", encoding="utf-8") + + entries = ClaudeParser().parse_jsonl_file(path) + + assert _contents(entries) == [record["message"]["content"] for record in records] + assert entries[0].session_context["event_count"] == 2 + + +def test_complete_prefix_survives_incomplete_suffix_and_next_line_is_independent(tmp_path): + path = tmp_path / "session.jsonl" + path.write_text( + json.dumps(_record("user", "Complete prefix survives")) + + '{"type":"assistant","sessionId":"session-1","message":\n' + + json.dumps(_record("assistant", "Next physical line survives")) + + "\n", + encoding="utf-8", + ) + + entries = ClaudeParser().parse_jsonl_file(path) + + assert _contents(entries) == ["Complete prefix survives", "Next physical line survives"] + assert entries[0].session_context["event_count"] == 2 + + +def test_incomplete_records_are_never_joined_across_physical_lines(tmp_path): + path = tmp_path / "session.jsonl" + split_record = json.dumps(_record("user", "This split record must not become a message")) + path.write_text( + split_record.replace('"content": ', '"content": \n', 1) + + "\n" + + json.dumps(_record("assistant", "Independent complete message")) + + "\n", + encoding="utf-8", + ) + + entries = ClaudeParser().parse_jsonl_file(path) + + assert _contents(entries) == ["Independent complete message"] + assert entries[0].session_context["event_count"] == 1 diff --git a/Sensor/tests/test_observer.py b/Sensor/tests/test_observer.py index fabaff9..2ccdd96 100644 --- a/Sensor/tests/test_observer.py +++ b/Sensor/tests/test_observer.py @@ -9,6 +9,7 @@ import pytest from adr_sensor.observer import AgentObserver +from adr_sensor.parsers.claude_parser import ClaudeParser from adr_sensor.schemas.agent_event_schema import AgentEvent, ChatMessage, ToolUsage @@ -194,8 +195,6 @@ def test_filter_entries_by_existing_files(self, tmp_path): # Create an existing session file existing_file = tmp_path / "adr.claude_session1.20250615_103000.json" - existing_file.write_text("{}") - entries = [ AgentEvent( timestamp=datetime(2025, 6, 15, 10, 30, 0, tzinfo=timezone.utc), @@ -214,12 +213,101 @@ def test_filter_entries_by_existing_files(self, tmp_path): chat_history=[ChatMessage(role="user", content="new")], ), ] + existing_file.write_text(json.dumps(entries[0].get_non_null_fields()), encoding="utf-8") filtered = observer.filter_entries_by_existing_files(entries, output_dir=tmp_path) - # session1 has same timestamp, should be filtered; session2 is new + # session1 is unchanged, should be filtered; session2 is new assert len(filtered) == 1 assert filtered[0].session_id == "claude_session2" + def test_claude_exports_refresh_after_tool_results_and_resumed_turns(self, tmp_path): + transcript = tmp_path / "session.jsonl" + output_dir = tmp_path / "exports" + observer = AgentObserver(output_dir=output_dir) + records = [ + { + "type": "assistant", + "sessionId": "session", + "timestamp": "2026-09-19T10:00:00.100000Z", + "message": { + "content": [{"type": "tool_use", "id": "call", "name": "Bash", "input": {"command": "pwd"}}] + }, + } + ] + + def parse_snapshot(): + transcript.write_text("\n".join(json.dumps(record) for record in records), encoding="utf-8") + return ClaudeParser().parse_jsonl_file(transcript)[0] + + pending = parse_snapshot() + saved = observer.save_sessions_to_individual_files([pending], output_dir=output_dir) + assert len(saved) == 1 + assert observer.filter_entries_by_existing_files([pending], output_dir=output_dir) == [] + + records.append( + { + "type": "user", + "sessionId": "session", + "timestamp": "2026-09-19T10:00:00.900000Z", + "message": { + "content": [{"type": "tool_result", "tool_use_id": "call", "content": "/synthetic/project"}] + }, + } + ) + completed = parse_snapshot() + assert completed.timestamp == pending.timestamp + assert observer.filter_entries_by_existing_files([completed], output_dir=output_dir) == [completed] + assert observer.save_sessions_to_individual_files([completed], output_dir=output_dir) == saved + assert json.loads(saved[0].read_text())["chat_history"][0]["tools"][0]["result"] == "/synthetic/project" + + records.append( + { + "type": "user", + "sessionId": "session", + "timestamp": "2026-09-20T11:00:00Z", + "message": {"content": [{"type": "text", "text": "Continue with the next task"}]}, + } + ) + resumed = parse_snapshot() + assert resumed.timestamp == pending.timestamp + assert observer.filter_entries_by_existing_files([resumed], output_dir=output_dir) == [resumed] + assert observer.save_sessions_to_individual_files([resumed], output_dir=output_dir) == saved + assert observer.filter_entries_by_existing_files([resumed], output_dir=output_dir) == [] + assert observer.save_sessions_to_individual_files([completed], output_dir=output_dir) == [] + persisted = json.loads(saved[0].read_text()) + assert persisted["chat_history"][-1]["content"] == "Continue with the next task" + assert list(output_dir.glob("adr.*.json")) == saved + + def test_claude_exports_keep_parent_and_subagent_snapshots_separate(self, tmp_path): + project = tmp_path / "project" + subagents = project / "parent" / "subagents" + subagents.mkdir(parents=True) + for path, text in [ + (project / "parent.jsonl", "Inspect this project"), + (subagents / "agent-child.jsonl", "Inspect this subtask"), + ]: + path.write_text( + json.dumps( + { + "type": "user", + "sessionId": "parent", + "timestamp": "2026-09-19T10:00:00Z", + "message": {"content": text}, + } + ), + encoding="utf-8", + ) + parser = ClaudeParser() + parser.base_path = project + entries = parser.parse_all() + observer = AgentObserver(output_dir=tmp_path / "exports") + saved = observer.save_sessions_to_individual_files(entries, output_dir=observer.output_dir) + + assert len(saved) == 2 + snapshots = {json.loads(path.read_text())["session_id"]: json.loads(path.read_text()) for path in saved} + assert snapshots["claude_parent"]["chat_history"][0]["content"] == "Inspect this project" + assert snapshots["claude_parent_agent_child"]["chat_history"][0]["content"] == "Inspect this subtask" + 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)