feat(streaming): add automatic Thread SSE resumption

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

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.

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
 
 
 
+<a name="ResumableSessionThreadEventStream"></a>
+## type [ResumableSessionThreadEventStream](<https://github.com/QoderAI/qoder-cloud-agents-sdk-go/blob/main/forward/resumablestream.go>)
+
+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
+}
+```
+
+<a name="ResumableSessionThreadEventStream.Close"></a>
+### func \(\*ResumableSessionThreadEventStream\) [Close](<https://github.com/QoderAI/qoder-cloud-agents-sdk-go/blob/main/forward/resumablestream.go>)
+
+```go
+func (s *ResumableSessionThreadEventStream) Close() error
+```
+
+
+
+<a name="ResumableSessionThreadEventStream.Current"></a>
+### func \(\*ResumableSessionThreadEventStream\) [Current](<https://github.com/QoderAI/qoder-cloud-agents-sdk-go/blob/main/forward/resumablestream.go>)
+
+```go
+func (s *ResumableSessionThreadEventStream) Current() SessionEvent
+```
+
+
+
+<a name="ResumableSessionThreadEventStream.Err"></a>
+### func \(\*ResumableSessionThreadEventStream\) [Err](<https://github.com/QoderAI/qoder-cloud-agents-sdk-go/blob/main/forward/resumablestream.go>)
+
+```go
+func (s *ResumableSessionThreadEventStream) Err() error
+```
+
+
+
+<a name="ResumableSessionThreadEventStream.LastEventID"></a>
+### func \(\*ResumableSessionThreadEventStream\) [LastEventID](<https://github.com/QoderAI/qoder-cloud-agents-sdk-go/blob/main/forward/resumablestream.go>)
+
+```go
+func (s *ResumableSessionThreadEventStream) LastEventID() string
+```
+
+
+
+<a name="ResumableSessionThreadEventStream.Next"></a>
+### func \(\*ResumableSessionThreadEventStream\) [Next](<https://github.com/QoderAI/qoder-cloud-agents-sdk-go/blob/main/forward/resumablestream.go>)
+
+```go
+func (s *ResumableSessionThreadEventStream) Next() bool
+```
+
+
+
 <a name="Schedule"></a>
 ## type [Schedule](<https://github.com/QoderAI/qoder-cloud-agents-sdk-go/blob/main/forward/schedule.go>)
 
@@ -11823,6 +11886,15 @@ func (r *SessionThreadEventService) ListAutoPaging(ctx context.Context, sessionI
 
 
 
+<a name="SessionThreadEventService.NewResumableStream"></a>
+### func \(\*SessionThreadEventService\) [NewResumableStream](<https://github.com/QoderAI/qoder-cloud-agents-sdk-go/blob/main/forward/resumablestream.go>)
+
+```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.
+
 <a name="SessionThreadEventService.StreamEvents"></a>
 ### func \(\*SessionThreadEventService\) [StreamEvents](<https://github.com/QoderAI/qoder-cloud-agents-sdk-go/blob/main/forward/sessionthreadevent.go>)
 
@@ -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
 
 
 
+<a name="ResumableSessionThreadEventStream"></a>
+## type [ResumableSessionThreadEventStream](<https://github.com/QoderAI/qoder-cloud-agents-sdk-go/blob/main/managed/resumablestream.go>)
+
+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
+}
+```
+
+<a name="ResumableSessionThreadEventStream.Close"></a>
+### func \(\*ResumableSessionThreadEventStream\) [Close](<https://github.com/QoderAI/qoder-cloud-agents-sdk-go/blob/main/managed/resumablestream.go>)
+
+```go
+func (s *ResumableSessionThreadEventStream) Close() error
+```
+
+
+
+<a name="ResumableSessionThreadEventStream.Current"></a>
+### func \(\*ResumableSessionThreadEventStream\) [Current](<https://github.com/QoderAI/qoder-cloud-agents-sdk-go/blob/main/managed/resumablestream.go>)
+
+```go
+func (s *ResumableSessionThreadEventStream) Current() ManagedAgentsStreamSessionThreadEventsUnion
+```
+
+
+
+<a name="ResumableSessionThreadEventStream.Err"></a>
+### func \(\*ResumableSessionThreadEventStream\) [Err](<https://github.com/QoderAI/qoder-cloud-agents-sdk-go/blob/main/managed/resumablestream.go>)
+
+```go
+func (s *ResumableSessionThreadEventStream) Err() error
+```
+
+
+
+<a name="ResumableSessionThreadEventStream.LastEventID"></a>
+### func \(\*ResumableSessionThreadEventStream\) [LastEventID](<https://github.com/QoderAI/qoder-cloud-agents-sdk-go/blob/main/managed/resumablestream.go>)
+
+```go
+func (s *ResumableSessionThreadEventStream) LastEventID() string
+```
+
+
+
+<a name="ResumableSessionThreadEventStream.Next"></a>
+### func \(\*ResumableSessionThreadEventStream\) [Next](<https://github.com/QoderAI/qoder-cloud-agents-sdk-go/blob/main/managed/resumablestream.go>)
+
+```go
+func (s *ResumableSessionThreadEventStream) Next() bool
+```
+
+
+
 <a name="SearchResultBlockParam"></a>
 ## type [SearchResultBlockParam](<https://github.com/QoderAI/qoder-cloud-agents-sdk-go/blob/main/managed/message.go>)
 
@@ -57005,6 +57140,15 @@ func (r *SessionThreadEventService) ListAutoPaging(ctx context.Context, threadID
 
 List Session Thread Events
 
+<a name="SessionThreadEventService.NewResumableStream"></a>
+### func \(\*SessionThreadEventService\) [NewResumableStream](<https://github.com/QoderAI/qoder-cloud-agents-sdk-go/blob/main/managed/resumablestream.go>)
+
+```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.
+
 <a name="SessionThreadEventService.StreamEvents"></a>
 ### func \(\*SessionThreadEventService\) [StreamEvents](<https://github.com/QoderAI/qoder-cloud-agents-sdk-go/blob/main/managed/sessionthreadevent.go>)
 
@@ -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

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
 				}
 			}

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() }