import json
import os
import sqlite3
import subprocess
import sys
import tempfile
def main():
goosedump_bin = os.environ.get("GOOSEDUMP_BIN", "")
if not goosedump_bin:
print("FAIL GOOSEDUMP_BIN not set")
sys.exit(1)
pass_count = 0
fail_count = 0
def passed(name):
nonlocal pass_count
pass_count += 1
print(f"PASS {name}")
def failed(name, reason):
nonlocal fail_count
fail_count += 1
print(f"FAIL {name} -- {reason}")
def run(args, env=None, cwd=None):
result = subprocess.run(
[goosedump_bin] + args,
capture_output=True,
text=True,
env=env,
cwd=cwd,
)
return result.returncode, result.stdout, result.stderr
def expect_success(name, args, env, cwd, contains):
status, stdout, stderr = run(args, env=env, cwd=cwd)
if status != 0:
failed(name, f"expected status 0, got {status} stderr={stderr[:200]}")
return
if contains is not None and contains not in stdout:
failed(name, f"expected output to contain '{contains}'. stdout: {stdout[:400]}")
return
passed(name)
def expect_not_contains(name, args, env, cwd, not_contains):
status, stdout, stderr = run(args, env=env, cwd=cwd)
if status != 0:
failed(name, f"expected status 0, got {status} stderr={stderr[:200]}")
return
if not_contains in stdout:
failed(name, f"output must not contain '{not_contains}'. stdout: {stdout[:400]}")
return
passed(name)
def expect_failure(name, args, env, cwd):
status, stdout, _stderr = run(args, env=env, cwd=cwd)
if status != 0:
passed(name)
else:
failed(name, f"expected failure but got success. stdout: {stdout[:200]}")
def expect_stdout(name, args, env, cwd, *patterns):
status, stdout, stderr = run(args, env=env, cwd=cwd)
if status != 0:
failed(name, f"expected status 0, got {status} stderr={stderr[:200]}")
return
for p in patterns:
if p not in stdout:
failed(name, f"output missing '{p}'. stdout: {stdout[:400]}")
return
passed(name)
def expect_failure_stdout(name, args, env, cwd, expected_status, *patterns):
status, stdout, _stderr = run(args, env=env, cwd=cwd)
if status != expected_status:
failed(name, f"expected status {expected_status}, got {status}")
return
for p in patterns:
if p not in stdout:
failed(name, f"output missing '{p}'. stdout: {stdout[:400]}")
return
passed(name)
def write_jsonl(file, value):
file.write(json.dumps(value) + "\n")
with tempfile.TemporaryDirectory() as tmpdir:
oc_dir = os.path.join(tmpdir, "opencode")
os.makedirs(oc_dir)
oc_db = os.path.join(oc_dir, "opencode.db")
conn = sqlite3.connect(oc_db)
conn.execute(
"CREATE TABLE message(session_id TEXT, id TEXT, data TEXT, time_created INTEGER)"
)
conn.execute(
"CREATE TABLE part(session_id TEXT, message_id TEXT, data TEXT, time_created INTEGER)"
)
sid1 = "sesh-alpha"
conn.execute(
"INSERT INTO message VALUES(?, ?, ?, ?)",
(
sid1,
"msg-a",
json.dumps({"role": "user", "id": "entry-alpha", "parentID": ""}),
1000,
),
)
conn.execute(
"INSERT INTO part VALUES(?, ?, ?, ?)",
(
sid1,
"msg-a",
json.dumps({"type": "text", "text": "search keyword STARTREK-DISCO"}),
1000,
),
)
conn.execute(
"INSERT INTO message VALUES(?, ?, ?, ?)",
(
sid1,
"msg-b",
json.dumps(
{"role": "assistant", "id": "entry-bravo", "parentID": "entry-alpha"}
),
2000,
),
)
conn.execute(
"INSERT INTO part VALUES(?, ?, ?, ?)",
(
sid1,
"msg-b",
json.dumps({"type": "reasoning", "text": "need to read a file"}),
2000,
),
)
conn.execute(
"INSERT INTO part VALUES(?, ?, ?, ?)",
(
sid1,
"msg-b",
json.dumps({"type": "text", "text": "found STARTREK-DISCO in source"}),
2001,
),
)
conn.execute(
"INSERT INTO part VALUES(?, ?, ?, ?)",
(
sid1,
"msg-b",
json.dumps(
{
"type": "tool",
"name": "read",
"arguments": {
"filePath": "src/main.rs",
"description": "read tool",
},
}
),
2002,
),
)
sid2 = "sesh-bravo"
conn.execute(
"INSERT INTO message VALUES(?, ?, ?, ?)",
(
sid2,
"msg-c",
json.dumps({"role": "user", "id": "entry-charlie", "parentID": ""}),
3000,
),
)
conn.execute(
"INSERT INTO part VALUES(?, ?, ?, ?)",
(
sid2,
"msg-c",
json.dumps({"type": "text", "text": "query about TANGO-SIERRA"}),
3000,
),
)
conn.commit()
conn.close()
crush_db = os.path.join(tmpdir, "crush.db")
conn = sqlite3.connect(crush_db)
conn.execute(
"CREATE TABLE messages(session_id TEXT, role TEXT, parts TEXT, created_at TEXT)"
)
csid = "sesh-crush"
conn.execute(
"INSERT INTO messages VALUES(?, ?, ?, ?)",
(
csid,
"user",
json.dumps([{"type": "text", "text": "explain ZULU-NOVEMBER"}]),
"2026-01-01T00:00:00Z",
),
)
conn.execute(
"INSERT INTO messages VALUES(?, ?, ?, ?)",
(
csid,
"assistant",
json.dumps(
[
{
"type": "text",
"text": "ZULU-NOVEMBER is a code pattern",
},
{
"type": "tool_use",
"name": "grep",
"input": {"pattern": "ZULU", "filePath": "lib.rs"},
},
]
),
"2026-01-01T00:00:01Z",
),
)
conn.commit()
conn.close()
with open(os.path.join(tmpdir, ".crush.json"), "w") as f:
json.dump({"data_directory": "."}, f)
pi_dir = os.path.join(tmpdir, "pi-sessions")
os.makedirs(pi_dir)
pi_file = os.path.join(pi_dir, "pi-sesh.jsonl")
with open(pi_file, "w") as f:
f.write(
json.dumps(
{
"type": "session",
"id": "sesh-pi",
"cwd": "/tmp/project",
}
)
+ "\n"
)
f.write(
json.dumps(
{
"type": "message",
"id": "pi-entry-1",
"parentId": "",
"message": {
"role": "user",
"content": [
{"type": "text", "text": "find TRIPLA-ESPRESSO"}
],
},
}
)
+ "\n"
)
f.write(
json.dumps(
{
"type": "message",
"id": "pi-entry-2",
"parentId": "pi-entry-1",
"message": {
"role": "assistant",
"content": [
{
"type": "thinking",
"text": "searching for TRIPLA-ESPRESSO",
},
{
"type": "text",
"text": "found TRIPLA-ESPRESSO in file",
},
{
"type": "toolCall",
"name": "bash",
"arguments": {
"command": "grep TRIPLA src/*.rs"
},
},
],
},
}
)
+ "\n"
)
f.write(
json.dumps(
{
"type": "message",
"id": "pi-entry-3",
"parentId": "pi-entry-2",
"message": {
"role": "bashExecution",
"command": "grep TRIPLA src/*.rs",
"content": [
{
"type": "text",
"text": "src/main.rs:123: TRIPLA-ESPRESSO\ntest result: ok. 1 passed",
}
],
},
}
)
+ "\n"
)
f.write(
json.dumps(
{
"type": "message",
"id": "pi-entry-4",
"parentId": "pi-entry-3",
"message": {
"role": "toolResult",
"toolName": "todo",
"content": "Created #99: Review compact parity (pending)\nUpdated #99 (pending -> in_progress)\n1:928|// SPDX-License-Identifier\n.git/\nnode_modules/",
},
}
)
+ "\n"
)
codex_home = os.path.join(tmpdir, "codex-home")
codex_sessions_dir = os.path.join(codex_home, "sessions", "2026", "06", "11")
os.makedirs(codex_sessions_dir)
codex_id = "019eb466-8e72-7dd0-8a10-9c8963f7d86c"
codex_file = os.path.join(
codex_sessions_dir,
"rollout-2026-06-11T04-58-00-019eb466-8e72-7dd0-8a10-9c8963f7d86c.jsonl",
)
codex_entries = [
{
"timestamp": "2026-06-11T01:58:35.740Z",
"type": "session_meta",
"payload": {
"id": codex_id,
"timestamp": "2026-06-11T01:58:00.570Z",
"cwd": "/tmp/project",
"originator": "codex-tui",
},
},
{
"timestamp": "2026-06-11T01:58:35.777Z",
"type": "response_item",
"payload": {
"type": "message",
"id": "codex-user-1",
"role": "user",
"content": [{"type": "input_text", "text": "find DEADBEEF"}],
},
},
{
"timestamp": "2026-06-11T01:58:36.000Z",
"type": "response_item",
"payload": {
"type": "message",
"id": "codex-assistant-1",
"role": "assistant",
"content": [
{"type": "output_text", "text": "I found DEADBEEF in source"}
],
},
},
{
"timestamp": "2026-06-11T01:58:36.250Z",
"type": "response_item",
"payload": {
"type": "function_call",
"call_id": "call-codex-grep",
"name": "grep",
"arguments": json.dumps({"pattern": "DEADBEEF", "path": "src/main.rs"}),
},
},
{
"timestamp": "2026-06-11T01:58:36.300Z",
"type": "response_item",
"payload": {
"type": "reasoning",
"id": "codex-reasoning-1",
"summary": [
{
"type": "summary_text",
"text": "consider DEADBEEF search plan",
}
],
},
},
{
"timestamp": "2026-06-11T01:58:36.350Z",
"type": "response_item",
"payload": {
"type": "custom_tool_call",
"call_id": "call-codex-custom",
"name": "lookup",
"input": {"query": "DEADBEEF custom"},
},
},
{
"timestamp": "2026-06-11T01:58:36.400Z",
"type": "response_item",
"payload": {
"type": "local_shell_call",
"call_id": "call-codex-shell",
"action": {"command": "grep DEADBEEF src/main.rs"},
},
},
{
"timestamp": "2026-06-11T01:58:36.500Z",
"type": "response_item",
"payload": {
"type": "function_call_output",
"call_id": "call-codex-grep",
"output": "src/main.rs:7: DEADBEEF",
},
},
]
with open(codex_file, "w") as f:
for entry in codex_entries:
write_jsonl(f, entry)
goose_sessions_dir = os.path.join(tmpdir, "goose", "sessions")
os.makedirs(goose_sessions_dir)
goose_db = os.path.join(goose_sessions_dir, "sessions.db")
conn = sqlite3.connect(goose_db)
conn.execute("CREATE TABLE sessions(id TEXT PRIMARY KEY, name TEXT, working_dir TEXT, updated_at TIMESTAMP)")
conn.execute(
"CREATE TABLE messages(id INTEGER PRIMARY KEY AUTOINCREMENT, message_id TEXT, session_id TEXT, role TEXT, content_json TEXT, created_timestamp INTEGER)"
)
gsid = "sesh-goose"
conn.execute(
"INSERT INTO sessions VALUES(?, ?, ?, ?)",
(gsid, "Review Session", "/tmp/project", "2026-06-10T00:00:00Z"),
)
conn.execute(
"INSERT INTO messages VALUES(?, ?, ?, ?, ?, ?)",
(
1,
"msg-1",
gsid,
"user",
json.dumps([{"type": "text", "text": "find YANKEE-ECHO"}]),
1000,
),
)
conn.execute(
"INSERT INTO messages VALUES(?, ?, ?, ?, ?, ?)",
(
2,
"msg-2",
gsid,
"assistant",
json.dumps([
{"type": "thinking", "thinking": "searching for YANKEE-ECHO", "signature": ""},
{"type": "text", "text": "let me search for that"},
]),
2000,
),
)
conn.execute(
"INSERT INTO messages VALUES(?, ?, ?, ?, ?, ?)",
(
3,
"msg-3",
gsid,
"assistant",
json.dumps([
{"type": "toolRequest", "id": "call_00", "toolCall": {"status": "success", "value": {"name": "grep", "arguments": {"pattern": "YANKEE"}}}},
]),
3000,
),
)
conn.execute(
"INSERT INTO messages VALUES(?, ?, ?, ?, ?, ?)",
(
4,
"msg-4",
gsid,
"user",
json.dumps([
{"type": "toolResponse", "id": "call_00", "toolResult": {"status": "success", "value": {"name": "grep", "content": [{"type": "text", "text": "src/main.rs:42: YANKEE-ECHO"}]}}},
]),
4000,
),
)
conn.commit()
conn.close()
base_env = os.environ.copy()
base_env["GOOSEDUMP_DATA_DIR"] = tmpdir
base_env["PI_CODING_AGENT_SESSION_DIR"] = pi_dir
base_env["CODEX_HOME"] = codex_home
expect_stdout(
"opencode: list contexts",
["list", "opencode"],
base_env,
None,
"Sessions:",
"sesh-alpha",
"sesh-bravo",
)
expect_stdout(
"opencode: display messages",
["show", "opencode", "sesh-alpha"],
base_env,
None,
"entryId: entry-alpha (user)",
"STARTREK-DISCO",
"entryId: entry-bravo (assistant)",
)
expect_stdout(
"opencode: clipped display",
["show", "opencode", "sesh-alpha", "--clip"],
base_env,
None,
"entryId: entry-alpha (user)",
"entryId: entry-bravo (assistant)",
"tools: read(path=src/main.rs)",
)
expect_success(
"opencode: grep literal match",
["grep", "opencode", "sesh-alpha", "STARTREK-DISCO"],
base_env,
None,
"STARTREK-DISCO",
)
expect_success(
"opencode: grep fuzzy match",
["grep", "opencode", "sesh-alpha", "STRTRK-DSC"],
base_env,
None,
"STARTREK-DISCO",
)
expect_success(
"opencode: grep no match",
["grep", "opencode", "sesh-alpha", "NONEXISTENT-ZZ"],
base_env,
None,
"no results",
)
expect_success(
"opencode: ranked search",
["search", "opencode", "sesh-alpha", "STARTREK"],
base_env,
None,
"results for",
)
expect_stdout(
"opencode: ID filter",
["show", "opencode", "sesh-alpha", "--ids", "entry-alpha"],
base_env,
None,
"entryId: entry-alpha (user)",
)
expect_not_contains(
"opencode: ID filter excludes other",
["show", "opencode", "sesh-alpha", "--ids", "entry-alpha"],
base_env,
None,
"entry-bravo",
)
expect_stdout(
"opencode: lineage scope",
["show", "opencode", "sesh-alpha", "--scope", "lineage"],
base_env,
None,
"entryId: entry-alpha (user)",
"entryId: entry-bravo (assistant)",
)
expect_stdout(
"crush: list contexts",
["list", "crush"],
base_env,
tmpdir,
"Sessions:",
"sesh-crush",
)
expect_stdout(
"crush: display messages",
["show", "crush", "sesh-crush"],
base_env,
tmpdir,
"entryId: crush-0 (user)",
"ZULU-NOVEMBER",
"entryId: crush-1 (assistant)",
"code pattern",
)
expect_stdout(
"pi: list contexts",
["list", "pi"],
base_env,
None,
"Sessions:",
"sesh-pi",
)
expect_stdout(
"pi: display messages",
["show", "pi", "sesh-pi"],
base_env,
None,
"entryId: pi-entry-1 (user)",
"TRIPLA-ESPRESSO",
"entryId: pi-entry-2 (assistant)",
"entryId: pi-entry-3 (bash)",
"grep TRIPLA src/*.rs",
)
expect_stdout(
"pi: compact compaction context",
["compact", "pi", "sesh-pi"],
base_env,
None,
"## Session Goal",
"find TRIPLA-ESPRESSO",
"## Status",
"Recent check passed",
"test result: ok. 1 passed",
"## Outstanding Context",
"No outstanding items identified",
"## Constraints",
"No specific constraints identified",
)
expect_stdout(
"pi: compact from entry",
["compact", "pi", "sesh-pi", "--from", "pi-entry-2"],
base_env,
None,
"## Status",
"Recent check passed",
)
expect_not_contains(
"pi: compact from excludes earlier entry",
["compact", "pi", "sesh-pi", "--from", "pi-entry-2"],
base_env,
None,
"find TRIPLA-ESPRESSO",
)
expect_not_contains(
"pi: compact until excludes later entry",
["compact", "pi", "sesh-pi", "--from", "pi-entry-2", "--until", "pi-entry-3"],
base_env,
None,
"Recent check passed",
)
expect_failure(
"pi: compact invalid range",
["compact", "pi", "sesh-pi", "--from", "pi-entry-4", "--until", "pi-entry-2"],
base_env,
None,
)
expect_failure(
"pi: compact from with ids",
["compact", "pi", "sesh-pi", "--from", "pi-entry-2", "--ids", "pi-entry-2"],
base_env,
None,
)
expect_not_contains(
"pi: compact filters todo noise",
["compact", "pi", "sesh-pi"],
base_env,
None,
"Created #99",
)
expect_not_contains(
"pi: compact filters hashline noise",
["compact", "pi", "sesh-pi"],
base_env,
None,
"1:928|//",
)
expect_not_contains(
"pi: compact filters file-list noise",
["compact", "pi", "sesh-pi"],
base_env,
None,
".git/",
)
expect_stdout(
"codex: list contexts",
["list", "codex"],
base_env,
None,
"Sessions:",
codex_id,
"/tmp/project",
)
expect_stdout(
"codex: display messages",
["show", "codex", codex_id],
base_env,
None,
"entryId: codex-user-1 (user)",
"DEADBEEF",
"entryId: codex-assistant-1 (assistant)",
"entryId: call-codex-grep (assistant)",
"grep: path=src/main.rs",
"entryId: codex-reasoning-1 (assistant)",
"consider DEADBEEF search plan",
"entryId: call-codex-custom (assistant)",
"lookup: query=DEADBEEF custom",
"entryId: call-codex-shell (assistant)",
"local_shell_call: command=grep DEADBEEF src/main.rs",
"entryId: call-codex-grep-output (tool/call-codex-grep)",
)
expect_stdout(
"codex: clipped display",
["show", "codex", codex_id, "--clip"],
base_env,
None,
"entryId: codex-user-1 (user)",
"entryId: codex-assistant-1 (assistant)",
"tools: grep(path=src/main.rs)",
"tools: lookup(query=DEADBEEF custom)",
"tools: local_shell_call(command=grep DEADBEEF src/main.rs)",
)
expect_success(
"codex: grep match",
["grep", "codex", codex_id, "DEADBEEF"],
base_env,
None,
"DEADBEEF",
)
expect_success(
"codex: grep command match",
["grep", "codex", codex_id, "grep DEADBEEF src/main.rs"],
base_env,
None,
"grep DEADBEEF src/main.rs",
)
expect_stdout(
"goose: list contexts",
["list", "goose"],
base_env,
None,
"Sessions:",
"sesh-goose",
"Review Session",
)
expect_stdout(
"goose: display messages",
["show", "goose", "sesh-goose"],
base_env,
None,
"entryId: goose-1 (user)",
"YANKEE-ECHO",
"entryId: goose-2 (assistant)",
"let me search",
"entryId: goose-3 (assistant)",
"grep: pattern=YANKEE",
"entryId: goose-4 (tool/grep)",
"YANKEE-ECHO",
)
expect_stdout(
"goose: clipped display",
["show", "goose", "sesh-goose", "--clip"],
base_env,
None,
"entryId: goose-1 (user)",
"entryId: goose-2 (assistant)",
"entryId: goose-3 (assistant)",
"entryId: goose-4 (tool/grep)",
)
expect_success(
"goose: grep match",
["grep", "goose", "sesh-goose", "YANKEE-ECHO"],
base_env,
None,
"YANKEE-ECHO",
)
expect_stdout(
"goose: lineage scope",
["show", "goose", "sesh-goose", "--scope", "lineage"],
base_env,
None,
"entryId: goose-1 (user)",
"entryId: goose-4 (tool/grep)",
)
expect_failure(
"cli: unknown client",
["list", "bogus"],
base_env,
None,
)
expect_failure_stdout(
"cli: no arguments",
[],
base_env,
None,
2,
"Usage: goosedump <command> [<args>]",
)
expect_failure(
"cli: grep without context-id",
["grep", "opencode", "foo"],
base_env,
None,
)
expect_failure(
"cli: missing subcommand",
["opencode", "sesh-alpha"],
base_env,
None,
)
expect_failure(
"cli: --page with grep",
["grep", "opencode", "sesh-alpha", "STARTREK", "--page", "1"],
base_env,
None,
)
expect_failure(
"cli: invalid --scope",
["show", "opencode", "sesh-alpha", "--scope", "bogus"],
base_env,
None,
)
expect_failure(
"cli: --page zero",
[
"search",
"opencode",
"sesh-alpha",
"STARTREK",
"--page",
"0",
],
base_env,
None,
)
print(f"SUMMARY pass={pass_count} fail={fail_count} tmp={tmpdir}")
if fail_count != 0:
sys.exit(1)
if __name__ == "__main__":
main()