From 8f70c21f15249643cafd7571d2540ce322eea4a9 Mon Sep 17 00:00:00 2001 From: Baris Ozbas Date: Sat, 19 Sep 2026 12:53:26 +0200 Subject: [PATCH] fix(sensor): preserve Claude transcript content and incremental updates Recover complete JSONL records conservatively, preserve mixed text and tool results, correlate results with exact tool invocations, and distinguish child sessions. Track content changes so incremental capture includes resumed turns. Add synthetic parser and observer regression coverage and document limitations. --- Sensor/README.md | 26 ++ Sensor/adr_sensor/observer.py | 2 +- Sensor/adr_sensor/parsers/claude_parser.py | 271 +++++++++------ Sensor/tests/test_claude_parser.py | 378 +++++++++++++++++++++ Sensor/tests/test_observer.py | 94 ++++- 5 files changed, 655 insertions(+), 116 deletions(-) create mode 100644 Sensor/tests/test_claude_parser.py 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)