feat(codex): add structured tool call retention from Codex rollout files (#778)
Parse all Codex rollout item types (function_call, local_shell_call, exec_command_end, patch_apply_end, mcp_tool_call_end, web_search_call) into structured JSON content blocks matching Claude Code's format. Enabled by default via retainToolCalls setting.
This commit is contained in:
Nicolò Boschi
GitHub
parent
a915584e39
commit
3461398b52
4 changed files with 722 additions and 26 deletions
|
|
@ -2,13 +2,22 @@
|
|||
|
||||
Adapts Openclaw/Claude Code content processing for Codex's transcript format.
|
||||
|
||||
Codex transcript format (JSONL):
|
||||
{"session_id": "...", "ts": 1234567890, "msg": {"type": "user_message", "message": "..."}}
|
||||
Codex rollout format (JSONL):
|
||||
Each line is a RolloutLine: {"timestamp": "...", "type": "<item_type>", ...}
|
||||
Item types: session_meta, response_item, event_msg, turn_context, compacted.
|
||||
|
||||
EventMsg types (from codex-rs/protocol/src/protocol.rs, serde snake_case):
|
||||
- user_message → role: user
|
||||
- agent_message → role: assistant
|
||||
- task_started, task_complete, exec, etc. → skipped
|
||||
ResponseItem types we care about:
|
||||
- message (role: user/assistant) — text content
|
||||
- local_shell_call — shell command invocations
|
||||
- function_call — tool/function calls with name + arguments
|
||||
- function_call_output — tool return values
|
||||
- custom_tool_call / custom_tool_call_output — freeform tools
|
||||
- web_search_call — web search invocations
|
||||
|
||||
EventMsg types (Extended persistence mode):
|
||||
- exec_command_end — shell command results with stdout/stderr
|
||||
- patch_apply_end — code patch application results
|
||||
- mcp_tool_call_end — MCP tool call results
|
||||
"""
|
||||
|
||||
import json
|
||||
|
|
@ -16,6 +25,9 @@ import os
|
|||
import re
|
||||
from datetime import datetime, timezone
|
||||
|
||||
# Maximum length for tool output content in JSON format.
|
||||
_MAX_TOOL_OUTPUT_CHARS = 2000
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Memory tag stripping (anti-feedback-loop)
|
||||
|
|
@ -38,20 +50,47 @@ def strip_memory_tags(content: str) -> str:
|
|||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def read_transcript(transcript_path: str) -> list:
|
||||
"""Read a Codex JSONL transcript and return list of {role, content} dicts.
|
||||
def read_transcript(transcript_path: str, include_tool_calls: bool = False) -> list:
|
||||
"""Read a Codex JSONL transcript and return list of message dicts.
|
||||
|
||||
Codex disk format (rollout-*.jsonl):
|
||||
User: {"type":"response_item","payload":{"type":"message","role":"user",
|
||||
"content":[{"type":"input_text","text":"..."}]}}
|
||||
Assistant: {"type":"response_item","payload":{"type":"message","role":"assistant",
|
||||
"content":[{"type":"output_text","text":"..."}],"phase":"final_answer"}}
|
||||
When include_tool_calls is False (legacy mode), returns simple
|
||||
{role, content} dicts with text-only content.
|
||||
|
||||
When include_tool_calls is True, returns richer message dicts with
|
||||
structured content blocks (matching Claude Code's JSON format):
|
||||
- {"role": "user", "content": [{"type": "text", "text": "..."}]}
|
||||
- {"role": "assistant", "content": [
|
||||
{"type": "text", "text": "..."},
|
||||
{"type": "tool_use", "name": "shell", "input": {"command": ["ls"]}},
|
||||
{"type": "tool_result", "content": "file1.txt\\nfile2.txt"}
|
||||
]}
|
||||
|
||||
Codex rollout format (rollout-*.jsonl):
|
||||
ResponseItems:
|
||||
message: text messages with role
|
||||
local_shell_call: shell command invocations
|
||||
function_call: tool calls with name + arguments
|
||||
function_call_output: tool return values
|
||||
custom_tool_call: freeform tool calls
|
||||
custom_tool_call_output: freeform tool results
|
||||
EventMsgs:
|
||||
exec_command_end: shell results with stdout/stderr/exit_code
|
||||
patch_apply_end: code patch results
|
||||
mcp_tool_call_end: MCP tool results
|
||||
|
||||
Flat format for testing:
|
||||
{"role": "user", "content": "..."}
|
||||
"""
|
||||
if not transcript_path or not os.path.isfile(transcript_path):
|
||||
return []
|
||||
|
||||
if include_tool_calls:
|
||||
return _read_transcript_rich(transcript_path)
|
||||
return _read_transcript_text(transcript_path)
|
||||
|
||||
|
||||
def _read_transcript_text(transcript_path: str) -> list:
|
||||
"""Legacy text-only transcript reader."""
|
||||
messages = []
|
||||
try:
|
||||
with open(transcript_path) as f:
|
||||
|
|
@ -91,6 +130,242 @@ def read_transcript(transcript_path: str) -> list:
|
|||
return messages
|
||||
|
||||
|
||||
def _read_transcript_rich(transcript_path: str) -> list:
|
||||
"""Rich transcript reader that preserves tool calls as structured content blocks.
|
||||
|
||||
Collects all response_items and event_msgs into a sequence of messages
|
||||
with structured content blocks. Tool calls and their outputs are grouped
|
||||
under the assistant role.
|
||||
"""
|
||||
messages = []
|
||||
# Buffer for collecting assistant content blocks between user messages
|
||||
assistant_blocks = []
|
||||
|
||||
def _flush_assistant():
|
||||
nonlocal assistant_blocks
|
||||
if assistant_blocks:
|
||||
messages.append({"role": "assistant", "content": assistant_blocks})
|
||||
assistant_blocks = []
|
||||
|
||||
try:
|
||||
with open(transcript_path) as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
entry = json.loads(line)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
|
||||
# Flat format (testing / future compatibility)
|
||||
if "role" in entry and "content" in entry:
|
||||
content = entry["content"]
|
||||
if entry["role"] == "user":
|
||||
_flush_assistant()
|
||||
if isinstance(content, str):
|
||||
content = [{"type": "text", "text": content}]
|
||||
messages.append({"role": "user", "content": content})
|
||||
elif entry["role"] == "assistant":
|
||||
if isinstance(content, str):
|
||||
content = [{"type": "text", "text": content}]
|
||||
if isinstance(content, list):
|
||||
assistant_blocks.extend(content)
|
||||
else:
|
||||
assistant_blocks.append({"type": "text", "text": str(content)})
|
||||
continue
|
||||
|
||||
item_type = entry.get("type")
|
||||
|
||||
# --- response_item ---
|
||||
if item_type == "response_item":
|
||||
payload = entry.get("payload", {})
|
||||
ptype = payload.get("type")
|
||||
|
||||
if ptype == "message":
|
||||
role = payload.get("role", "")
|
||||
if role == "user":
|
||||
_flush_assistant()
|
||||
text = _extract_text_from_content_blocks(payload.get("content", []))
|
||||
if text:
|
||||
messages.append({"role": "user", "content": [{"type": "text", "text": text}]})
|
||||
elif role == "assistant":
|
||||
# Only include final_answer (not reasoning/intermediary)
|
||||
if payload.get("phase") != "final_answer":
|
||||
continue
|
||||
text = _extract_text_from_content_blocks(payload.get("content", []))
|
||||
if text:
|
||||
assistant_blocks.append({"type": "text", "text": text})
|
||||
|
||||
elif ptype == "local_shell_call":
|
||||
action = payload.get("action", {})
|
||||
command = action.get("command", [])
|
||||
assistant_blocks.append({
|
||||
"type": "tool_use",
|
||||
"name": "shell",
|
||||
"input": {"command": command},
|
||||
})
|
||||
|
||||
elif ptype == "function_call":
|
||||
name = payload.get("name", "unknown")
|
||||
arguments = payload.get("arguments", "{}")
|
||||
try:
|
||||
inp = json.loads(arguments)
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
inp = {"raw": arguments}
|
||||
assistant_blocks.append({
|
||||
"type": "tool_use",
|
||||
"name": name,
|
||||
"input": inp,
|
||||
})
|
||||
|
||||
elif ptype == "function_call_output":
|
||||
output = payload.get("output", "")
|
||||
output_text = _extract_function_output_text(output)
|
||||
if output_text:
|
||||
assistant_blocks.append({
|
||||
"type": "tool_result",
|
||||
"content": _truncate(output_text),
|
||||
})
|
||||
|
||||
elif ptype == "custom_tool_call":
|
||||
name = payload.get("name", "unknown")
|
||||
inp_str = payload.get("input", "{}")
|
||||
try:
|
||||
inp = json.loads(inp_str)
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
inp = {"raw": inp_str}
|
||||
assistant_blocks.append({
|
||||
"type": "tool_use",
|
||||
"name": name,
|
||||
"input": inp,
|
||||
})
|
||||
|
||||
elif ptype == "custom_tool_call_output":
|
||||
output = payload.get("output", "")
|
||||
output_text = _extract_function_output_text(output)
|
||||
if output_text:
|
||||
assistant_blocks.append({
|
||||
"type": "tool_result",
|
||||
"content": _truncate(output_text),
|
||||
})
|
||||
|
||||
elif ptype == "web_search_call":
|
||||
action = payload.get("action", {})
|
||||
query = action.get("query", "") if isinstance(action, dict) else ""
|
||||
if query:
|
||||
assistant_blocks.append({
|
||||
"type": "tool_use",
|
||||
"name": "web_search",
|
||||
"input": {"query": query},
|
||||
})
|
||||
|
||||
# --- event_msg ---
|
||||
elif item_type == "event_msg":
|
||||
payload = entry.get("payload", {})
|
||||
ptype = payload.get("type")
|
||||
|
||||
if ptype == "exec_command_end":
|
||||
command = payload.get("command", [])
|
||||
output = payload.get("aggregated_output", "")
|
||||
exit_code = payload.get("exit_code")
|
||||
status = payload.get("status", "")
|
||||
# Add as tool_use + tool_result pair
|
||||
assistant_blocks.append({
|
||||
"type": "tool_use",
|
||||
"name": "shell",
|
||||
"input": {"command": command},
|
||||
})
|
||||
result_parts = []
|
||||
if output:
|
||||
result_parts.append(output)
|
||||
if exit_code is not None and exit_code != 0:
|
||||
result_parts.append(f"exit_code: {exit_code}")
|
||||
if status and status != "completed":
|
||||
result_parts.append(f"status: {status}")
|
||||
if result_parts:
|
||||
assistant_blocks.append({
|
||||
"type": "tool_result",
|
||||
"content": _truncate("\n".join(result_parts)),
|
||||
})
|
||||
|
||||
elif ptype == "patch_apply_end":
|
||||
changes = payload.get("changes", [])
|
||||
status = payload.get("status", "")
|
||||
if changes:
|
||||
assistant_blocks.append({
|
||||
"type": "tool_use",
|
||||
"name": "patch",
|
||||
"input": {"changes": changes},
|
||||
})
|
||||
if status:
|
||||
assistant_blocks.append({
|
||||
"type": "tool_result",
|
||||
"content": f"status: {status}",
|
||||
})
|
||||
|
||||
elif ptype == "mcp_tool_call_end":
|
||||
result = payload.get("result", {})
|
||||
result_text = ""
|
||||
if isinstance(result, dict):
|
||||
content_items = result.get("content", [])
|
||||
if isinstance(content_items, list):
|
||||
texts = []
|
||||
for item in content_items:
|
||||
if isinstance(item, dict) and item.get("type") == "text":
|
||||
texts.append(item.get("text", ""))
|
||||
result_text = "\n".join(texts)
|
||||
elif isinstance(content_items, str):
|
||||
result_text = content_items
|
||||
elif isinstance(result, str):
|
||||
result_text = result
|
||||
if result_text:
|
||||
assistant_blocks.append({
|
||||
"type": "tool_result",
|
||||
"content": _truncate(result_text),
|
||||
})
|
||||
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
_flush_assistant()
|
||||
return messages
|
||||
|
||||
|
||||
def _extract_text_from_content_blocks(content_blocks: list) -> str:
|
||||
"""Extract text from Codex content blocks (input_text/output_text)."""
|
||||
text_parts = []
|
||||
for block in content_blocks:
|
||||
if isinstance(block, dict) and block.get("type") in ("input_text", "output_text"):
|
||||
t = block.get("text", "").strip()
|
||||
if t:
|
||||
text_parts.append(t)
|
||||
return "\n".join(text_parts).strip()
|
||||
|
||||
|
||||
def _extract_function_output_text(output) -> str:
|
||||
"""Extract text from a function_call_output payload.
|
||||
|
||||
The output can be either a plain string or a list of content items.
|
||||
"""
|
||||
if isinstance(output, str):
|
||||
return output.strip()
|
||||
if isinstance(output, list):
|
||||
texts = []
|
||||
for item in output:
|
||||
if isinstance(item, dict) and item.get("type") in ("input_text", "text"):
|
||||
texts.append(item.get("text", ""))
|
||||
return "\n".join(texts).strip()
|
||||
return ""
|
||||
|
||||
|
||||
def _truncate(text: str) -> str:
|
||||
"""Truncate text to _MAX_TOOL_OUTPUT_CHARS."""
|
||||
if len(text) > _MAX_TOOL_OUTPUT_CHARS:
|
||||
return text[:_MAX_TOOL_OUTPUT_CHARS] + "... (truncated)"
|
||||
return text
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Recall: query composition and truncation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
@ -259,18 +534,20 @@ def prepare_retention_transcript(
|
|||
messages: list,
|
||||
retain_roles: list = None,
|
||||
retain_full_window: bool = False,
|
||||
include_tool_calls: bool = False,
|
||||
) -> tuple:
|
||||
"""Format messages into a retention transcript.
|
||||
|
||||
Outputs plain text with [role: ...]...[role:end] markers.
|
||||
Codex doesn't have tool calls to retain (it's a coding agent with
|
||||
shell/patch commands, not MCP tools), so we use the text format only.
|
||||
When include_tool_calls is True, outputs JSON with full message structure
|
||||
including tool calls and their inputs (matching Claude Code's format).
|
||||
Otherwise outputs the legacy text format with [role: ...]...[role:end] markers.
|
||||
|
||||
Args:
|
||||
messages: List of {role, content} dicts.
|
||||
messages: List of message dicts with 'role' and 'content'.
|
||||
retain_roles: Roles to include (default: ['user', 'assistant']).
|
||||
retain_full_window: If True, retain all messages. If False, retain
|
||||
only the last turn (last user msg + responses).
|
||||
include_tool_calls: If True, output JSON format with full tool call data.
|
||||
|
||||
Returns:
|
||||
(transcript_text, message_count) or (None, 0) if nothing to retain.
|
||||
|
|
@ -291,9 +568,42 @@ def prepare_retention_transcript(
|
|||
target_messages = messages[last_user_idx:]
|
||||
|
||||
allowed_roles = set(retain_roles or ["user", "assistant"])
|
||||
|
||||
if include_tool_calls:
|
||||
return _prepare_json_transcript(target_messages, allowed_roles)
|
||||
return _prepare_text_transcript(target_messages, allowed_roles)
|
||||
|
||||
|
||||
def _prepare_json_transcript(messages: list, allowed_roles: set) -> tuple:
|
||||
"""Format messages as JSON with full tool call data."""
|
||||
structured_messages = []
|
||||
for msg in messages:
|
||||
role = msg.get("role", "unknown")
|
||||
if role not in allowed_roles:
|
||||
continue
|
||||
|
||||
content = msg.get("content", "")
|
||||
blocks = _strip_memory_tags_from_blocks(content)
|
||||
if not blocks:
|
||||
continue
|
||||
|
||||
structured_messages.append({"role": role, "content": blocks})
|
||||
|
||||
if not structured_messages:
|
||||
return None, 0
|
||||
|
||||
transcript = json.dumps(structured_messages, indent=None, ensure_ascii=False)
|
||||
if len(transcript.strip()) < 10:
|
||||
return None, 0
|
||||
|
||||
return transcript, len(structured_messages)
|
||||
|
||||
|
||||
def _prepare_text_transcript(messages: list, allowed_roles: set) -> tuple:
|
||||
"""Format messages as legacy text with [role:]...[role:end] markers."""
|
||||
parts = []
|
||||
|
||||
for msg in target_messages:
|
||||
for msg in messages:
|
||||
role = msg.get("role", "unknown")
|
||||
if role not in allowed_roles:
|
||||
continue
|
||||
|
|
@ -316,3 +626,32 @@ def prepare_retention_transcript(
|
|||
return None, 0
|
||||
|
||||
return transcript, len(parts)
|
||||
|
||||
|
||||
def _strip_memory_tags_from_blocks(content) -> list:
|
||||
"""Strip memory tags from content, handling both string and list formats.
|
||||
|
||||
Returns a list of content blocks with memory tags removed.
|
||||
"""
|
||||
if isinstance(content, str):
|
||||
cleaned = strip_memory_tags(content).strip()
|
||||
return [{"type": "text", "text": cleaned}] if cleaned else []
|
||||
|
||||
if not isinstance(content, list):
|
||||
return []
|
||||
|
||||
blocks = []
|
||||
for block in content:
|
||||
if not isinstance(block, dict):
|
||||
continue
|
||||
block_type = block.get("type", "")
|
||||
|
||||
if block_type == "text":
|
||||
text = strip_memory_tags(block.get("text", "")).strip()
|
||||
if text:
|
||||
blocks.append({"type": "text", "text": text})
|
||||
elif block_type in ("tool_use", "tool_result"):
|
||||
# Pass through tool blocks as-is
|
||||
blocks.append(block)
|
||||
|
||||
return blocks
|
||||
|
|
|
|||
|
|
@ -56,7 +56,8 @@ def main():
|
|||
transcript_path = hook_input.get("transcript_path", "")
|
||||
|
||||
# Read full transcript
|
||||
all_messages = read_transcript(transcript_path)
|
||||
include_tool_calls = config.get("retainToolCalls", True)
|
||||
all_messages = read_transcript(transcript_path, include_tool_calls=include_tool_calls)
|
||||
if not all_messages:
|
||||
debug_log(config, "No messages in transcript, skipping retain")
|
||||
return
|
||||
|
|
@ -93,7 +94,7 @@ def main():
|
|||
# Format transcript
|
||||
retain_roles = config.get("retainRoles", ["user", "assistant"])
|
||||
transcript, message_count = prepare_retention_transcript(
|
||||
messages_to_retain, retain_roles, retain_full_window
|
||||
messages_to_retain, retain_roles, retain_full_window, include_tool_calls=include_tool_calls
|
||||
)
|
||||
|
||||
if not transcript:
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@
|
|||
"recallMaxQueryChars": 800,
|
||||
"recallRoles": ["user", "assistant"],
|
||||
"recallPromptPreamble": "Relevant memories from past conversations (prioritize recent when conflicting). Only use memories that are directly useful to continue this conversation; ignore the rest:",
|
||||
"retainToolCalls": true,
|
||||
"retainRoles": ["user", "assistant"],
|
||||
"retainEveryNTurns": 10,
|
||||
"retainOverlapTurns": 2,
|
||||
|
|
|
|||
|
|
@ -17,6 +17,12 @@ from lib.content import (
|
|||
)
|
||||
|
||||
|
||||
def _write_jsonl(tmp_path, entries):
|
||||
f = tmp_path / "transcript.jsonl"
|
||||
f.write_text("\n".join(json.dumps(e) for e in entries))
|
||||
return str(f)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# strip_memory_tags
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
@ -50,12 +56,6 @@ class TestStripMemoryTags:
|
|||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _write_jsonl(tmp_path, entries):
|
||||
f = tmp_path / "transcript.jsonl"
|
||||
f.write_text("\n".join(json.dumps(e) for e in entries))
|
||||
return str(f)
|
||||
|
||||
|
||||
class TestReadTranscriptFlat:
|
||||
def test_reads_flat_format(self, tmp_path):
|
||||
path = _write_jsonl(tmp_path, [
|
||||
|
|
@ -362,3 +362,358 @@ class TestPrepareRetentionTranscript:
|
|||
msgs = [{"role": "assistant", "content": "only assistant"}]
|
||||
result, _ = prepare_retention_transcript(msgs, retain_full_window=False)
|
||||
assert result is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# read_transcript — rich format (include_tool_calls=True)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestReadTranscriptRich:
|
||||
def test_reads_messages_as_structured_blocks(self, tmp_path):
|
||||
entries = [
|
||||
{
|
||||
"type": "response_item",
|
||||
"payload": {
|
||||
"type": "message",
|
||||
"role": "user",
|
||||
"content": [{"type": "input_text", "text": "hello"}],
|
||||
},
|
||||
},
|
||||
{
|
||||
"type": "response_item",
|
||||
"payload": {
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"content": [{"type": "output_text", "text": "hi there"}],
|
||||
"phase": "final_answer",
|
||||
},
|
||||
},
|
||||
]
|
||||
path = _write_jsonl(tmp_path, entries)
|
||||
msgs = read_transcript(path, include_tool_calls=True)
|
||||
assert len(msgs) == 2
|
||||
assert msgs[0]["role"] == "user"
|
||||
assert msgs[0]["content"] == [{"type": "text", "text": "hello"}]
|
||||
assert msgs[1]["role"] == "assistant"
|
||||
assert msgs[1]["content"] == [{"type": "text", "text": "hi there"}]
|
||||
|
||||
def test_reads_function_call(self, tmp_path):
|
||||
entries = [
|
||||
{
|
||||
"type": "response_item",
|
||||
"payload": {
|
||||
"type": "message",
|
||||
"role": "user",
|
||||
"content": [{"type": "input_text", "text": "list files"}],
|
||||
},
|
||||
},
|
||||
{
|
||||
"type": "response_item",
|
||||
"payload": {
|
||||
"type": "function_call",
|
||||
"name": "Read",
|
||||
"arguments": '{"file_path": "/tmp/foo.txt"}',
|
||||
"call_id": "call_1",
|
||||
},
|
||||
},
|
||||
{
|
||||
"type": "response_item",
|
||||
"payload": {
|
||||
"type": "function_call_output",
|
||||
"call_id": "call_1",
|
||||
"output": "file contents here",
|
||||
},
|
||||
},
|
||||
]
|
||||
path = _write_jsonl(tmp_path, entries)
|
||||
msgs = read_transcript(path, include_tool_calls=True)
|
||||
assert len(msgs) == 2 # user + assistant
|
||||
assistant = msgs[1]
|
||||
assert assistant["role"] == "assistant"
|
||||
assert len(assistant["content"]) == 2
|
||||
assert assistant["content"][0]["type"] == "tool_use"
|
||||
assert assistant["content"][0]["name"] == "Read"
|
||||
assert assistant["content"][0]["input"] == {"file_path": "/tmp/foo.txt"}
|
||||
assert assistant["content"][1]["type"] == "tool_result"
|
||||
assert "file contents here" in assistant["content"][1]["content"]
|
||||
|
||||
def test_reads_local_shell_call(self, tmp_path):
|
||||
entries = [
|
||||
{
|
||||
"type": "response_item",
|
||||
"payload": {
|
||||
"type": "message",
|
||||
"role": "user",
|
||||
"content": [{"type": "input_text", "text": "run ls"}],
|
||||
},
|
||||
},
|
||||
{
|
||||
"type": "response_item",
|
||||
"payload": {
|
||||
"type": "local_shell_call",
|
||||
"call_id": "call_1",
|
||||
"status": "completed",
|
||||
"action": {"type": "exec", "command": ["ls", "-la"]},
|
||||
},
|
||||
},
|
||||
]
|
||||
path = _write_jsonl(tmp_path, entries)
|
||||
msgs = read_transcript(path, include_tool_calls=True)
|
||||
assistant = msgs[1]
|
||||
assert assistant["content"][0]["type"] == "tool_use"
|
||||
assert assistant["content"][0]["name"] == "shell"
|
||||
assert assistant["content"][0]["input"]["command"] == ["ls", "-la"]
|
||||
|
||||
def test_reads_exec_command_end_event(self, tmp_path):
|
||||
entries = [
|
||||
{
|
||||
"type": "response_item",
|
||||
"payload": {
|
||||
"type": "message",
|
||||
"role": "user",
|
||||
"content": [{"type": "input_text", "text": "check git status"}],
|
||||
},
|
||||
},
|
||||
{
|
||||
"type": "event_msg",
|
||||
"payload": {
|
||||
"type": "exec_command_end",
|
||||
"call_id": "call_1",
|
||||
"command": ["git", "status"],
|
||||
"aggregated_output": "On branch main\nnothing to commit",
|
||||
"exit_code": 0,
|
||||
"status": "completed",
|
||||
},
|
||||
},
|
||||
]
|
||||
path = _write_jsonl(tmp_path, entries)
|
||||
msgs = read_transcript(path, include_tool_calls=True)
|
||||
assistant = msgs[1]
|
||||
assert assistant["content"][0]["type"] == "tool_use"
|
||||
assert assistant["content"][0]["name"] == "shell"
|
||||
assert assistant["content"][0]["input"]["command"] == ["git", "status"]
|
||||
assert assistant["content"][1]["type"] == "tool_result"
|
||||
assert "On branch main" in assistant["content"][1]["content"]
|
||||
|
||||
def test_reads_exec_command_end_with_nonzero_exit(self, tmp_path):
|
||||
entries = [
|
||||
{
|
||||
"type": "response_item",
|
||||
"payload": {
|
||||
"type": "message",
|
||||
"role": "user",
|
||||
"content": [{"type": "input_text", "text": "run bad cmd"}],
|
||||
},
|
||||
},
|
||||
{
|
||||
"type": "event_msg",
|
||||
"payload": {
|
||||
"type": "exec_command_end",
|
||||
"call_id": "call_1",
|
||||
"command": ["false"],
|
||||
"aggregated_output": "",
|
||||
"exit_code": 1,
|
||||
"status": "failed",
|
||||
},
|
||||
},
|
||||
]
|
||||
path = _write_jsonl(tmp_path, entries)
|
||||
msgs = read_transcript(path, include_tool_calls=True)
|
||||
assistant = msgs[1]
|
||||
result = assistant["content"][1]
|
||||
assert "exit_code: 1" in result["content"]
|
||||
assert "status: failed" in result["content"]
|
||||
|
||||
def test_reads_patch_apply_end_event(self, tmp_path):
|
||||
entries = [
|
||||
{
|
||||
"type": "response_item",
|
||||
"payload": {
|
||||
"type": "message",
|
||||
"role": "user",
|
||||
"content": [{"type": "input_text", "text": "fix the bug"}],
|
||||
},
|
||||
},
|
||||
{
|
||||
"type": "event_msg",
|
||||
"payload": {
|
||||
"type": "patch_apply_end",
|
||||
"call_id": "call_1",
|
||||
"changes": [{"file": "main.py", "diff": "+fixed line"}],
|
||||
"status": "applied",
|
||||
},
|
||||
},
|
||||
]
|
||||
path = _write_jsonl(tmp_path, entries)
|
||||
msgs = read_transcript(path, include_tool_calls=True)
|
||||
assistant = msgs[1]
|
||||
assert assistant["content"][0]["type"] == "tool_use"
|
||||
assert assistant["content"][0]["name"] == "patch"
|
||||
assert assistant["content"][1]["content"] == "status: applied"
|
||||
|
||||
def test_reads_web_search_call(self, tmp_path):
|
||||
entries = [
|
||||
{
|
||||
"type": "response_item",
|
||||
"payload": {
|
||||
"type": "message",
|
||||
"role": "user",
|
||||
"content": [{"type": "input_text", "text": "search for python docs"}],
|
||||
},
|
||||
},
|
||||
{
|
||||
"type": "response_item",
|
||||
"payload": {
|
||||
"type": "web_search_call",
|
||||
"status": "completed",
|
||||
"action": {"type": "search", "query": "python documentation"},
|
||||
},
|
||||
},
|
||||
]
|
||||
path = _write_jsonl(tmp_path, entries)
|
||||
msgs = read_transcript(path, include_tool_calls=True)
|
||||
assistant = msgs[1]
|
||||
assert assistant["content"][0]["type"] == "tool_use"
|
||||
assert assistant["content"][0]["name"] == "web_search"
|
||||
assert assistant["content"][0]["input"]["query"] == "python documentation"
|
||||
|
||||
def test_skips_non_final_answer_in_rich_mode(self, tmp_path):
|
||||
entries = [
|
||||
{
|
||||
"type": "response_item",
|
||||
"payload": {
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"content": [{"type": "output_text", "text": "thinking..."}],
|
||||
"phase": "reasoning",
|
||||
},
|
||||
},
|
||||
{
|
||||
"type": "response_item",
|
||||
"payload": {
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"content": [{"type": "output_text", "text": "The answer"}],
|
||||
"phase": "final_answer",
|
||||
},
|
||||
},
|
||||
]
|
||||
path = _write_jsonl(tmp_path, entries)
|
||||
msgs = read_transcript(path, include_tool_calls=True)
|
||||
assert len(msgs) == 1
|
||||
assert msgs[0]["content"][0]["text"] == "The answer"
|
||||
|
||||
def test_flat_format_works_in_rich_mode(self, tmp_path):
|
||||
path = _write_jsonl(tmp_path, [
|
||||
{"role": "user", "content": "hello"},
|
||||
{"role": "assistant", "content": "hi"},
|
||||
])
|
||||
msgs = read_transcript(path, include_tool_calls=True)
|
||||
assert len(msgs) == 2
|
||||
assert msgs[0]["content"] == [{"type": "text", "text": "hello"}]
|
||||
assert msgs[1]["content"] == [{"type": "text", "text": "hi"}]
|
||||
|
||||
def test_truncates_long_tool_output(self, tmp_path):
|
||||
long_output = "x" * 3000
|
||||
entries = [
|
||||
{
|
||||
"type": "response_item",
|
||||
"payload": {
|
||||
"type": "message",
|
||||
"role": "user",
|
||||
"content": [{"type": "input_text", "text": "do something"}],
|
||||
},
|
||||
},
|
||||
{
|
||||
"type": "response_item",
|
||||
"payload": {
|
||||
"type": "function_call",
|
||||
"name": "Read",
|
||||
"arguments": "{}",
|
||||
"call_id": "call_1",
|
||||
},
|
||||
},
|
||||
{
|
||||
"type": "response_item",
|
||||
"payload": {
|
||||
"type": "function_call_output",
|
||||
"call_id": "call_1",
|
||||
"output": long_output,
|
||||
},
|
||||
},
|
||||
]
|
||||
path = _write_jsonl(tmp_path, entries)
|
||||
msgs = read_transcript(path, include_tool_calls=True)
|
||||
result = msgs[1]["content"][1]
|
||||
assert len(result["content"]) < 3000
|
||||
assert "truncated" in result["content"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# prepare_retention_transcript — JSON format (include_tool_calls=True)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestPrepareRetentionTranscriptJson:
|
||||
def test_json_format_basic(self):
|
||||
msgs = [
|
||||
{"role": "user", "content": [{"type": "text", "text": "hello"}]},
|
||||
{"role": "assistant", "content": [{"type": "text", "text": "hi"}]},
|
||||
]
|
||||
transcript, count = prepare_retention_transcript(
|
||||
msgs, retain_full_window=True, include_tool_calls=True
|
||||
)
|
||||
assert count == 2
|
||||
parsed = json.loads(transcript)
|
||||
assert len(parsed) == 2
|
||||
assert parsed[0]["role"] == "user"
|
||||
assert parsed[1]["role"] == "assistant"
|
||||
|
||||
def test_json_format_with_tool_calls(self):
|
||||
msgs = [
|
||||
{"role": "user", "content": [{"type": "text", "text": "list files"}]},
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": [
|
||||
{"type": "tool_use", "name": "shell", "input": {"command": ["ls"]}},
|
||||
{"type": "tool_result", "content": "file1.txt"},
|
||||
{"type": "text", "text": "Here are the files."},
|
||||
],
|
||||
},
|
||||
]
|
||||
transcript, count = prepare_retention_transcript(
|
||||
msgs, retain_full_window=True, include_tool_calls=True
|
||||
)
|
||||
parsed = json.loads(transcript)
|
||||
assistant = parsed[1]
|
||||
assert any(b["type"] == "tool_use" for b in assistant["content"])
|
||||
assert any(b["type"] == "tool_result" for b in assistant["content"])
|
||||
|
||||
def test_json_format_strips_memory_tags(self):
|
||||
msgs = [
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "text", "text": "<hindsight_memories>secret</hindsight_memories> real question"},
|
||||
],
|
||||
},
|
||||
]
|
||||
transcript, _ = prepare_retention_transcript(
|
||||
msgs, retain_full_window=True, include_tool_calls=True
|
||||
)
|
||||
assert "hindsight_memories" not in transcript
|
||||
assert "secret" not in transcript
|
||||
assert "real question" in transcript
|
||||
|
||||
def test_json_format_filters_roles(self):
|
||||
msgs = [
|
||||
{"role": "user", "content": [{"type": "text", "text": "user msg"}]},
|
||||
{"role": "assistant", "content": [{"type": "text", "text": "assistant msg"}]},
|
||||
]
|
||||
transcript, count = prepare_retention_transcript(
|
||||
msgs, retain_roles=["user"], retain_full_window=True, include_tool_calls=True
|
||||
)
|
||||
parsed = json.loads(transcript)
|
||||
assert count == 1
|
||||
assert all(m["role"] == "user" for m in parsed)
|
||||
|
|
|
|||
Loading…
Reference in a new issue