From 446b718dbd21106edd41513854cfb7ecda70830c Mon Sep 17 00:00:00 2001 From: Asim Aslam Date: Mon, 21 Sep 2026 08:01:39 +0100 Subject: [PATCH 1/2] fix(agent): resume timeout and rate-limited runs from checkpoints --- agent/checkpoint.go | 5 +- agent/resilience_test.go | 10 +- agent/run_errors_test.go | 2 +- agent/transient_resume_test.go | 110 ++++++++++++++++++ .../en/docs/guides/debugging-agents.md | 23 ++++ 5 files changed, 144 insertions(+), 6 deletions(-) create mode 100644 agent/transient_resume_test.go diff --git a/agent/checkpoint.go b/agent/checkpoint.go index f90e548524..780d49aa80 100644 --- a/agent/checkpoint.go +++ b/agent/checkpoint.go @@ -90,7 +90,8 @@ func (a *agentImpl) ResumeInput(ctx context.Context, runID, input string) (*Resp // Resume returns the response for a checkpointed agent run. Completed runs are // returned from the checkpoint without calling the model or replaying tool -// calls; failed or in-progress runs continue from the saved input message. +// calls; failed, interrupted (timeout/rate_limited), or in-progress runs +// continue from the saved input message and reuse completed tool results. func Resume(ctx context.Context, ag Agent, runID string) (*Response, error) { a, ok := ag.(Resumer) if !ok { @@ -201,7 +202,7 @@ func (a *agentImpl) pending(ctx context.Context) ([]flow.Run, error) { func terminalAgentRunStatus(status string) bool { switch status { - case "done", "canceled", "timeout", "rate_limited", "expired": + case "done", "canceled", "expired": return true default: return false diff --git a/agent/resilience_test.go b/agent/resilience_test.go index 7400d62a62..ec3f5ec3f8 100644 --- a/agent/resilience_test.go +++ b/agent/resilience_test.go @@ -289,7 +289,7 @@ func TestSlowProviderTimeoutPreventsLateToolSideEffects(t *testing.T) { } } -func TestAskCheckpointRecordsTerminalOperationalFailureStatus(t *testing.T) { +func TestAskCheckpointRecordsOperationalFailureStatus(t *testing.T) { tests := []struct { name string err error @@ -333,8 +333,12 @@ func TestAskCheckpointRecordsTerminalOperationalFailureStatus(t *testing.T) { if got := runs[0].Steps[0].ErrorKind; got != string(model.ClassifyError(tt.err)) { t.Fatalf("step error kind = %q, want %q", got, model.ClassifyError(tt.err)) } - if pending, err := Pending(context.Background(), a); err != nil || len(pending) != 0 { - t.Fatalf("Pending = %#v, %v; want no terminal run", pending, err) + wantPending := 1 + if tt.want == "canceled" { + wantPending = 0 + } + if pending, err := Pending(context.Background(), a); err != nil || len(pending) != wantPending { + t.Fatalf("Pending = %#v, %v; want %d runs", pending, err, wantPending) } }) } diff --git a/agent/run_errors_test.go b/agent/run_errors_test.go index c4e769f066..68dc8c49e2 100644 --- a/agent/run_errors_test.go +++ b/agent/run_errors_test.go @@ -15,7 +15,7 @@ func TestTypedRunErrors(t *testing.T) { ctx := context.Background() cp := flow.StoreCheckpoint(store.NewMemoryStore(), "typed-errors") a := newTestAgent(Name("typed-errors"), WithCheckpoint(cp)) - for _, status := range []string{"timeout", "canceled", "rate_limited", "expired"} { + for _, status := range []string{"canceled", "expired"} { if err := cp.Save(ctx, flow.Run{ID: status, Status: status}); err != nil { t.Fatal(err) } diff --git a/agent/transient_resume_test.go b/agent/transient_resume_test.go new file mode 100644 index 0000000000..e38d0b7f4b --- /dev/null +++ b/agent/transient_resume_test.go @@ -0,0 +1,110 @@ +package agent + +import ( + "context" + "encoding/json" + "io" + "testing" + + "go-micro.dev/v6/flow" + "go-micro.dev/v6/model" + "go-micro.dev/v6/store" +) + +func TestResumeTransientFailureAfterRestart(t *testing.T) { + for _, failure := range []struct { + name string + err error + }{ + {"timeout", context.DeadlineExceeded}, {"rate_limited", testStatusError{code: 429}}, + } { + for _, mode := range []string{"resume", "stream", "pending"} { + t.Run(failure.name+"/"+mode, func(t *testing.T) { + ctx := context.Background() + st := store.NewMemoryStore() + cp := flow.StoreCheckpoint(st, "recovery") + modelCalls, toolCalls := 0, 0 + fakeGen = func(ctx context.Context, opts model.Options, req *model.Request) (*model.Response, error) { + modelCalls++ + if req.Prompt != "charge once" { + t.Fatalf("lost original request: %q", req.Prompt) + } + result := opts.ToolHandler(ctx, model.ToolCall{ID: "charge", Name: "charge", Input: map[string]any{"order": "42"}}) + if result.Content != "paid" { + t.Fatalf("lost tool result: %+v", result) + } + if modelCalls == 1 { + return nil, failure.err + } + return &model.Response{Reply: "recovered"}, nil + } + defer func() { fakeGen = nil }() + makeAgent := func() *agentImpl { + return newTestAgent(Name("recovery"), WithStore(st), WithCheckpoint(cp), WithTool("charge", "charge", nil, func(context.Context, map[string]any) (string, error) { toolCalls++; return "paid", nil })) + } + original := makeAgent() + if _, err := original.Ask(ctx, "charge once"); err == nil { + t.Fatal("expected provider failure") + } + runs, err := Pending(ctx, original) + if err != nil || len(runs) != 1 || runs[0].Status != failure.name { + t.Fatalf("pending: %+v %v", runs, err) + } + id := runs[0].ID + restarted := makeAgent() + var response *Response + switch mode { + case "resume": + response, err = Resume(ctx, restarted, id) + case "pending": + var failedID string + failedID, err = ResumePending(ctx, restarted) + if failedID != "" { + t.Fatalf("failed run: %s", failedID) + } + if err == nil { + run, _, loadErr := cp.Load(ctx, id) + if loadErr != nil { + t.Fatal(loadErr) + } + response = new(Response) + err = json.Unmarshal(run.State.Data, response) + } + case "stream": + var stream AgentStream + stream, err = ResumeStreamAsk(ctx, restarted, id) + if err != nil { + t.Fatal(err) + } + defer stream.Close() + for { + event, recvErr := stream.Recv() + if recvErr == io.EOF { + break + } + if recvErr != nil { + err = recvErr + break + } + if event.Type == StreamEventDone { + response = event.Response + } + } + } + if err != nil { + t.Fatal(err) + } + if response == nil || response.RunID != id || response.Reply != "recovered" { + t.Fatalf("response: %+v", response) + } + if toolCalls != 1 || modelCalls != 2 { + t.Fatalf("tool calls=%d, model calls=%d", toolCalls, modelCalls) + } + run, ok, err := cp.Load(ctx, id) + if err != nil || !ok || run.Status != "done" { + t.Fatalf("final checkpoint: %+v %v", run, err) + } + }) + } + } +} diff --git a/internal/website/content/en/docs/guides/debugging-agents.md b/internal/website/content/en/docs/guides/debugging-agents.md index 9f36f8b84b..14c40d5375 100644 --- a/internal/website/content/en/docs/guides/debugging-agents.md +++ b/internal/website/content/en/docs/guides/debugging-agents.md @@ -263,3 +263,26 @@ micro call '{}' Redact secrets and user data. If you enabled `agent.TraceInputs(true)`, inspect the JSON before sharing it because prompts may be present. + +### Recovering interrupted runs + +With `agent.WithCheckpoint(...)`, provider `timeout` and `rate_limited` outcomes +remain discoverable through `agent.Pending` and can be continued with +`agent.Resume(ctx, ag, runID)`, `agent.ResumeStreamAsk`, or `agent.ResumePending`. +Use a fresh context after a deadline and wait for the provider's rate-limit +window before retrying. Recovery keeps the run ID and saved request and reuses +completed tool results, including after recreating the agent with the same +checkpoint store. Canceled and expired runs remain terminal. An interrupted +side effect without a saved result can still be retried: use idempotency keys +for such tools; checkpointing is not an exactly-once guarantee. + +The defaults remain 30 seconds per model call and 30 seconds per tool call. +For slower models or multi-tool turns, configure the budgets explicitly: + +```go +agent.ModelCallTimeout(120 * time.Second) +agent.ToolCallTimeout(60 * time.Second) +``` + +The caller's context and RPC request deadline must also allow the whole turn to +finish. Increasing a per-call timeout cannot extend an earlier caller deadline. From 38876570749422c104a7cd035b46cb5e79fc8013 Mon Sep 17 00:00:00 2001 From: Asim Aslam Date: Mon, 21 Sep 2026 08:06:36 +0100 Subject: [PATCH 2/2] docs: align pending run contract with transient recovery --- internal/website/content/en/docs/guides/durability.md | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/internal/website/content/en/docs/guides/durability.md b/internal/website/content/en/docs/guides/durability.md index 30d246552e..c2e042a632 100644 --- a/internal/website/content/en/docs/guides/durability.md +++ b/internal/website/content/en/docs/guides/durability.md @@ -72,9 +72,10 @@ one run therefore share a completed result. This is not a cross-run idempotency key and does not guarantee that a model will choose the same arguments after a restart. -Agent pending runs exclude terminal `done`, `canceled`, `timeout`, -`rate_limited`, and `expired` statuses. Paused runs can still appear in pending -results; an input-required pause needs the input helper, so an unattended +Agent pending runs exclude terminal `done`, `canceled`, and `expired` statuses. +Interrupted `timeout` and `rate_limited` runs remain pending and can resume with +a fresh context, reusing saved input and completed tool results. Paused runs can +still appear in pending results; an input-required pause needs the input helper, so an unattended `ResumePending` loop can stop there. ## What is not guaranteed