diff --git a/CHANGELOG.md b/CHANGELOG.md index 1238f67..7b541c5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,11 @@ existing `0.1.0` release; earlier development prereleases are not listed. ## [Unreleased] +### Added + +- Add automatic Thread SSE resumption for Forward and Managed, preserving the latest event cursor and request options across retryable disconnects; thread termination stops the subscription while idle events keep it open. +- Add a typed `Last-Event-ID` parameter to Managed Thread streams. + ## [0.13.0] ### Added diff --git a/README.md b/README.md index 293d917..219b791 100644 --- a/README.md +++ b/README.md @@ -478,6 +478,43 @@ if err := stream.Err(); err != nil { The resumable stream checkpoints `LastEventID` only after yielding a decoded event, then reconnects transport/read failures and unexpected EOF with cancellable jittered exponential backoff from 500 ms to 10 seconds. Backoff resets only after an event arrives on a connection that has remained healthy for more than five seconds. It preserves stream params and request options on every connection. HTTP retry classification matches ordinary SDK requests: 408, 429, 5xx and transport failures retry (subject to `x-should-retry`), while 409 and other non-retryable 4xx errors stop. It neither clears an invalid cursor nor queries event history. Events are not deduplicated because multiple delta frames can legitimately share one ID. A yielded `session.status_terminated` or `session.deleted` event ends the stream normally; context cancellation and `Close` both stop an active connection or backoff wait. + +### Thread event resumption + +Forward and Managed also support automatic resumption for a single Thread. Use the +same retry and cancellation behavior as Session streams, with a thread-scoped +`Last-Event-ID` cursor: + +```go +// Forward +stream := forwardClient.Sessions.Threads.Events.NewResumableStream(ctx, sessionID, threadID, forward.SessionThreadEventStreamParams{ + LastEventID: forward.String(afterID), +}) +defer stream.Close() +for stream.Next() { + handle(stream.Current()) +} +if err := stream.Err(); err != nil { + return err +} + +// Managed uses the parent session in params. +managedStream := managedClient.Sessions.Threads.Events.NewResumableStream(ctx, threadID, managed.SessionThreadEventStreamParams{ + SessionID: sessionID, + LastEventID: managed.String(afterID), +}) +defer managedStream.Close() +``` + +`session.thread_status_idle` ends only the current turn, so the subscription stays +open. `session.thread_status_terminated`, `session.status_terminated`, or +`session.deleted` ends it normally. A non-retryable error (including an invalid +cursor or missing thread) is returned to the caller; the SDK never clears the +cursor to silently start over. The existing one-shot Thread stream remains +available. See the [Forward Thread stream](https://docs.qoder.com/cloud-agents/api/forward/sessions/stream-thread-events) +and [Managed Thread stream](https://docs.qoder.com/cloud-agents/api/sessions/threads/events/stream) +contracts for cursor validity. + ## Request fields Required fields take plain values; optional scalars take `param.Opt[T]`. Use `forward.String`, `managed.String`, `Bool`, `Int` and `Float` to set an optional value explicitly. diff --git a/docs/api/reference.md b/docs/api/reference.md index 0f45ae8..ae46612 100644 --- a/docs/api/reference.md +++ b/docs/api/reference.md @@ -687,6 +687,12 @@ import "github.com/QoderAI/qoder-cloud-agents-sdk-go/forward" - [func \(s \*ResumableSessionEventStream\) Err\(\) error](<#ResumableSessionEventStream.Err>) - [func \(s \*ResumableSessionEventStream\) LastEventID\(\) string](<#ResumableSessionEventStream.LastEventID>) - [func \(s \*ResumableSessionEventStream\) Next\(\) bool](<#ResumableSessionEventStream.Next>) +- [type ResumableSessionThreadEventStream](<#ResumableSessionThreadEventStream>) + - [func \(s \*ResumableSessionThreadEventStream\) Close\(\) error](<#ResumableSessionThreadEventStream.Close>) + - [func \(s \*ResumableSessionThreadEventStream\) Current\(\) SessionEvent](<#ResumableSessionThreadEventStream.Current>) + - [func \(s \*ResumableSessionThreadEventStream\) Err\(\) error](<#ResumableSessionThreadEventStream.Err>) + - [func \(s \*ResumableSessionThreadEventStream\) LastEventID\(\) string](<#ResumableSessionThreadEventStream.LastEventID>) + - [func \(s \*ResumableSessionThreadEventStream\) Next\(\) bool](<#ResumableSessionThreadEventStream.Next>) - [type Schedule](<#Schedule>) - [func \(r Schedule\) RawJSON\(\) string](<#Schedule.RawJSON>) - [func \(r \*Schedule\) UnmarshalJSON\(data \[\]byte\) error](<#Schedule.UnmarshalJSON>) @@ -871,6 +877,7 @@ import "github.com/QoderAI/qoder-cloud-agents-sdk-go/forward" - [func NewSessionThreadEventService\(opts ...option.RequestOption\) SessionThreadEventService](<#NewSessionThreadEventService>) - [func \(r \*SessionThreadEventService\) List\(ctx context.Context, sessionID string, threadID string, params SessionThreadEventListParams, opts ...option.RequestOption\) \(res \*pagination.Page\[SessionEvent\], err error\)](<#SessionThreadEventService.List>) - [func \(r \*SessionThreadEventService\) ListAutoPaging\(ctx context.Context, sessionID string, threadID string, params SessionThreadEventListParams, opts ...option.RequestOption\) \*pagination.PageAutoPager\[SessionEvent\]](<#SessionThreadEventService.ListAutoPaging>) + - [func \(r \*SessionThreadEventService\) NewResumableStream\(ctx context.Context, sessionID string, threadID string, params SessionThreadEventStreamParams, opts ...option.RequestOption\) \*ResumableSessionThreadEventStream](<#SessionThreadEventService.NewResumableStream>) - [func \(r \*SessionThreadEventService\) StreamEvents\(ctx context.Context, sessionID string, threadID string, params SessionThreadEventStreamParams, opts ...option.RequestOption\) \*ssestream.Stream\[SessionEvent\]](<#SessionThreadEventService.StreamEvents>) - [type SessionThreadEventStreamParams](<#SessionThreadEventStreamParams>) - [func \(r SessionThreadEventStreamParams\) URLQuery\(\) \(url.Values, error\)](<#SessionThreadEventStreamParams.URLQuery>) @@ -9364,6 +9371,62 @@ func (s *ResumableSessionEventStream) Next() bool + +## type [ResumableSessionThreadEventStream]() + +ResumableSessionThreadEventStream reconnects a single thread's event stream. A thread/session termination or session deletion event ends the stream normally. + +```go +type ResumableSessionThreadEventStream struct { + // contains filtered or unexported fields +} +``` + + +### func \(\*ResumableSessionThreadEventStream\) [Close]() + +```go +func (s *ResumableSessionThreadEventStream) Close() error +``` + + + + +### func \(\*ResumableSessionThreadEventStream\) [Current]() + +```go +func (s *ResumableSessionThreadEventStream) Current() SessionEvent +``` + + + + +### func \(\*ResumableSessionThreadEventStream\) [Err]() + +```go +func (s *ResumableSessionThreadEventStream) Err() error +``` + + + + +### func \(\*ResumableSessionThreadEventStream\) [LastEventID]() + +```go +func (s *ResumableSessionThreadEventStream) LastEventID() string +``` + + + + +### func \(\*ResumableSessionThreadEventStream\) [Next]() + +```go +func (s *ResumableSessionThreadEventStream) Next() bool +``` + + + ## type [Schedule]() @@ -11823,6 +11886,15 @@ func (r *SessionThreadEventService) ListAutoPaging(ctx context.Context, sessionI + +### func \(\*SessionThreadEventService\) [NewResumableStream]() + +```go +func (r *SessionThreadEventService) NewResumableStream(ctx context.Context, sessionID string, threadID string, params SessionThreadEventStreamParams, opts ...option.RequestOption) *ResumableSessionThreadEventStream +``` + +NewResumableStream subscribes to Thread Events and reconnects retryable failures with the latest yielded Last\-Event\-ID cursor. Thread idle events do not stop it. + ### func \(\*SessionThreadEventService\) [StreamEvents]() @@ -17469,6 +17541,12 @@ Qoder managed API definitions. - [func \(s \*ResumableSessionEventStream\) Err\(\) error](<#ResumableSessionEventStream.Err>) - [func \(s \*ResumableSessionEventStream\) LastEventID\(\) string](<#ResumableSessionEventStream.LastEventID>) - [func \(s \*ResumableSessionEventStream\) Next\(\) bool](<#ResumableSessionEventStream.Next>) +- [type ResumableSessionThreadEventStream](<#ResumableSessionThreadEventStream>) + - [func \(s \*ResumableSessionThreadEventStream\) Close\(\) error](<#ResumableSessionThreadEventStream.Close>) + - [func \(s \*ResumableSessionThreadEventStream\) Current\(\) ManagedAgentsStreamSessionThreadEventsUnion](<#ResumableSessionThreadEventStream.Current>) + - [func \(s \*ResumableSessionThreadEventStream\) Err\(\) error](<#ResumableSessionThreadEventStream.Err>) + - [func \(s \*ResumableSessionThreadEventStream\) LastEventID\(\) string](<#ResumableSessionThreadEventStream.LastEventID>) + - [func \(s \*ResumableSessionThreadEventStream\) Next\(\) bool](<#ResumableSessionThreadEventStream.Next>) - [type SearchResultBlockParam](<#SearchResultBlockParam>) - [func \(r SearchResultBlockParam\) MarshalJSON\(\) \(data \[\]byte, err error\)](<#SearchResultBlockParam.MarshalJSON>) - [func \(r \*SearchResultBlockParam\) UnmarshalJSON\(data \[\]byte\) error](<#SearchResultBlockParam.UnmarshalJSON>) @@ -17631,6 +17709,7 @@ Qoder managed API definitions. - [func NewSessionThreadEventService\(opts ...option.RequestOption\) \(r SessionThreadEventService\)](<#NewSessionThreadEventService>) - [func \(r \*SessionThreadEventService\) List\(ctx context.Context, threadID string, params SessionThreadEventListParams, opts ...option.RequestOption\) \(res \*pagination.PageCursor\[ManagedAgentsSessionEventUnion\], err error\)](<#SessionThreadEventService.List>) - [func \(r \*SessionThreadEventService\) ListAutoPaging\(ctx context.Context, threadID string, params SessionThreadEventListParams, opts ...option.RequestOption\) \*pagination.PageCursorAutoPager\[ManagedAgentsSessionEventUnion\]](<#SessionThreadEventService.ListAutoPaging>) + - [func \(r \*SessionThreadEventService\) NewResumableStream\(ctx context.Context, threadID string, params SessionThreadEventStreamParams, opts ...option.RequestOption\) \*ResumableSessionThreadEventStream](<#SessionThreadEventService.NewResumableStream>) - [func \(r \*SessionThreadEventService\) StreamEvents\(ctx context.Context, threadID string, params SessionThreadEventStreamParams, opts ...option.RequestOption\) \(stream \*ssestream.Stream\[ManagedAgentsStreamSessionThreadEventsUnion\]\)](<#SessionThreadEventService.StreamEvents>) - [type SessionThreadEventStreamParams](<#SessionThreadEventStreamParams>) - [func \(r SessionThreadEventStreamParams\) URLQuery\(\) \(v url.Values, err error\)](<#SessionThreadEventStreamParams.URLQuery>) @@ -54941,6 +55020,62 @@ func (s *ResumableSessionEventStream) Next() bool + +## type [ResumableSessionThreadEventStream]() + +ResumableSessionThreadEventStream reconnects a single thread's event stream. A thread/session termination or session deletion event ends the stream normally. + +```go +type ResumableSessionThreadEventStream struct { + // contains filtered or unexported fields +} +``` + + +### func \(\*ResumableSessionThreadEventStream\) [Close]() + +```go +func (s *ResumableSessionThreadEventStream) Close() error +``` + + + + +### func \(\*ResumableSessionThreadEventStream\) [Current]() + +```go +func (s *ResumableSessionThreadEventStream) Current() ManagedAgentsStreamSessionThreadEventsUnion +``` + + + + +### func \(\*ResumableSessionThreadEventStream\) [Err]() + +```go +func (s *ResumableSessionThreadEventStream) Err() error +``` + + + + +### func \(\*ResumableSessionThreadEventStream\) [LastEventID]() + +```go +func (s *ResumableSessionThreadEventStream) LastEventID() string +``` + + + + +### func \(\*ResumableSessionThreadEventStream\) [Next]() + +```go +func (s *ResumableSessionThreadEventStream) Next() bool +``` + + + ## type [SearchResultBlockParam]() @@ -57005,6 +57140,15 @@ func (r *SessionThreadEventService) ListAutoPaging(ctx context.Context, threadID List Session Thread Events + +### func \(\*SessionThreadEventService\) [NewResumableStream]() + +```go +func (r *SessionThreadEventService) NewResumableStream(ctx context.Context, threadID string, params SessionThreadEventStreamParams, opts ...option.RequestOption) *ResumableSessionThreadEventStream +``` + +NewResumableStream subscribes to Thread Events and reconnects retryable failures with the latest yielded Last\-Event\-ID cursor. Thread idle events do not stop it. + ### func \(\*SessionThreadEventService\) [StreamEvents]() @@ -57021,7 +57165,9 @@ Stream Session Thread Events ```go type SessionThreadEventStreamParams struct { - SessionID string `path:"session_id" api:"required" json:"-"` + // Resume after this public event in the thread. + LastEventID param.Opt[string] `header:"Last-Event-ID,omitzero" json:"-"` + SessionID string `path:"session_id" api:"required" json:"-"` // When set, this connection also receives streaming deltas (`event_start`, // `event_delta`) while an event is being produced, before the event itself // arrives. Deltas are best-effort; when the final event is produced it carries the diff --git a/forward/client_test.go b/forward/client_test.go index 83d8473..1e6b203 100644 --- a/forward/client_test.go +++ b/forward/client_test.go @@ -26,7 +26,7 @@ func TestForwardAPIInventory(t *testing.T) { for j := 0; j < methods.NumMethod(); j++ { name := methods.Method(j).Name key := field.Type().Name() + "." + name - if !strings.HasSuffix(name, "AutoPaging") && key != "SessionEventService.NewResumableStream" && key != "RealtimeService.Connect" { + if !strings.HasSuffix(name, "AutoPaging") && key != "SessionEventService.NewResumableStream" && key != "SessionThreadEventService.NewResumableStream" && key != "RealtimeService.Connect" { actual[key] = true } } diff --git a/forward/resumablestream.go b/forward/resumablestream.go index f3068b8..d4b84d7 100644 --- a/forward/resumablestream.go +++ b/forward/resumablestream.go @@ -45,3 +45,42 @@ func (s *ResumableSessionEventStream) Current() SessionEvent { return s.stream.C func (s *ResumableSessionEventStream) Err() error { return s.stream.Err() } func (s *ResumableSessionEventStream) LastEventID() string { return s.stream.LastEventID() } func (s *ResumableSessionEventStream) Close() error { return s.stream.Close() } + +// NewResumableStream subscribes to Thread Events and reconnects retryable failures +// with the latest yielded Last-Event-ID cursor. Thread idle events do not stop it. +func (r *SessionThreadEventService) NewResumableStream(ctx context.Context, sessionID string, threadID string, params SessionThreadEventStreamParams, opts ...option.RequestOption) *ResumableSessionThreadEventStream { + if sessionID == "" { + return &ResumableSessionThreadEventStream{stream: resumablestream.NewError[SessionEvent](ctx, fmt.Errorf("missing required session_id parameter"))} + } + if threadID == "" { + return &ResumableSessionThreadEventStream{stream: resumablestream.NewError[SessionEvent](ctx, fmt.Errorf("missing required thread_id parameter"))} + } + opts = slices.Clone(opts) + service := NewSessionThreadEventService(slices.Clone(r.Options)...) + lastEventID := "" + if params.LastEventID.Valid() { + lastEventID = params.LastEventID.Value + } + stream := resumablestream.New(ctx, lastEventID, func(ctx context.Context, cursor string, resume bool) resumablestream.Child[SessionEvent] { + requestOpts := slices.Clone(opts) + if resume { + requestOpts = append(requestOpts, option.WithHeader("Last-Event-ID", cursor)) + } + return service.StreamEvents(ctx, sessionID, threadID, params, requestOpts...) + }, func(event SessionEvent) bool { + return event.Type == "session.thread_status_terminated" || event.Type == "session.status_terminated" || event.Type == "session.deleted" + }) + return &ResumableSessionThreadEventStream{stream: stream} +} + +// ResumableSessionThreadEventStream reconnects a single thread's event stream. +// A thread/session termination or session deletion event ends the stream normally. +type ResumableSessionThreadEventStream struct { + stream *resumablestream.Stream[SessionEvent] +} + +func (s *ResumableSessionThreadEventStream) Next() bool { return s.stream.Next() } +func (s *ResumableSessionThreadEventStream) Current() SessionEvent { return s.stream.Current() } +func (s *ResumableSessionThreadEventStream) Err() error { return s.stream.Err() } +func (s *ResumableSessionThreadEventStream) LastEventID() string { return s.stream.LastEventID() } +func (s *ResumableSessionThreadEventStream) Close() error { return s.stream.Close() } diff --git a/forward/sessionthreadevent_resumable_test.go b/forward/sessionthreadevent_resumable_test.go new file mode 100644 index 0000000..d6a70cc --- /dev/null +++ b/forward/sessionthreadevent_resumable_test.go @@ -0,0 +1,156 @@ +package forward_test + +import ( + "context" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "reflect" + "strconv" + "sync" + "testing" + "time" + + "github.com/QoderAI/qoder-cloud-agents-sdk-go/convention/option" + "github.com/QoderAI/qoder-cloud-agents-sdk-go/forward" +) + +func TestForwardThreadResumableStreamReconnects(t *testing.T) { + for _, ending := range []string{"eof", "truncated"} { + t.Run(ending, func(t *testing.T) { + first := "id: checkpoint\nevent: session.thread_status_idle\ndata: {\"id\":\"payload-id\",\"type\":\"session.thread_status_idle\"}\n\n" + + "id: partial\ndata: {\"id\":\"not-delivered\"}" + last := "id: terminal\nevent: session.thread_status_terminated\ndata: {\"type\":\"session.thread_status_terminated\"}\n\n" + var mu sync.Mutex + var cursors []string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + defer mu.Unlock() + cursors = append(cursors, r.Header.Get("Last-Event-ID")) + if r.Method != http.MethodGet || r.URL.Path != "/sessions/sess/threads/thread/stream" || r.URL.RawQuery != "" { + t.Errorf("unexpected request: %s %s", r.Method, r.URL) + } + if r.Header.Get("X-Test") != "kept" || r.Header.Get("Accept") != "text/event-stream" { + t.Errorf("headers lost: %v", r.Header) + } + + w.Header().Set("Content-Type", "text/event-stream") + if len(cursors) == 1 { + if ending == "truncated" { + w.Header().Set("Content-Length", strconv.Itoa(len(first)+128)) + } + _, _ = io.WriteString(w, first) + } else { + _, _ = io.WriteString(w, last) + } + })) + defer server.Close() + client := forward.NewClient(option.WithPAT("test"), option.WithBaseURL(server.URL), option.WithHTTPClient(server.Client()), option.WithMaxRetries(0)) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + stream := client.Sessions.Threads.Events.NewResumableStream(ctx, "sess", "thread", forward.SessionThreadEventStreamParams{LastEventID: forward.String("initial")}, option.WithHeader("X-Test", "kept"), option.WithHeader("last-event-id", "override")) + defer stream.Close() + var types []string + for stream.Next() { + types = append(types, stream.Current().Type) + } + if stream.Err() != nil { + t.Fatal(stream.Err()) + } + if !reflect.DeepEqual(types, []string{"session.thread_status_idle", "session.thread_status_terminated"}) || stream.LastEventID() != "terminal" { + t.Fatalf("types=%v cursor=%s", types, stream.LastEventID()) + } + mu.Lock() + defer mu.Unlock() + if !reflect.DeepEqual(cursors, []string{"override", "checkpoint"}) { + t.Fatalf("cursors=%v", cursors) + } + }) + } +} + +func TestForwardThreadResumableStreamHTTPPolicy(t *testing.T) { + for _, status := range []int{400, 401, 403, 404, 409, 429, 503} { + t.Run(strconv.Itoa(status), func(t *testing.T) { + calls := 0 + client := testClient(func(r *http.Request) (*http.Response, error) { + calls++ + if r.Header.Get("Last-Event-ID") != "initial" { + t.Errorf("cursor lost: %v", r.Header) + } + if calls == 1 { + return reply(r, status, `{"error":{"type":"invalid_request_error","message":"test failure"}}`), nil + } + return reply(r, 200, "event: session.thread_status_terminated\ndata: {\"type\":\"session.thread_status_terminated\"}\n\n"), nil + }, option.WithMaxRetries(0)) + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + stream := client.Sessions.Threads.Events.NewResumableStream(ctx, "sess", "thread", forward.SessionThreadEventStreamParams{LastEventID: forward.String("initial")}) + defer stream.Close() + got := stream.Next() + if status == 429 || status == 503 { + if !got || stream.Next() || stream.Err() != nil || calls != 2 { + t.Fatalf("retry failed: calls=%d err=%v", calls, stream.Err()) + } + } else if got || stream.Err() == nil || calls != 1 { + t.Fatalf("non-retryable request repeated: calls=%d err=%v", calls, stream.Err()) + } + }) + } +} + +func TestForwardThreadResumableStreamCancellation(t *testing.T) { + for _, closeStream := range []bool{false, true} { + t.Run(fmt.Sprint(closeStream), func(t *testing.T) { + started := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.(http.Flusher).Flush() + close(started) + <-r.Context().Done() + })) + defer server.Close() + client := forward.NewClient(option.WithPAT("test"), option.WithBaseURL(server.URL), option.WithHTTPClient(server.Client())) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + stream := client.Sessions.Threads.Events.NewResumableStream(ctx, "sess", "thread", forward.SessionThreadEventStreamParams{LastEventID: forward.String("initial")}) + defer stream.Close() + done := make(chan bool, 1) + go func() { done <- stream.Next() }() + select { + case <-started: + case <-time.After(3 * time.Second): + t.Fatal("request did not start") + } + if closeStream { + _ = stream.Close() + } else { + cancel() + } + select { + case got := <-done: + if got { + t.Fatal("yielded after cancellation") + } + case <-time.After(3 * time.Second): + t.Fatal("read not interrupted") + } + if !closeStream && !errors.Is(stream.Err(), context.Canceled) { + t.Fatalf("error=%v", stream.Err()) + } + }) + } +} + +func TestForwardThreadResumableStreamMissingIDs(t *testing.T) { + client := forward.NewClient(option.WithPAT("test")) + for _, ids := range [][2]string{{"", "thread"}, {"sess", ""}} { + stream := client.Sessions.Threads.Events.NewResumableStream(context.Background(), ids[0], ids[1], forward.SessionThreadEventStreamParams{}) + if stream.Next() || stream.Err() == nil { + t.Fatal("missing IDs accepted") + } + _ = stream.Close() + } +} diff --git a/managed/client_test.go b/managed/client_test.go index 9bb1051..47c5111 100644 --- a/managed/client_test.go +++ b/managed/client_test.go @@ -33,7 +33,7 @@ func TestManagedAPIInventory(t *testing.T) { for i := 0; i < tp.NumMethod(); i++ { m := tp.Method(i) key := name + "." + m.Name - if strings.HasSuffix(m.Name, "AutoPaging") || key == "SessionEventService.NewResumableStream" { + if strings.HasSuffix(m.Name, "AutoPaging") || (key == "SessionEventService.NewResumableStream" || key == "SessionThreadEventService.NewResumableStream") { continue } if !expected[key] { diff --git a/managed/resumablestream.go b/managed/resumablestream.go index b2f0099..5178f54 100644 --- a/managed/resumablestream.go +++ b/managed/resumablestream.go @@ -48,3 +48,46 @@ func (s *ResumableSessionEventStream) Current() ManagedAgentsStreamSessionEvents func (s *ResumableSessionEventStream) Err() error { return s.stream.Err() } func (s *ResumableSessionEventStream) LastEventID() string { return s.stream.LastEventID() } func (s *ResumableSessionEventStream) Close() error { return s.stream.Close() } + +// NewResumableStream subscribes to Thread Events and reconnects retryable failures +// with the latest yielded Last-Event-ID cursor. Thread idle events do not stop it. +func (r *SessionThreadEventService) NewResumableStream(ctx context.Context, threadID string, params SessionThreadEventStreamParams, opts ...option.RequestOption) *ResumableSessionThreadEventStream { + if params.SessionID == "" { + return &ResumableSessionThreadEventStream{stream: resumablestream.NewError[ManagedAgentsStreamSessionThreadEventsUnion](ctx, errors.New("missing required session_id parameter"))} + } + if threadID == "" { + return &ResumableSessionThreadEventStream{stream: resumablestream.NewError[ManagedAgentsStreamSessionThreadEventsUnion](ctx, errors.New("missing required thread_id parameter"))} + } + params.EventDeltas = slices.Clone(params.EventDeltas) + params.Betas = slices.Clone(params.Betas) + opts = slices.Clone(opts) + service := NewSessionThreadEventService(slices.Clone(r.Options)...) + lastEventID := "" + if params.LastEventID.Valid() { + lastEventID = params.LastEventID.Value + } + stream := resumablestream.New(ctx, lastEventID, func(ctx context.Context, cursor string, resume bool) resumablestream.Child[ManagedAgentsStreamSessionThreadEventsUnion] { + requestOpts := slices.Clone(opts) + if resume { + requestOpts = append(requestOpts, option.WithHeader("Last-Event-ID", cursor)) + } + return service.StreamEvents(ctx, threadID, params, requestOpts...) + }, func(event ManagedAgentsStreamSessionThreadEventsUnion) bool { + return event.Type == "session.thread_status_terminated" || event.Type == "session.status_terminated" || event.Type == "session.deleted" + }) + return &ResumableSessionThreadEventStream{stream: stream} +} + +// ResumableSessionThreadEventStream reconnects a single thread's event stream. +// A thread/session termination or session deletion event ends the stream normally. +type ResumableSessionThreadEventStream struct { + stream *resumablestream.Stream[ManagedAgentsStreamSessionThreadEventsUnion] +} + +func (s *ResumableSessionThreadEventStream) Next() bool { return s.stream.Next() } +func (s *ResumableSessionThreadEventStream) Current() ManagedAgentsStreamSessionThreadEventsUnion { + return s.stream.Current() +} +func (s *ResumableSessionThreadEventStream) Err() error { return s.stream.Err() } +func (s *ResumableSessionThreadEventStream) LastEventID() string { return s.stream.LastEventID() } +func (s *ResumableSessionThreadEventStream) Close() error { return s.stream.Close() } diff --git a/managed/sessionthreadevent.go b/managed/sessionthreadevent.go index 758e29f..6cc86cb 100644 --- a/managed/sessionthreadevent.go +++ b/managed/sessionthreadevent.go @@ -76,6 +76,9 @@ func (r *SessionThreadEventService) StreamEvents(ctx context.Context, threadID s raw *http.Response err error ) + if params.LastEventID.Valid() { + opts = append([]option.RequestOption{option.WithHeader("Last-Event-ID", params.LastEventID.Value)}, opts...) + } for _, v := range params.Betas { opts = append(opts, option.WithHeaderAdd("x-qoder-beta", fmt.Sprintf("%v", v))) } @@ -118,7 +121,9 @@ func (r SessionThreadEventListParams) URLQuery() (v url.Values, err error) { } type SessionThreadEventStreamParams struct { - SessionID string `path:"session_id" api:"required" json:"-"` + // Resume after this public event in the thread. + LastEventID param.Opt[string] `header:"Last-Event-ID,omitzero" json:"-"` + SessionID string `path:"session_id" api:"required" json:"-"` // When set, this connection also receives streaming deltas (`event_start`, // `event_delta`) while an event is being produced, before the event itself // arrives. Deltas are best-effort; when the final event is produced it carries the diff --git a/managed/sessionthreadevent_resumable_test.go b/managed/sessionthreadevent_resumable_test.go new file mode 100644 index 0000000..7db67f3 --- /dev/null +++ b/managed/sessionthreadevent_resumable_test.go @@ -0,0 +1,158 @@ +package managed_test + +import ( + "context" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "reflect" + "strconv" + "sync" + "testing" + "time" + + "github.com/QoderAI/qoder-cloud-agents-sdk-go/convention/option" + "github.com/QoderAI/qoder-cloud-agents-sdk-go/managed" +) + +func TestManagedThreadResumableStreamReconnects(t *testing.T) { + for _, ending := range []string{"eof", "truncated"} { + t.Run(ending, func(t *testing.T) { + first := "id: checkpoint\nevent: session.thread_status_idle\ndata: {\"id\":\"payload-id\",\"type\":\"session.thread_status_idle\"}\n\n" + + "id: partial\ndata: {\"id\":\"not-delivered\"}" + last := "id: terminal\nevent: session.thread_status_terminated\ndata: {\"type\":\"session.thread_status_terminated\"}\n\n" + var mu sync.Mutex + var cursors []string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + defer mu.Unlock() + cursors = append(cursors, r.Header.Get("Last-Event-ID")) + if r.Method != http.MethodGet || r.URL.Path != "/sessions/sess/threads/thread/stream" || r.URL.RawQuery != "" { + t.Errorf("unexpected request: %s %s", r.Method, r.URL) + } + if r.Header.Get("X-Test") != "kept" || r.Header.Get("Accept") != "text/event-stream" { + t.Errorf("headers lost: %v", r.Header) + } + if r.Header.Get("x-qoder-beta") != "beta-one" { + t.Errorf("beta lost") + } + w.Header().Set("Content-Type", "text/event-stream") + if len(cursors) == 1 { + if ending == "truncated" { + w.Header().Set("Content-Length", strconv.Itoa(len(first)+128)) + } + _, _ = io.WriteString(w, first) + } else { + _, _ = io.WriteString(w, last) + } + })) + defer server.Close() + client := managed.NewClient(option.WithPAT("test"), option.WithBaseURL(server.URL), option.WithHTTPClient(server.Client()), option.WithMaxRetries(0)) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + stream := client.Sessions.Threads.Events.NewResumableStream(ctx, "thread", managed.SessionThreadEventStreamParams{SessionID: "sess", LastEventID: managed.String("initial"), Betas: []managed.QoderBeta{"beta-one"}}, option.WithHeader("X-Test", "kept"), option.WithHeader("last-event-id", "override")) + defer stream.Close() + var types []string + for stream.Next() { + types = append(types, stream.Current().Type) + } + if stream.Err() != nil { + t.Fatal(stream.Err()) + } + if !reflect.DeepEqual(types, []string{"session.thread_status_idle", "session.thread_status_terminated"}) || stream.LastEventID() != "terminal" { + t.Fatalf("types=%v cursor=%s", types, stream.LastEventID()) + } + mu.Lock() + defer mu.Unlock() + if !reflect.DeepEqual(cursors, []string{"override", "checkpoint"}) { + t.Fatalf("cursors=%v", cursors) + } + }) + } +} + +func TestManagedThreadResumableStreamHTTPPolicy(t *testing.T) { + for _, status := range []int{400, 401, 403, 404, 409, 429, 503} { + t.Run(strconv.Itoa(status), func(t *testing.T) { + calls := 0 + client := testClient(func(r *http.Request) (*http.Response, error) { + calls++ + if r.Header.Get("Last-Event-ID") != "initial" { + t.Errorf("cursor lost: %v", r.Header) + } + if calls == 1 { + return reply(r, status, `{"error":{"type":"invalid_request_error","message":"test failure"}}`), nil + } + return reply(r, 200, "event: session.thread_status_terminated\ndata: {\"type\":\"session.thread_status_terminated\"}\n\n"), nil + }, option.WithMaxRetries(0)) + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + stream := client.Sessions.Threads.Events.NewResumableStream(ctx, "thread", managed.SessionThreadEventStreamParams{SessionID: "sess", LastEventID: managed.String("initial"), Betas: []managed.QoderBeta{"beta-one"}}) + defer stream.Close() + got := stream.Next() + if status == 429 || status == 503 { + if !got || stream.Next() || stream.Err() != nil || calls != 2 { + t.Fatalf("retry failed: calls=%d err=%v", calls, stream.Err()) + } + } else if got || stream.Err() == nil || calls != 1 { + t.Fatalf("non-retryable request repeated: calls=%d err=%v", calls, stream.Err()) + } + }) + } +} + +func TestManagedThreadResumableStreamCancellation(t *testing.T) { + for _, closeStream := range []bool{false, true} { + t.Run(fmt.Sprint(closeStream), func(t *testing.T) { + started := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.(http.Flusher).Flush() + close(started) + <-r.Context().Done() + })) + defer server.Close() + client := managed.NewClient(option.WithPAT("test"), option.WithBaseURL(server.URL), option.WithHTTPClient(server.Client())) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + stream := client.Sessions.Threads.Events.NewResumableStream(ctx, "thread", managed.SessionThreadEventStreamParams{SessionID: "sess", LastEventID: managed.String("initial"), Betas: []managed.QoderBeta{"beta-one"}}) + defer stream.Close() + done := make(chan bool, 1) + go func() { done <- stream.Next() }() + select { + case <-started: + case <-time.After(3 * time.Second): + t.Fatal("request did not start") + } + if closeStream { + _ = stream.Close() + } else { + cancel() + } + select { + case got := <-done: + if got { + t.Fatal("yielded after cancellation") + } + case <-time.After(3 * time.Second): + t.Fatal("read not interrupted") + } + if !closeStream && !errors.Is(stream.Err(), context.Canceled) { + t.Fatalf("error=%v", stream.Err()) + } + }) + } +} + +func TestManagedThreadResumableStreamMissingIDs(t *testing.T) { + client := managed.NewClient(option.WithPAT("test")) + for _, ids := range [][2]string{{"", "thread"}, {"sess", ""}} { + stream := client.Sessions.Threads.Events.NewResumableStream(context.Background(), ids[1], managed.SessionThreadEventStreamParams{SessionID: ids[0]}) + if stream.Next() || stream.Err() == nil { + t.Fatal("missing IDs accepted") + } + _ = stream.Close() + } +}