From 33b2cde4d28c0433ea807ae85a040290014e31ac Mon Sep 17 00:00:00 2001 From: Austin Kurpuis Date: Sun, 23 Aug 2026 05:43:55 -0700 Subject: [PATCH] Don't report a bridged agent turn as failed on gRPC transport cancellation /invoke's status poll treated any handle.Get error other than ctx's own deadline as a genuine turn failure. But the Temporal SDK's WorkflowUpdateServiceTimeoutOrCanceledError (e.g. "stream terminated by RST_STREAM with error code: CANCEL") is explicitly documented as being about the client call, not the update -- the workflow keeps running either way. Misclassifying it made agent-orchestrator report a bridged coding agent's routine multi-minute turn (e.g. launching claude-code-swe- agent to open a PR) as an outright failure while the AgentRun kept working in the background. Now any WorkflowUpdateServiceTimeoutOrCanceledError is treated as pending, matching how the streaming chat facade's awaitTurnResult already handles the same SDK error. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01D3zV2JdVTYGuS1KGwyXCJz --- engines/temporal/internal/gateway/invoke.go | 24 +++- .../internal/gateway/invoke_status_test.go | 107 ++++++++++++++++++ 2 files changed, 129 insertions(+), 2 deletions(-) create mode 100644 engines/temporal/internal/gateway/invoke_status_test.go diff --git a/engines/temporal/internal/gateway/invoke.go b/engines/temporal/internal/gateway/invoke.go index c5c99f0..9e8b950 100644 --- a/engines/temporal/internal/gateway/invoke.go +++ b/engines/temporal/internal/gateway/invoke.go @@ -294,8 +294,15 @@ func (s *Server) handleInvokeStatus(c *gin.Context) { ToolCalls: result.PendingToolCalls, }) - case ctx.Err() != nil && c.Request.Context().Err() == nil: - // Our own deadline, not the client's: the turn is simply still running. + case ctx.Err() != nil && c.Request.Context().Err() == nil, isTransportTimeoutOrCancel(err): + // Either our own invokePollTimeout lapsed, or the SDK's long-poll gRPC + // call was itself cancelled/timed out (WorkflowUpdateServiceTimeoutOr + // CanceledError -- e.g. "stream terminated by RST_STREAM with error + // code: CANCEL"). The SDK documents that error as being about the + // client call, NOT the update: the workflow is still running either + // way. Reporting it as "failed" here made agent-orchestrator surface + // a bridged coding agent's routine multi-minute turn as an outright + // failure while the AgentRun kept working in the background. // Best-effort narration alongside it -- a query failure or an inactive // turn (nothing narrated yet) just means an empty Progress, never an // error response, since the pending status itself is still accurate. @@ -326,6 +333,19 @@ func isUnknownUpdate(err error) bool { return errors.As(err, ¬Found) } +// isTransportTimeoutOrCancel reports whether err is the SDK's +// WorkflowUpdateServiceTimeoutOrCanceledError (e.g. wrapping "stream +// terminated by RST_STREAM with error code: CANCEL") -- a gRPC long-poll call +// that got cancelled or timed out on its own terms, independent of our +// invokePollTimeout ever firing. The SDK's own doc comment on the type is +// explicit that this is "not related to any general concept of timing out or +// cancelling a running update": the workflow update itself is unaffected. +// See awaitTurnResult in server.go for the streaming endpoint's analogous fix. +func isTransportTimeoutOrCancel(err error) bool { + var pollErr *client.WorkflowUpdateServiceTimeoutOrCanceledError + return errors.As(err, &pollErr) +} + // Invocation ids join the two halves Temporal needs to reconstruct an update // handle. A '.' is unambiguous as the separator: sanitizeID maps everything // outside [A-Za-z0-9_-] to '-', so the workflow id half never contains one, diff --git a/engines/temporal/internal/gateway/invoke_status_test.go b/engines/temporal/internal/gateway/invoke_status_test.go new file mode 100644 index 0000000..53d94e3 --- /dev/null +++ b/engines/temporal/internal/gateway/invoke_status_test.go @@ -0,0 +1,107 @@ +package gateway + +// Internal test: handleInvokeStatus is the /invoke poll endpoint +// agent-orchestrator drives for every bridged-agent turn. A fake +// client.Client (embedding the interface so only the two methods this +// handler calls need overriding) is enough to reach every branch without +// standing up Temporal. + +import ( + "context" + "errors" + "net/http/httptest" + "testing" + + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" + "go.temporal.io/api/serviceerror" + "go.temporal.io/sdk/client" + "go.temporal.io/sdk/converter" +) + +// fakeTemporalClient overrides only what handleInvokeStatus calls; every +// other client.Client method panics if reached, which would fail the test +// loudly rather than silently doing the wrong thing. +type fakeTemporalClient struct { + client.Client + handle client.WorkflowUpdateHandle +} + +func (f *fakeTemporalClient) GetWorkflowUpdateHandle(client.GetWorkflowUpdateHandleOptions) client.WorkflowUpdateHandle { + return f.handle +} + +func (f *fakeTemporalClient) QueryWorkflow(context.Context, string, string, string, ...interface{}) (converter.EncodedValue, error) { + return nil, errors.New("no progress recorded in this test") +} + +func invokeStatusRequest(t *testing.T, s *Server, invocationID string) *httptest.ResponseRecorder { + t.Helper() + gin.SetMode(gin.TestMode) + rec := httptest.NewRecorder() + c, _ := gin.CreateTestContext(rec) + c.Params = gin.Params{{Key: "id", Value: invocationID}} + c.Request = httptest.NewRequest("GET", "/invoke/"+invocationID, nil) + s.handleInvokeStatus(c) + return rec +} + +// The regression this exists for: the SDK's long-poll gRPC call getting +// cancelled or timing out on its own terms (surfacing as e.g. "stream +// terminated by RST_STREAM with error code: CANCEL") is NOT the update +// failing -- the SDK's own doc comment says so explicitly -- but reporting it +// as invokeStatusFailed made agent-orchestrator show a bridged coding agent's +// routine multi-minute turn as an outright failure while the AgentRun kept +// working in the background. +func TestHandleInvokeStatusTreatsTransportCancelAsPending(t *testing.T) { + handle := &fakeUpdateHandle{errs: []error{ + client.NewWorkflowUpdateServiceTimeoutOrCanceledError(errors.New("stream terminated by RST_STREAM with error code: CANCEL")), + }} + s := NewServer(&fakeTemporalClient{handle: handle}, "task-queue", nil) + + rec := invokeStatusRequest(t, s, encodeInvocationID("conversation-abc", "6ba7b810-9dad-11d1-80b4-00c04fd430c8")) + + require.Equal(t, 200, rec.Code) + require.JSONEq(t, `{"id":"conversation-abc.6ba7b810-9dad-11d1-80b4-00c04fd430c8","status":"pending"}`, rec.Body.String()) +} + +// A genuinely failed update (the workflow rejected it, or the turn errored) +// must still surface as failed -- only the transport-cancellation shape is +// reclassified. +func TestHandleInvokeStatusReportsARealFailure(t *testing.T) { + handle := &fakeUpdateHandle{errs: []error{errors.New("turn failed: launch_error")}} + s := NewServer(&fakeTemporalClient{handle: handle}, "task-queue", nil) + + rec := invokeStatusRequest(t, s, encodeInvocationID("conversation-abc", "6ba7b810-9dad-11d1-80b4-00c04fd430c8")) + + require.Equal(t, 200, rec.Code) + require.JSONEq(t, `{"id":"conversation-abc.6ba7b810-9dad-11d1-80b4-00c04fd430c8","status":"failed","error":"turn failed: launch_error"}`, rec.Body.String()) +} + +// An id naming an update Temporal has never heard of (aged-out workflow, or a +// caller-forged id) is a 404, distinct from a turn that ran and failed. +func TestHandleInvokeStatusUnknownUpdateIs404(t *testing.T) { + handle := &fakeUpdateHandle{errs: []error{serviceerror.NewNotFound("not found")}} + s := NewServer(&fakeTemporalClient{handle: handle}, "task-queue", nil) + + rec := invokeStatusRequest(t, s, encodeInvocationID("conversation-abc", "6ba7b810-9dad-11d1-80b4-00c04fd430c8")) + + require.Equal(t, 404, rec.Code) +} + +// A turn that succeeds on the very next poll (no transport hiccup at all) +// still reports normally -- the transport-cancel handling above must not +// swallow or delay an ordinary success. +func TestHandleInvokeStatusSucceededReportsTheReply(t *testing.T) { + handle := &fakeUpdateHandle{reply: "Opened PR #42."} + s := NewServer(&fakeTemporalClient{handle: handle}, "task-queue", nil) + + rec := invokeStatusRequest(t, s, encodeInvocationID("conversation-abc", "6ba7b810-9dad-11d1-80b4-00c04fd430c8")) + + require.Equal(t, 200, rec.Code) + require.JSONEq(t, `{ + "id":"conversation-abc.6ba7b810-9dad-11d1-80b4-00c04fd430c8", + "status":"succeeded", + "result":"Opened PR #42." + }`, rec.Body.String()) +}