Durable Execution

streamCtx := context.WithoutCancel(r.Context()) // run outlives the HTTP request
eventsCtx, cancelEvents := context.WithCancel(r.Context())
defer cancelEvents() // disconnect stops subscriber only

s, err := a.Stream(streamCtx, prompt, nil)
ch, err := s.Events(eventsCtx)

// Stream ctx owns the run — cancel it only when you want to stop the agent.
// Use a separate Events ctx so a subscriber disconnect does not CancelWorkflow.
agentStream, err := a.Stream(ctx, prompt, nil)
if err != nil {
    return err
}
runID := agentStream.ID()
// Persist runID → DB, file, or cache — before reading events.
// It must survive a process crash.
saveRunID(sessionID, runID)

eventCh, err := agentStream.Events(eventsCtx)
if err != nil {
    return err
}

for ev := range eventCh {
    if ev == nil {
        continue
    }
    if ob, ok := ev.(interface{ Offset() (int64, bool) }); ok {
        if off, has := ob.Offset(); has {
            saveOffset(sessionID, off) // persist before processing
        }
    }
    handleEvent(ev)
}

savedRunID, savedOffset := loadSavedRun(sessionID)
agentStream, err := a.GetAgentStream(ctx, savedRunID)
if err != nil {
    if errors.Is(err, agent.ErrRunAlreadyCompleted) {
        // Workflow finished while disconnected — see below.
        clearSavedRun(sessionID)
        return
    }
    return err
}
resumeCh, err := agentStream.Events(ctx, agent.WithOffset(savedOffset))
if err != nil {
    return err
}

for ev := range resumeCh {
    if ev == nil {
        continue
    }
    if ob, ok := ev.(interface{ Offset() (int64, bool) }); ok {
        if off, has := ob.Offset(); has && off <= savedOffset {
            continue // already processed before the crash
        }
    }
    handleEvent(ev)
}

case agent.AgentEventTypeRunFinished, agent.AgentEventTypeRunError:
    clearSavedRun(sessionID)

agentStream, err := a.GetAgentStream(ctx, savedRunID)
if errors.Is(err, agent.ErrRunAlreadyCompleted) {
    // The run completed successfully while you were disconnected.
    // The work was done — only the streaming view is lost.
    // If WithConversation is configured, the response is in history; start a new turn.
    // Otherwise, start a new run to get a fresh response.
    clearSavedRun(sessionID)
    return
}

agentRun, err := a.Run(ctx, prompt, nil)
if err != nil {
    return err
}
saveRunID(sessionID, agentRun.ID()) // before Get / Done

// ... process crash ...

savedRunID := loadSavedRunID(sessionID)
run, err := a.GetAgentRun(ctx, savedRunID)
if err != nil {
    if errors.Is(err, agent.ErrRunAlreadyCompleted) {
        // Run finished while disconnected — outcome is not on this handle.
        // Load from conversation/memory, or start a new run.
        clearSavedRun(sessionID)
        return nil
    }
    return err // including ErrRunNotFound
}
result, err := run.Get(ctx) // or <-run.Done() then Get
if err != nil {
    return err
}
clearSavedRun(sessionID)
fmt.Println(result.Content)

// GetAgentStream returns a handle for a prior stream run (must still be live).
func (a *Agent) GetAgentStream(ctx context.Context, runID string) (AgentStream, error)

// Events resumes delivery; WithOffset skips already-consumed durable log entries.
func (s AgentStream) Events(ctx context.Context, opts ...AgentStreamOption) (<-chan AgentEvent, error)

func WithOffset(offset int64) AgentStreamOption

// GetAgentRun resumes a still-running non-stream Run by ID.
func (a *Agent) GetAgentRun(ctx context.Context, runID string) (AgentRun, error)