Coverage for astrocyte/harness/transcript.py: 99%
150 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-10-29 22:10 +0000
1"""Turn an agent transcript (JSONL) into conversation turns to capture.
3Claude Code's format is described first; Antigravity's further down.
5What is kept, per the transcript structure measured on Claude Code 2.1:
7* **User prompts** — ``type == "user"`` lines with string content whose
8 ``origin.kind`` is ``"human"`` (older transcripts lack ``origin``; untagged
9 plain text counts as human). Task notifications, command wrappers and tool
10 results are not the user speaking.
11* **Assistant prose** — ``text`` blocks only. ``thinking`` is internal and
12 ``tool_use`` / ``tool_result`` are bulky and can carry file contents.
13* **Files the turn touched** — the paths of the assistant's file tools
14 (``Read``, ``Edit``, ``Write``, ``MultiEdit``, ``NotebookEdit``), not their
15 contents: what ties a captured turn to the code it was about.
17Hook output (including memories *we* inject) lives in separate ``attachment``
18lines, so it is never re-captured: there is no feedback loop.
20Reading is incremental from a byte offset. The returned offset never passes a
21half-written line or a turn whose answer has not arrived, so the next read
22resumes exactly where this one stopped — no turn is lost or captured twice.
23"""
25from __future__ import annotations
27import json
28import re
29from dataclasses import dataclass, field
30from datetime import datetime
31from pathlib import Path
33# Long messages are clipped: memory wants the gist, not a 40 KB paste.
34MAX_USER_CHARS = 2_000
35MAX_ASSISTANT_CHARS = 4_000
37_REMINDER = re.compile(r"<system-reminder>.*?</system-reminder>", re.DOTALL)
38_WRAPPER = re.compile(r"^\s*<(command-name|command-message|command-args|local-command-[a-z]+|task-notification)>")
41@dataclass
42class Turn:
43 user: str
44 assistant: list[str] = field(default_factory=list)
45 started_at: datetime | None = None
46 files: list[str] = field(default_factory=list) # touched, in first-touch order
48 def render(self) -> str:
49 """Conversation-engine input (``**role**: text``)."""
50 answer = _clip("\n\n".join(self.assistant), MAX_ASSISTANT_CHARS)
51 return f"**user**: {_clip(self.user, MAX_USER_CHARS)}\n\n**assistant**: {answer}"
54def _clip(text: str, limit: int) -> str:
55 text = text.strip()
56 return text if len(text) <= limit else text[: limit - 1].rstrip() + "…"
59def _parse_time(raw: object) -> datetime | None:
60 if not isinstance(raw, str):
61 return None
62 try:
63 return datetime.fromisoformat(raw.replace("Z", "+00:00"))
64 except ValueError:
65 return None
68def _human_prompt(line: dict) -> str | None:
69 if line.get("type") != "user" or line.get("isSidechain") or line.get("isMeta"):
70 return None
71 content = (line.get("message") or {}).get("content")
72 if isinstance(content, list): # text blocks typed by a human; tool results are not
73 if any(isinstance(b, dict) and b.get("type") == "tool_result" for b in content):
74 return None
75 content = "\n".join(b.get("text", "") for b in content if isinstance(b, dict) and b.get("type") == "text")
76 if not isinstance(content, str):
77 return None
78 origin = line.get("origin")
79 if isinstance(origin, dict) and origin.get("kind") not in (None, "human"):
80 return None
81 if _WRAPPER.match(content):
82 return None
83 text = _REMINDER.sub("", content).strip()
84 return text or None
87def _assistant_text(line: dict) -> list[str]:
88 if line.get("type") != "assistant" or line.get("isSidechain"):
89 return []
90 content = (line.get("message") or {}).get("content")
91 if isinstance(content, str):
92 return [content] if content.strip() else []
93 if not isinstance(content, list):
94 return []
95 return [b["text"] for b in content if isinstance(b, dict) and b.get("type") == "text" and b.get("text", "").strip()]
98_FILE_TOOLS = {"Read": "file_path", "Edit": "file_path", "Write": "file_path", "MultiEdit": "file_path",
99 "NotebookEdit": "notebook_path"}
102def _touched_files(line: dict) -> list[str]:
103 """Paths the assistant's file tools name in this line (Claude Code)."""
104 if line.get("type") != "assistant" or line.get("isSidechain"):
105 return []
106 content = (line.get("message") or {}).get("content")
107 if not isinstance(content, list):
108 return []
109 found = []
110 for block in content:
111 if isinstance(block, dict) and block.get("type") == "tool_use" and (key := _FILE_TOOLS.get(block.get("name"))):
112 path = (block.get("input") or {}).get(key)
113 if isinstance(path, str) and path:
114 found.append(path)
115 return found
118# ── Antigravity (transcript.jsonl under brain/<conversation>/) ───────────
119#
120# Steps, measured on agy 1.2 and the Antigravity app: the user's message is a
121# USER_INPUT step whose content wraps the text in <USER_REQUEST> (followed by
122# <ADDITIONAL_METADATA> the app adds); replies are PLANNER_RESPONSE steps with
123# string content (tool-calling ones have none). Context our hooks inject is an
124# EPHEMERAL_MESSAGE step, so it is never re-captured.
126_USER_REQUEST = re.compile(r"<USER_REQUEST>\s*(.*?)\s*</USER_REQUEST>", re.DOTALL)
129def antigravity_prompt(line: dict) -> str | None:
130 if line.get("type") != "USER_INPUT" or line.get("source") not in (None, "USER_EXPLICIT"):
131 return None
132 content = line.get("content")
133 if not isinstance(content, str):
134 return None
135 m = _USER_REQUEST.search(content)
136 text = (m.group(1) if m else content).strip()
137 return text or None
140# File tools by name, and the argument holding the path. The envelope —
141# PLANNER_RESPONSE ``tool_calls: [{"name", "args"}]`` with each argument value
142# JSON-encoded as a string — is measured on agy 1.2; the tool names and their
143# path arguments come from Antigravity's hook documentation (no file-tool
144# call has been recorded here yet).
145_ANTIGRAVITY_FILE_TOOLS = {"view_file": "AbsolutePath", "write_to_file": "TargetFile",
146 "replace_file_content": "TargetFile", "multi_replace_file_content": "TargetFile"}
149def _antigravity_files(line: dict) -> list[str]:
150 if line.get("type") != "PLANNER_RESPONSE" or not isinstance(line.get("tool_calls"), list):
151 return []
152 found = []
153 for call in line["tool_calls"]:
154 if not isinstance(call, dict) or not (key := _ANTIGRAVITY_FILE_TOOLS.get(call.get("name"))):
155 continue
156 value = (call.get("args") or {}).get(key)
157 if isinstance(value, str) and value.startswith('"'):
158 try:
159 value = json.loads(value)
160 except ValueError:
161 continue
162 if isinstance(value, str) and value:
163 found.append(value)
164 return found
167def _antigravity_reply(line: dict) -> list[str]:
168 if line.get("type") != "PLANNER_RESPONSE":
169 return []
170 content = line.get("content")
171 return [content] if isinstance(content, str) and content.strip() else []
174def read_new_turns(path: str | Path, offset: int = 0, *, antigravity: bool = False) -> tuple[list[Turn], int]:
175 """Complete turns after ``offset``, and the offset to resume from."""
176 p = Path(path)
177 try:
178 size = p.stat().st_size
179 except OSError:
180 return [], offset
181 if offset > size: # transcript was rewritten or truncated; start over
182 offset = 0
183 with p.open("rb") as fh:
184 fh.seek(offset)
185 data = fh.read()
187 turns: list[Turn] = []
188 current: Turn | None = None
189 current_start = offset # where the in-progress turn began
190 resume = offset
191 pos = offset
192 for raw in data.splitlines(keepends=True):
193 if not raw.endswith(b"\n"):
194 break # half-written line: leave it for the next read
195 line_start, pos = pos, pos + len(raw) # pos only ever covers complete lines
196 try:
197 line = json.loads(raw)
198 except (json.JSONDecodeError, UnicodeDecodeError):
199 resume = pos if current is None else resume
200 continue
201 if not isinstance(line, dict):
202 continue
203 prompt = antigravity_prompt(line) if antigravity else _human_prompt(line)
204 if prompt is not None:
205 if current is not None and current.assistant:
206 turns.append(current)
207 resume = line_start
208 when = line.get("created_at") if antigravity else line.get("timestamp")
209 current, current_start = Turn(user=prompt, started_at=_parse_time(when)), line_start
210 continue
211 texts = _antigravity_reply(line) if antigravity else _assistant_text(line)
212 if texts and current is not None:
213 current.assistant.extend(texts)
214 if current is not None:
215 touched = _antigravity_files(line) if antigravity else _touched_files(line)
216 current.files.extend(f for f in touched if f not in current.files)
217 if current is None:
218 resume = pos # nothing pending: safe to move past this line
220 if current is not None and current.assistant:
221 # Stop fires when the assistant has finished, so the last turn is done.
222 turns.append(current)
223 resume = pos
224 elif current is not None:
225 resume = current_start # prompt without an answer yet: re-read next time
226 return turns, resume