Reconnect

# Temporal
task infra:temporal:up && task infra:temporal:wait
AGENT_RUNTIME=temporal go run ./agent_with_reconnect "What time is it?"

# Restate
task infra:restate:up && task infra:restate:wait
AGENT_RUNTIME=restate go run ./agent_with_reconnect "What time is it?"

// Stream ctx owns the agent run — keep it alive across a "subscriber crash".
agentStream, err := a.Stream(context.Background(), prompt, nil)
runID := agentStream.ID() // Step 1: save before consuming events

eventsCtx, cancelEvents := context.WithCancel(context.Background())
defer cancelEvents()
eventCh, err := agentStream.Events(eventsCtx)

var lastOffset int64
for ev := range eventCh {
    // Step 2: track offset on every event.
    if ob, ok := ev.(interface{ Offset() (int64, bool) }); ok {
        if off, has := ob.Offset(); has {
            lastOffset = off
        }
    }

    // ... handle event ...

    // Simulate crash: cancel Events only (not Stream).
    if ev.Type() == agent.AgentEventTypeTextMessageContent {
        cancelEvents() // drains and closes eventCh; durable run keeps running
        break
    }
}

// Step 3: reconnect from the last offset on "restart".
agentStream, err = a.GetAgentStream(ctx, runID)
if errors.Is(err, agent.ErrRunAlreadyCompleted) {
    log.Fatal("run finished while disconnected — start a new run")
}
resumeCh, err := agentStream.Events(ctx, agent.WithOffset(lastOffset))

// Step 4: events at offset ≤ lastOffset may be redelivered;
// discard duplicates if needed, then consume normally.
for ev := range resumeCh {
    // ... handle event, clear saved state on RUN_FINISHED / RUN_ERROR ...
}

user: What time is it?
[run_id] agent-stream-reconnect-agent-<uuid>
--- stream start (phase A: first text chunk, then simulating crash) ---
[RUN_STARTED] runID=agent-stream-reconnect-agent-<uuid>
[TEXT_START] msgID=<id>
The current time is

=== simulated crash: saved runID=agent-stream-reconnect-agent-<uuid> lastOffset=3 ===

=== process restart: reconnecting from offset 3 ===

--- stream resumed (phase B: events from last saved offset) ---
 ...
[TEXT_END]
[RUN_FINISHED]
--- stream end ---