Skip to content

Show queued prompts in transcripts #112

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
roli-lpci wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Show queued prompts in transcripts #112

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 37 additions & 4 deletions src/claude_code_transcripts/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -131,20 +131,30 @@ def _get_jsonl_summary(filepath, max_length=200):
except json.JSONDecodeError:
continue

# Second pass: find first non-meta user message
# Second pass: find first non-meta user or queued prompt
with open(filepath, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
obj = json.loads(line)
if (
is_user_prompt = (
obj.get("type") == "user"
and not obj.get("isMeta")
and obj.get("message", {}).get("content")
):
content = obj["message"]["content"]
)
is_queued_prompt = (
obj.get("type") == "queue-operation"
and obj.get("operation") == "enqueue"
and obj.get("content")
)
if is_user_prompt or is_queued_prompt:
content = (
obj["message"]["content"]
if is_user_prompt
else obj["content"]
)
text = extract_text_from_content(content)
if text and not text.startswith("<"):
if len(text) > max_length:
Expand Down Expand Up @@ -477,6 +487,27 @@ def _parse_jsonl_file(filepath):
obj = json.loads(line)
entry_type = obj.get("type")

if (
entry_type == "queue-operation"
and obj.get("operation") == "enqueue"
):
content = obj.get("content")
queued_text = extract_text_from_content(content)
if not queued_text:
continue
loglines.append(
{
"type": "user",
"timestamp": obj.get("timestamp", ""),
"message": {
"role": "user",
"content": content,
"isQueuedPrompt": True,
},
}
)
continue

# Skip non-message entries
if entry_type not in ("user", "assistant"):
continue
Expand Down Expand Up @@ -958,6 +989,8 @@ def render_message(log_type, message_json, timestamp):
# Check if this is a tool result message
if is_tool_result_message(message_data):
role_class, role_label = "tool-reply", "Tool reply"
elif message_data.get("isQueuedPrompt"):
role_class, role_label = "user queued", "Queued prompt"
else:
role_class, role_label = "user", "User"
elif log_type == "assistant":
Expand Down
111 changes: 108 additions & 3 deletions tests/test_generate_html.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -213,7 +213,6 @@ def test_image_block(self, snapshot_html):
# 200x200 black GIF - minimal valid GIF with black pixels
# Generated with: from PIL import Image; img = Image.new('RGB', (200, 200), (0, 0, 0)); img.save('black.gif')
import base64
import io

# Create a minimal 200x200 black GIF using raw bytes
# GIF89a header + logical screen descriptor + global color table + image data
Expand Down Expand Up @@ -574,7 +573,6 @@ class TestCreateGist:
def test_creates_gist_successfully(self, output_dir, monkeypatch):
"""Test successful gist creation."""
import subprocess
import click

# Create test HTML files
(output_dir / "index.html").write_text(
Expand Down Expand Up @@ -1133,6 +1131,114 @@ def test_jsonl_preserves_message_content(self):
user_msg = next(e for e in result["loglines"] if e["type"] == "user")
assert user_msg["message"]["content"] == "Create a hello world function"

def test_jsonl_preserves_queued_prompt_and_later_user_delivery(
self, tmp_path, output_dir
):
jsonl_file = tmp_path / "queued.jsonl"
entries = [
{
"type": "user",
"timestamp": "2026-04-23T04:59:00.000Z",
"message": {"role": "user", "content": "Start the work"},
},
{
"type": "assistant",
"timestamp": "2026-04-23T04:59:10.000Z",
"message": {
"role": "assistant",
"content": [{"type": "text", "text": "Working on it"}],
},
},
{
"type": "queue-operation",
"operation": "enqueue",
"timestamp": "2026-04-23T04:59:40.480Z",
"content": "Use red/green TDD",
},
{
"type": "assistant",
"timestamp": "2026-04-23T04:59:50.000Z",
"message": {
"role": "assistant",
"content": [{"type": "text", "text": "Finishing the first task"}],
},
},
{
"type": "queue-operation",
"operation": "dequeue",
"timestamp": "2026-04-23T05:00:00.000Z",
},
{
"type": "user",
"timestamp": "2026-04-23T05:00:01.000Z",
"message": {"role": "user", "content": "Use red/green TDD"},
},
{
"type": "assistant",
"timestamp": "2026-04-23T05:00:10.000Z",
"message": {
"role": "assistant",
"content": [{"type": "text", "text": "Using TDD"}],
},
},
]
jsonl_file.write_text("".join(json.dumps(entry) + "\n" for entry in entries))

result = parse_session_file(jsonl_file)
queued = [
entry
for entry in result["loglines"]
if entry.get("message", {}).get("isQueuedPrompt")
]
assert len(queued) == 1
assert queued[0]["timestamp"] == "2026-04-23T04:59:40.480Z"
delivered = [
entry
for entry in result["loglines"]
if entry.get("message", {}).get("content") == "Use red/green TDD"
and not entry["message"].get("isQueuedPrompt")
]
assert len(delivered) == 1
assert delivered[0]["timestamp"] == "2026-04-23T05:00:01.000Z"

generate_html(jsonl_file, output_dir)
page_html = (output_dir / "page-001.html").read_text(encoding="utf-8")
assert page_html.count("Use red/green TDD") == 2
assert "Queued prompt" in page_html
assert "2026-04-23T04:59:40.480Z" in page_html
assert "2026-04-23T05:00:01.000Z" in page_html
assert page_html.index("Working on it") < page_html.index("Use red/green TDD")
assert page_html.index("Use red/green TDD") < page_html.index(
"Finishing the first task"
)

def test_jsonl_preserves_queued_prompt_with_array_content(
self, tmp_path, output_dir
):
jsonl_file = tmp_path / "queued-array.jsonl"
jsonl_file.write_text(
json.dumps(
{
"type": "queue-operation",
"operation": "enqueue",
"timestamp": "2026-04-23T05:03:18.122Z",
"content": [{"type": "text", "text": "Use the PDF.js renderer"}],
}
)
+ "\n"
)

assert get_session_summary(jsonl_file) == "Use the PDF.js renderer"
result = parse_session_file(jsonl_file)
assert result["loglines"][0]["message"]["content"] == [
{"type": "text", "text": "Use the PDF.js renderer"}
]

generate_html(jsonl_file, output_dir)
page_html = (output_dir / "page-001.html").read_text(encoding="utf-8")
assert "Queued prompt" in page_html
assert "Use the PDF.js renderer" in page_html

def test_jsonl_generates_html(self, output_dir, snapshot_html):
"""Test that JSONL files can be converted to HTML."""
fixture_path = Path(__file__).parent / "sample_session.jsonl"
Expand Down Expand Up @@ -1416,7 +1522,6 @@ def test_json_output_auto_uses_cwd_when_no_output(self, tmp_path, monkeypatch):
"""Test that json -a uses current directory when -o not specified."""
from click.testing import CliRunner
from claude_code_transcripts import cli
import os

fixture_path = Path(__file__).parent / "sample_session.json"

Expand Down