Sessions & Replay

import gatewaysdk.tracing as tracing

with tracing.session("checkout-regression-001"):
    with tracing.trace("turn-1", input=prompt):
        ...   # every trace, LLM call, and tool span in here is grouped

with tracing.session("checkout-regression-001"):
    history = None
    for message in user_messages:
        turn_input = [*history, {"role": "user", "content": message}] if history else message
        result = await Runner.run(agent, turn_input)     # one trace per turn
        history = result.to_input_list()

from gatewaysdk.sessions import load_session

session = load_session("session-id")   # uses GATEWAY_HOST / _PUBLIC_KEY / _SECRET_KEY

for turn in session.turns():           # traces in chronological order
    print(turn.input, "→", turn.output)

session.turn(0)                        # nth turn (negative indices ok)
session.turn(0).user_message()         # the user message that started the turn
session.user_messages()                # every turn's user message, in order
session.final_output()                 # last turn's output
session.tool_calls()                   # every TOOL observation, session-wide
session.generations()                  # every GENERATION observation
session.select(type="TOOL", name="checkout")   # generic filter

session.to_fixture("tests/fixtures/checkout_regression.json")

from gatewaysdk.replay import ReplaySuite

suite = ReplaySuite.from_task_set("checkout-regressions")   # name or datasetId

for case in suite:
    print(case.name, "→", case.prompt)
    recording = case.recording    # full SessionRecording, loaded lazily

case = suite["earbuds-regression"]
stubs = case.stubs()                          # ToolStubRegistry

out = my_agent(case.prompt, tools=stubs.as_registry())

stubs = case.stubs(miss_policy="reuse", local={"compare_products": my_impl})

result = case.check(out, stubs=stubs)
assert result.passed, result.explain()

result = case.check(final_reply, stubs=stubs, mode="exact")
print(result.trace_diff.explain())
# 1:1 replay: all 9 recorded tool calls reproduced in order
#   -- or --
# trace diverges from recording (3 matched):
#   not reproduced: deal_scout("best wireless earbuds")
#   extra call: search_products({"query": "usb chargers"}) [reuse]
#   out of order: add_to_cart({"product": "Flux 65W GaN Charger"})

recording = case.recording
recording.turns()               # every turn (trace), in order
recording.turn(1).tool_calls    # a specific turn's tools, with full IO
recording.final_output()        # recorded final answer, e.g. for a judge
recording.user_messages()       # per-turn user inputs

import gatewaysdk.replay as replay

report = replay.suite("checkout-regressions").run(my_agent)
print(report.explain())                  # "14 PASS · 1 FAIL · 1 DRIFT" + details
assert report.ok                         # ok == no FAILs (drift doesn't block)
report.run_name                          # the platform run this was recorded as
report.recording_errors                  # non-empty if platform recording failed

def my_agent(prompt, tools): ...   # the common shape (tools = stub registry)
def my_agent(prompt): ...          # prompt-only agents
def my_agent(case): ...            # full ReplayCase (case.stubs(), case.recording)

import pytest
from gatewaysdk.replay.pytest import replay_cases

@replay_cases("checkout-regressions")
def test_regression(case):
    stubs = case.stubs(miss_policy="reuse")
    out = my_agent(case.prompt, tools=stubs.as_registry())
    result = case.check(out, stubs=stubs)
    assert result.status != "FAIL", result.explain()   # regression → block
    if result.drifted:                                 # stale recording →
        pytest.xfail(result.explain())                 # flag, don't block

suite.snapshot("tests/fixtures/")   # one <case>.recording.json per case