From 007b78e8dfae1bf493c2b8cced112b6ceafb0b20 Mon Sep 17 00:00:00 2001 From: David Nicholas Date: Mon, 17 Aug 2026 17:14:43 -0700 Subject: [PATCH] Re-poll the update handle so long turns stop reporting failure updateHandle.Get issues a long poll bounded by the SDK's own pollUpdateTimeout (60s in v1.46). When that window closes the gRPC call returns Canceled/DeadlineExceeded and the SDK surfaces WorkflowUpdateServiceTimeoutOrCanceledError WITHOUT retrying, even though the caller's context is still live. That error is about the client call, not the update -- the SDK documents it as "not related to any general concept of timing out or cancelling a running update". The workflow keeps running and the result still arrives; only the poll gave up. So any turn longer than the poll window reports failure while the work quietly succeeds. Observed with a bridged claude-code-swe-agent run: the UI showed Timeout or cancellation waiting for update: stream terminated by RST_STREAM with error code: CANCEL while AgentRun reached Succeeded, BridgedAgentWorkflow reached Completed, and the agent had already pushed its work. Minutes-long runs are the norm for a coding agent, so this is the common case rather than an edge. awaitTurnResult re-polls on that error and is used by both the streaming and non-streaming paths. The loop is bounded by ctx: the SDK returns the same error when the caller hangs up, so ctx.Err() is what separates "the window closed" from "nobody is listening", and prevents spinning. /invoke is deliberately untouched -- it sets its own invokePollTimeout and reports "still running" by design. Co-Authored-By: Claude Opus 5 --- .../internal/gateway/await_turn_test.go | 91 +++++++++++++++++++ engines/temporal/internal/gateway/server.go | 44 ++++++++- 2 files changed, 133 insertions(+), 2 deletions(-) create mode 100644 engines/temporal/internal/gateway/await_turn_test.go diff --git a/engines/temporal/internal/gateway/await_turn_test.go b/engines/temporal/internal/gateway/await_turn_test.go new file mode 100644 index 0000000..893487b --- /dev/null +++ b/engines/temporal/internal/gateway/await_turn_test.go @@ -0,0 +1,91 @@ +package gateway + +// Internal test: awaitTurnResult is the difference between a long agent turn +// succeeding and reporting failure, and it is reachable without standing up +// Temporal — a fake handle is enough to drive every branch. + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/require" + "go.temporal.io/sdk/client" + + "github.com/controller-agent/temporal-engine/internal/temporal/workflows" +) + +// fakeUpdateHandle returns the queued errors in order, then succeeds with reply. +type fakeUpdateHandle struct { + client.WorkflowUpdateHandle + errs []error + reply string + calls int + // onCall runs before each result is returned, so a test can cancel the + // context mid-flight the way a disconnecting client would. + onCall func(call int) +} + +func (f *fakeUpdateHandle) Get(_ context.Context, valuePtr interface{}) error { + f.calls++ + if f.onCall != nil { + f.onCall(f.calls) + } + if len(f.errs) > 0 { + err := f.errs[0] + f.errs = f.errs[1:] + return err + } + if out, ok := valuePtr.(*workflows.TurnResult); ok { + out.Reply = f.reply + } + return nil +} + +func pollTimeout() error { + // Mirrors what the SDK surfaces when its 60s client-side poll window closes + // while the update is still running. + return client.NewWorkflowUpdateServiceTimeoutOrCanceledError(errors.New("context deadline exceeded")) +} + +// The regression this exists for: one poll window closing must not be reported +// as a failed turn. A bridged coding agent routinely outruns the window. +func TestAwaitTurnResultRetriesPastPollWindow(t *testing.T) { + h := &fakeUpdateHandle{errs: []error{pollTimeout(), pollTimeout()}, reply: "done"} + + var result workflows.TurnResult + require.NoError(t, awaitTurnResult(context.Background(), h, &result)) + require.Equal(t, "done", result.Reply) + require.Equal(t, 3, h.calls, "should re-poll after each timeout, then return the result") +} + +// A real update failure must surface immediately rather than being retried — +// otherwise a genuinely broken turn hangs until the client gives up. +func TestAwaitTurnResultReturnsRealErrors(t *testing.T) { + boom := errors.New("workflow update rejected") + h := &fakeUpdateHandle{errs: []error{boom}} + + var result workflows.TurnResult + err := awaitTurnResult(context.Background(), h, &result) + require.ErrorIs(t, err, boom) + require.Equal(t, 1, h.calls, "a non-poll error must not be retried") +} + +// The SDK returns the SAME error type when the caller's context is done, so the +// loop has to check ctx or it spins forever on a disconnected client. +func TestAwaitTurnResultStopsWhenCallerGoesAway(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + + h := &fakeUpdateHandle{ + errs: []error{pollTimeout(), pollTimeout()}, + reply: "unreachable", + onCall: func(call int) { cancel() }, // client hangs up during the first poll + } + + var result workflows.TurnResult + err := awaitTurnResult(ctx, h, &result) + + var pollErr *client.WorkflowUpdateServiceTimeoutOrCanceledError + require.ErrorAs(t, err, &pollErr) + require.Equal(t, 1, h.calls, "must not keep polling once the caller is gone") +} diff --git a/engines/temporal/internal/gateway/server.go b/engines/temporal/internal/gateway/server.go index 8136047..db5d946 100644 --- a/engines/temporal/internal/gateway/server.go +++ b/engines/temporal/internal/gateway/server.go @@ -9,6 +9,7 @@ package gateway import ( "context" "encoding/json" + "errors" "fmt" "log" "net/http" @@ -286,7 +287,7 @@ func (s *Server) handleChatCompletions(c *gin.Context) { } var result workflows.TurnResult - if err := updateHandle.Get(c.Request.Context(), &result); err != nil { + if err := awaitTurnResult(c.Request.Context(), updateHandle, &result); err != nil { log.Printf("turn failed: workflow=%s err=%v", workflowID, err) writeError(c, http.StatusBadGateway, "turn failed: "+err.Error()) return @@ -512,6 +513,45 @@ const ( // keep-alive comments while quiet, then the reply as a content delta once // the update completes. Mid-turn lines come from polling TurnProgressQuery; // the final flush uses the authoritative narration in the turn result. +// awaitTurnResult waits for a user-turn update to finish, re-polling across the +// SDK's client-side poll window. +// +// updateHandle.Get issues a long poll bounded by the SDK's own pollUpdateTimeout +// (60s in v1.46). When that window closes the gRPC call returns Canceled or +// DeadlineExceeded and the SDK surfaces WorkflowUpdateServiceTimeoutOrCanceledError +// WITHOUT retrying, even though the caller's context is still live. That error is +// explicitly about the client call and not the update -- the SDK documents it as +// "not related to any general concept of timing out or cancelling a running +// update" -- so the workflow is still running and the result is still coming. +// +// Left unhandled, every turn longer than the poll window reports failure while +// the work quietly succeeds. That is not an edge case for agent turns: a bridged +// coding agent routinely runs for minutes, so the user sees +// "Timeout or cancellation waiting for update" on a turn that then completes +// normally and writes its result. +// +// Re-polling is the whole fix. The loop is bounded by ctx: once the caller goes +// away, ctx.Err() is set and the same error is returned rather than spinning. +func awaitTurnResult(ctx context.Context, handle client.WorkflowUpdateHandle, result *workflows.TurnResult) error { + for { + err := handle.Get(ctx, result) + if err == nil { + return nil + } + + var pollErr *client.WorkflowUpdateServiceTimeoutOrCanceledError + if !errors.As(err, &pollErr) { + return err + } + + // Distinguishes "the poll window closed" from "the caller hung up": + // the SDK returns this same error for both. + if ctx.Err() != nil { + return err + } + } +} + func (s *Server) streamTurn(c *gin.Context, workflowID, completionID string, updateHandle client.WorkflowUpdateHandle) { c.Header("Content-Type", "text/event-stream") c.Header("Cache-Control", "no-cache") @@ -541,7 +581,7 @@ func (s *Server) streamTurn(c *gin.Context, workflowID, completionID string, upd done := make(chan turnDone, 1) go func() { var result workflows.TurnResult - err := updateHandle.Get(c.Request.Context(), &result) + err := awaitTurnResult(c.Request.Context(), updateHandle, &result) done <- turnDone{result, err} }()