generated from coulomb/repo-seed
session-memory Phase 0: normalized schema (T01) + Claude adapter (T02)
- session_memory/core/schema.py: Session/SessionEvent/Cost dataclasses, flavor-prefixed uids, watermarks, kind/outcome validation (T01) - session_memory/adapters/claude.py: JSONL -> Normalized bundle, turn DAG via uuid/parentUuid, kind mapping, cost from message.usage (T02) - tests: schema round-trip + adapter (synthetic + real local session) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
99
tests/test_claude_adapter.py
Normal file
99
tests/test_claude_adapter.py
Normal file
@@ -0,0 +1,99 @@
|
||||
"""Claude adapter tests (T02): synthetic fixture + a real on-disk session."""
|
||||
|
||||
import glob
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from session_memory.adapters.claude import parse_session # noqa: E402
|
||||
|
||||
REPO_MAP = {"agentic-resources": "helix_forge"}
|
||||
|
||||
|
||||
def _write_jsonl(path, records):
|
||||
with open(path, "w", encoding="utf-8") as f:
|
||||
for r in records:
|
||||
f.write(json.dumps(r) + "\n")
|
||||
|
||||
|
||||
def test_synthetic_session(tmp_path):
|
||||
p = tmp_path / "11111111-2222-3333-4444-555555555555.jsonl"
|
||||
_write_jsonl(p, [
|
||||
{"type": "user", "uuid": "u1", "parentUuid": None,
|
||||
"timestamp": "2026-06-06T10:00:00Z", "sessionId": "sess-1",
|
||||
"cwd": "/home/worsch/agentic-resources", "gitBranch": "main",
|
||||
"version": "1.0", "message": {"role": "user", "content": "fix the bug"}},
|
||||
{"type": "assistant", "uuid": "a1", "parentUuid": "u1",
|
||||
"timestamp": "2026-06-06T10:00:05Z", "sessionId": "sess-1",
|
||||
"message": {"role": "assistant", "model": "claude-opus-4-8",
|
||||
"usage": {"input_tokens": 100, "output_tokens": 20,
|
||||
"cache_read_input_tokens": 10},
|
||||
"content": [
|
||||
{"type": "thinking", "thinking": "let me look"},
|
||||
{"type": "text", "text": "I'll edit the file."},
|
||||
{"type": "tool_use", "name": "Edit",
|
||||
"input": {"file_path": "x.py", "old_string": "a", "new_string": "b"}},
|
||||
{"type": "tool_use", "name": "Bash",
|
||||
"input": {"command": "pytest -q"}},
|
||||
]}},
|
||||
{"type": "user", "uuid": "u2", "parentUuid": "a1",
|
||||
"timestamp": "2026-06-06T10:00:10Z", "sessionId": "sess-1",
|
||||
"message": {"role": "user",
|
||||
"content": [{"type": "tool_result", "content": "6 passed"}]}},
|
||||
])
|
||||
|
||||
norm = parse_session(str(p), REPO_MAP)
|
||||
assert norm is not None
|
||||
s = norm.session
|
||||
assert s.session_uid == "claude:sess-1"
|
||||
assert s.repo == "agentic-resources" and s.domain == "helix_forge"
|
||||
assert s.model == "claude-opus-4-8"
|
||||
assert s.cost.input_tokens == 100 and s.cost.output_tokens == 20
|
||||
assert s.cost.cache_tokens == 10
|
||||
assert s.cost.turns == 1
|
||||
assert s.cost.wall_clock_s == 10.0
|
||||
|
||||
kinds = [e.kind for e in norm.events]
|
||||
assert kinds == ["user_msg", "thinking", "assistant_msg", "edit", "test_run", "tool_result"]
|
||||
|
||||
# turn DAG: assistant events link back to the first user msg (seq 0)
|
||||
edit_ev = next(e for e in norm.events if e.kind == "edit")
|
||||
assert edit_ev.parent_seq == 0
|
||||
assert edit_ev.tool == "Edit"
|
||||
|
||||
# bodies captured as blobs, referenced by payload_ref
|
||||
assert edit_ev.payload_ref in norm.blobs
|
||||
assert "x.py" in norm.blobs[edit_ev.payload_ref]
|
||||
|
||||
|
||||
def test_sidechain_filename_marks_events(tmp_path):
|
||||
p = tmp_path / "agent-deadbeef.jsonl"
|
||||
_write_jsonl(p, [
|
||||
{"type": "assistant", "uuid": "a1", "sessionId": "side-1",
|
||||
"timestamp": "2026-06-06T10:00:00Z",
|
||||
"message": {"role": "assistant", "content": [{"type": "text", "text": "hi"}]}},
|
||||
])
|
||||
norm = parse_session(str(p), REPO_MAP)
|
||||
assert norm.events[0].is_sidechain is True
|
||||
|
||||
|
||||
def test_real_local_session_if_available():
|
||||
"""Smoke-parse a real Claude transcript on this workstation, if present."""
|
||||
base = os.path.expanduser("~/.claude/projects/-home-worsch-agentic-resources")
|
||||
files = sorted(glob.glob(os.path.join(base, "*.jsonl")))
|
||||
if not files:
|
||||
return # environment without local sessions; synthetic tests cover logic
|
||||
parsed = 0
|
||||
for fp in files:
|
||||
norm = parse_session(fp, REPO_MAP)
|
||||
if norm is None:
|
||||
continue
|
||||
parsed += 1
|
||||
assert norm.session.session_uid.startswith("claude:")
|
||||
# seq is monotonic and unique
|
||||
seqs = [e.seq for e in norm.events]
|
||||
assert seqs == sorted(seqs)
|
||||
assert len(seqs) == len(set(seqs))
|
||||
assert parsed >= 1
|
||||
Reference in New Issue
Block a user