Time-travel replay

from shipit_agent.replay import TraceReplayer
from shipit_agent.tracing import FileTraceStore

# 1. Load a saved trace
store = FileTraceStore('./traces')
replayer = TraceReplayer.from_store(store, trace_id='run-2026-05-09-abc')

# 2. Inspect — find every tool call
tool_indices = replayer.event_indices_by_type('tool_called')
for i in tool_indices:
    ev = replayer.events[i]
    print(f"event {i}: {ev.payload.get('tool')}")

# 3. Fork before the third tool call, with a tweaked prompt
fork = replayer.fork(
    at_event=tool_indices[2],
    edit_user_message='Try answering without searching the web first.',
)

# 4. Resume on a fresh agent
from shipit_agent import Agent
result = fork.continue_from(agent=Agent(llm=my_llm))

print(result.output)

# From an in-memory or file-backed store
TraceReplayer.from_store(store, trace_id='abc')

# Already-loaded record
TraceReplayer.from_record(record)

# Direct JSON file (FileTraceStore format)
TraceReplayer.from_file('./traces/run-abc.json')

replayer.trace_id                                # 'run-abc'
replayer.events                                  # list[AgentEvent]
replayer.metadata                                # dict carried with the trace
len(replayer)                                    # number of events

# Find events by type — useful for tool-call analysis
replayer.event_indices_by_type('tool_called')    # → [3, 7, 12]
replayer.event_indices_by_type('tool_completed') # → [4, 8, 13]

# Reconstruct messages as they existed at any point
replayer.messages_at(event_index=12)             # list[Message]

# All user messages with their event indices
replayer.find_user_messages()                    # → [(0, 'first prompt'), ...]

fork = replayer.fork(
    at_event=12,                            # 0-indexed event to fork at
    edit_user_message='Try a narrower question.',  # optional
    extra_metadata={'reason': 'debug-1'},   # merged into checkpoint metadata
)
# returns a ReplayCheckpoint

result = fork.continue_from(agent=fresh_agent, max_validation_retries=2)

from shipit_agent.replay import diff_traces

d = diff_traces(baseline_replayer, fork_replayer)

if d.identical:
    print("Same behaviour — your edit had no effect.")
else:
    print(f"Diverged at event {d.diverged_at}")
    print(f"  matched events: {d.matched}")
    print(f"  type mismatches: {len(d.type_mismatches)}")
    for line in d.to_lines():
        print(line)

replayer = TraceReplayer.from_file('./traces/run-2026-05-09-abc.json')

# 1. What was the last tool that ran successfully?
tool_completes = replayer.event_indices_by_type('tool_completed')
print(f"last successful tool at event {tool_completes[-1]}")

# 2. What did the agent see right before the failure?
last_good = tool_completes[-1]
context = replayer.messages_at(last_good)
print(f"history at fork point: {len(context)} messages")

# 3. Fork there with a more constrained prompt
fork = replayer.fork(
    at_event=last_good,
    edit_user_message=(
        "We've already gathered the data we need — "
        "don't search the web again. Just produce the report."
    ),
)

# 4. Resume on a fresh agent
from shipit_agent import Agent
result = fork.continue_from(agent=Agent(llm=opus_llm, tools=[ReportTool()]))

# 5. Compare to baseline
baseline = TraceReplayer.from_file('./traces/run-2026-05-09-abc.json')
fresh = TraceReplayer.from_record(result.agent_result.events_as_record())
print('\n'.join(diff_traces(baseline, fresh).to_lines()))

TraceReplayer(record: TraceRecord)
TraceReplayer.from_record(record)
TraceReplayer.from_store(store, trace_id)
TraceReplayer.from_file(path)