diff --git a/CHANGELOG.md b/CHANGELOG.md index ea6bfc9..bdd6465 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 42b18fd..aab6c5a 100644 --- a/README.md +++ b/README.md @@ -344,6 +344,49 @@ for await (const event of stream) { The resumable stream does not query event history, clear an invalid cursor, or deduplicate IDs: delta frames may legitimately share an event ID. It stops after `[DONE]`, `session.status_terminated`, or `session.deleted`. Stop it early with `stream.controller.abort()`, an `AbortSignal` in `RequestOptions`, or `await stream.close()`. +### 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: + +```ts +import { ForwardClient, ManagedClient } from 'qca-sdk'; + +const forwardClient = new ForwardClient(); +const managedClient = new ManagedClient(); +const controller = new AbortController(); +// Replace these with IDs returned by the API. +const sessionID = 'sess_...'; +const threadID = 'sthr_...'; +const afterID = 'evt_...'; + +// Forward +const stream = forwardClient.sessions.threads.events.resumableStream( + sessionID, threadID, { last_event_id: afterID }, { signal: controller.signal }, +); +try { + for await (const event of stream) console.log(event.type); +} finally { + await stream.close(); +} + +// Managed uses the parent session in params. +const managedStream = managedClient.sessions.threads.events.resumableStream( + threadID, { session_id: sessionID, last_event_id: afterID }, +); +// Iterate as above and close when finished. +``` + +`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. + ## Forward Decisions (Beta) The Decision API is in Beta. Its behavior may change in future updates. diff --git a/docs/api/forward/classes/SessionThreadEvents.md b/docs/api/forward/classes/SessionThreadEvents.md index 9cb7a09..1fe86c8 100644 --- a/docs/api/forward/classes/SessionThreadEvents.md +++ b/docs/api/forward/classes/SessionThreadEvents.md @@ -76,6 +76,36 @@ listSessionThreadEvents *** +### resumableStream() + +> **resumableStream**(`sessionID`, `threadID`, `params?`, `options?`): [`ResumableSessionEventStream`](../../index/classes/ResumableSessionEventStream.md)\<[`SessionEvent`](../interfaces/SessionEvent.md)\> + +Reconnect the thread stream from the last received SSE cursor. + +#### Parameters + +##### sessionID + +`string` + +##### threadID + +`string` + +##### params? + +[`SessionThreadEventStreamParams`](../interfaces/SessionThreadEventStreamParams.md) = `{}` + +##### options? + +[`RequestOptions`](../../index/interfaces/RequestOptions.md) + +#### Returns + +[`ResumableSessionEventStream`](../../index/classes/ResumableSessionEventStream.md)\<[`SessionEvent`](../interfaces/SessionEvent.md)\> + +*** + ### streamEvents() > **streamEvents**(`sessionID`, `threadID`, `params?`, `options?`): [`APIPromise`](../../index/classes/APIPromise.md)\<[`Stream`](../../index/classes/Stream.md)\<[`SessionEvent`](../interfaces/SessionEvent.md)\>\> diff --git a/docs/api/index/classes/ResumableSessionEventStream.md b/docs/api/index/classes/ResumableSessionEventStream.md index 40a73f9..ece8f8b 100644 --- a/docs/api/index/classes/ResumableSessionEventStream.md +++ b/docs/api/index/classes/ResumableSessionEventStream.md @@ -6,7 +6,7 @@ # Class: ResumableSessionEventStream\ -A stateful, single-consumer Session Events stream that reconnects after retryable +A stateful, single-consumer Session or Thread Events stream that reconnects after retryable failures and resumes from the last fully decoded SSE frame. ## Type Parameters @@ -23,7 +23,7 @@ failures and resumes from the last fully decoded SSE frame. ### Constructor -> **new ResumableSessionEventStream**\<`T`\>(`openStream`, `initialLastEventID?`, `signal?`): `ResumableSessionEventStream`\<`T`\> +> **new ResumableSessionEventStream**\<`T`\>(`openStream`, `initialLastEventID?`, `signal?`, `isTerminal?`): `ResumableSessionEventStream`\<`T`\> #### Parameters @@ -39,6 +39,10 @@ failures and resumes from the last fully decoded SSE frame. `AbortSignal` \| `null` +##### isTerminal? + +(`event`) => `boolean` + #### Returns `ResumableSessionEventStream`\<`T`\> diff --git a/docs/api/managed/classes/SessionsThreadsEvents.md b/docs/api/managed/classes/SessionsThreadsEvents.md index 5dd9a3e..dcc36cc 100644 --- a/docs/api/managed/classes/SessionsThreadsEvents.md +++ b/docs/api/managed/classes/SessionsThreadsEvents.md @@ -68,6 +68,32 @@ List Session Thread Events *** +### resumableStream() + +> **resumableStream**(`threadID`, `params`, `options?`): [`ResumableSessionEventStream`](../../index/classes/ResumableSessionEventStream.md)\<[`ManagedAgentsStreamSessionThreadEventsUnion`](../type-aliases/ManagedAgentsStreamSessionThreadEventsUnion.md)\> + +Reconnect the thread stream from the last received SSE cursor. + +#### Parameters + +##### threadID + +`string` + +##### params + +[`SessionThreadEventStreamParams`](../interfaces/SessionThreadEventStreamParams.md) + +##### options? + +[`RequestOptions`](../../index/interfaces/RequestOptions.md) + +#### Returns + +[`ResumableSessionEventStream`](../../index/classes/ResumableSessionEventStream.md)\<[`ManagedAgentsStreamSessionThreadEventsUnion`](../type-aliases/ManagedAgentsStreamSessionThreadEventsUnion.md)\> + +*** + ### streamEvents() > **streamEvents**(`threadID`, `params`, `options?`): [`APIPromise`](../../index/classes/APIPromise.md)\<[`Stream`](../../index/classes/Stream.md)\<[`ManagedAgentsStreamSessionThreadEventsUnion`](../type-aliases/ManagedAgentsStreamSessionThreadEventsUnion.md)\>\> diff --git a/docs/api/managed/interfaces/SessionThreadEventStreamParams.md b/docs/api/managed/interfaces/SessionThreadEventStreamParams.md index cfe47dd..d57a8ad 100644 --- a/docs/api/managed/interfaces/SessionThreadEventStreamParams.md +++ b/docs/api/managed/interfaces/SessionThreadEventStreamParams.md @@ -33,6 +33,14 @@ are sent. *** +### last\_event\_id? + +> `optional` **last\_event\_id?**: `string` \| `null` + +Resume after this public event in the thread. + +*** + ### session\_id > **session\_id**: `string` diff --git a/src/core/resumable-session-event-stream.ts b/src/core/resumable-session-event-stream.ts index b4ced0d..b7e412e 100644 --- a/src/core/resumable-session-event-stream.ts +++ b/src/core/resumable-session-event-stream.ts @@ -62,6 +62,12 @@ function isTerminalSessionEvent(value: unknown): boolean { return type === 'session.status_terminated' || type === 'session.deleted'; } +/** @internal Thread idle is not terminal; only termination ends the subscription. */ +export function isTerminalThreadEvent(value: unknown): boolean { + return isTerminalSessionEvent(value) || (!!value && typeof value === 'object' && + (value as { type?: unknown }).type === 'session.thread_status_terminated'); +} + async function waitForRetry(ms: number, signal: AbortSignal): Promise { if (signal.aborted) throw aborted(signal); let timer: ReturnType | undefined; @@ -79,7 +85,7 @@ async function waitForRetry(ms: number, signal: AbortSignal): Promise { } /** - * A stateful, single-consumer Session Events stream that reconnects after retryable + * A stateful, single-consumer Session or Thread Events stream that reconnects after retryable * failures and resumes from the last fully decoded SSE frame. */ export class ResumableSessionEventStream implements AsyncIterable { @@ -99,6 +105,7 @@ export class ResumableSessionEventStream implements AsyncIterable { private readonly openStream: StreamFactory, initialLastEventID?: string | null, signal?: AbortSignal | null, + private readonly isTerminal: (event: T) => boolean = isTerminalSessionEvent, ) { this.lastEventID = initialLastEventID ?? undefined; this.callerSignal = signal ?? undefined; @@ -155,7 +162,7 @@ export class ResumableSessionEventStream implements AsyncIterable { } if (this.closed) return; attempt = 0; - const terminal = isTerminalSessionEvent(result.value); + const terminal = this.isTerminal(result.value); yield result.value; if (terminal) return; } diff --git a/src/forward/session-thread-event.ts b/src/forward/session-thread-event.ts index a56bc69..a7ca7c4 100644 --- a/src/forward/session-thread-event.ts +++ b/src/forward/session-thread-event.ts @@ -3,6 +3,7 @@ import { APIResource } from '../core/resource.js'; import type { RequestOptions } from '../core/client.js'; import type { APIPromise } from '../core/api-promise.js'; import type { PagePromise } from '../core/pagination.js'; +import { initialLastEventID, resumableRequestHeaders, ResumableSessionEventStream, isTerminalThreadEvent } from '../core/resumable-session-event-stream.js'; import type { Stream } from '../core/streaming.js'; import { pathParam, requestHeaders } from './_utils.js'; import type { SessionEvent, SessionThreadEventListParams, SessionThreadEventStreamParams } from './types.js'; @@ -45,6 +46,30 @@ export class SessionThreadEvents extends APIResource { }); } + /** Reconnect the thread stream from the last received SSE cursor. */ + resumableStream( + sessionID: string, + threadID: string, + params: SessionThreadEventStreamParams = {}, + options?: RequestOptions, + ): ResumableSessionEventStream { + pathParam(sessionID, "session_id"); + pathParam(threadID, "thread_id"); + const values = { ...params }; + const initialCursor = initialLastEventID(values.last_event_id, this._client.defaultHeaders, options?.headers); + const requestOptions = { ...options, headers: resumableRequestHeaders(options?.headers, initialCursor) }; + return new ResumableSessionEventStream( + (lastEventID, signal) => this.streamEvents( + sessionID, + threadID, + { ...values, last_event_id: lastEventID }, + { ...requestOptions, signal, headers: resumableRequestHeaders(requestOptions.headers, lastEventID) }, + ), + initialCursor, + options?.signal, + isTerminalThreadEvent, + ); + } } export type { SessionEvent, SessionThreadEventListParams, SessionThreadEventStreamParams } from './types.js'; diff --git a/src/managed/resources/session-thread-event.ts b/src/managed/resources/session-thread-event.ts index 6760da4..8d3757c 100644 --- a/src/managed/resources/session-thread-event.ts +++ b/src/managed/resources/session-thread-event.ts @@ -3,6 +3,7 @@ import { APIResource } from '../../core/resource.js'; import type { RequestOptions } from '../../core/client.js'; import type { APIPromise } from '../../core/api-promise.js'; import type { PagePromise } from '../../core/pagination.js'; +import { initialLastEventID, resumableRequestHeaders, ResumableSessionEventStream, isTerminalThreadEvent } from '../../core/resumable-session-event-stream.js'; import type { Stream } from '../../core/streaming.js'; import type * as Types from '../types.js'; import { splitParams, pathParam, managedMultipart } from '../internal.js'; @@ -21,10 +22,33 @@ export class SessionsThreadsEvents extends APIResource { * Stream Session Thread Events */ streamEvents(threadID: string, params: Types.SessionThreadEventStreamParams, options?: RequestOptions): APIPromise> { - const request = splitParams(params, options, {"workspace_id": "qoder-workspace-id", "betas": "x-qoder-beta"}, ["session_id"]); + const request = splitParams(params, options, {"workspace_id": "qoder-workspace-id", "betas": "x-qoder-beta", "last_event_id": "Last-Event-ID"}, ["session_id"]); if (!request.headers.has('Accept')) request.headers.set('Accept', 'text/event-stream'); return this._client.request>({ ...request.options, method: "GET", path: `/sessions/${pathParam(params?.session_id, "session_id")}/threads/${pathParam(threadID, "thread_id")}/stream`, query: request.values, responseType: 'stream' }); } + + /** Reconnect the thread stream from the last received SSE cursor. */ + resumableStream( + threadID: string, + params: Types.SessionThreadEventStreamParams, + options?: RequestOptions, + ): ResumableSessionEventStream { + pathParam(params?.session_id, "session_id"); + pathParam(threadID, "thread_id"); + const values = { ...params, betas: params.betas?.slice(), event_deltas: params.event_deltas?.slice() }; + const initialCursor = initialLastEventID(values.last_event_id, this._client.defaultHeaders, options?.headers); + const requestOptions = { ...options, headers: resumableRequestHeaders(options?.headers, initialCursor) }; + return new ResumableSessionEventStream( + (lastEventID, signal) => this.streamEvents( + threadID, + { ...values, last_event_id: lastEventID }, + { ...requestOptions, signal, headers: resumableRequestHeaders(requestOptions.headers, lastEventID) }, + ), + initialCursor, + options?.signal, + isTerminalThreadEvent, + ); + } } export default SessionsThreadsEvents; diff --git a/src/managed/types.ts b/src/managed/types.ts index 9622ceb..665216f 100644 --- a/src/managed/types.ts +++ b/src/managed/types.ts @@ -10292,6 +10292,8 @@ export interface SessionThreadEventListParams { } export interface SessionThreadEventStreamParams { + /** Resume after this public event in the thread. */ + last_event_id?: string | null; session_id: string; workspace_id?: string; /** diff --git a/tests/api-contracts.test.mjs b/tests/api-contracts.test.mjs index 09465db..974dd06 100644 --- a/tests/api-contracts.test.mjs +++ b/tests/api-contracts.test.mjs @@ -4,8 +4,8 @@ import { test } from 'node:test'; import { sdk, methods, contracts, operations, lookup, argumentsFor, invoke, resource, response, testClient, assertRequest } from './helpers.mjs'; const handwrittenMethods = { - forward: ['sessions.events.resumableStream', 'realtime.connect'], - managed: ['sessions.events.resumableStream'], + forward: ['sessions.events.resumableStream', 'sessions.threads.events.resumableStream', 'realtime.connect'], + managed: ['sessions.events.resumableStream', 'sessions.threads.events.resumableStream'], }; for (const mode of ['forward', 'managed']) { diff --git a/tests/resumable-streaming.test.mjs b/tests/resumable-streaming.test.mjs index cbbea77..fa30e95 100644 --- a/tests/resumable-streaming.test.mjs +++ b/tests/resumable-streaming.test.mjs @@ -22,8 +22,8 @@ async function closeServer(server, sockets) { await closed; } -for (const mode of ['forward', 'managed']) { - test(`${mode}: public resumable stream reconnects after a truncated real HTTP response`, async () => { +for (const mode of ['forward', 'managed']) for (const scope of ['session', 'thread']) { + test(`${mode} ${scope}: public resumable stream reconnects after a truncated real HTTP response`, async () => { const requests = []; const sockets = new Set(); const events = [ @@ -67,7 +67,9 @@ for (const mode of ['forward', 'managed']) { maxRetries: 0, timeout: 5_000, }); - const stream = client.sessions.events.resumableStream('sess-one'); + const stream = scope === 'session' ? client.sessions.events.resumableStream('sess-one') + : mode === 'forward' ? client.sessions.threads.events.resumableStream('sess-one', 'thread-one') + : client.sessions.threads.events.resumableStream('thread-one', { session_id: 'sess-one' }); const received = []; for await (const event of stream) received.push(event); diff --git a/tests/thread-resumable-streaming.test.mjs b/tests/thread-resumable-streaming.test.mjs new file mode 100644 index 0000000..81e0380 --- /dev/null +++ b/tests/thread-resumable-streaming.test.mjs @@ -0,0 +1,128 @@ +import assert from 'node:assert/strict'; +import { test } from 'node:test'; +import { sdk, testClient } from './helpers.mjs'; + +const encoder = new TextEncoder(); +const frame = (id, type) => `id: ${id}\nevent: ${type}\ndata: ${JSON.stringify({ id: 'payload-id', type })}\n\n`; +function subscribe(client, mode, params = {}, options = {}) { + return mode === 'forward' + ? client.sessions.threads.events.resumableStream('sess-one', 'thread-one', params, options) + : client.sessions.threads.events.resumableStream('thread-one', { session_id: 'sess-one', ...params }, options); +} +async function retry(t) { + await new Promise(resolve => setImmediate(resolve)); + t.mock.timers.tick(500); +} + +for (const mode of ['forward', 'managed']) { + for (const ending of ['eof', 'transport']) { + test(`${mode} thread: ${ending} resumes with scoped route, cursor and options`, async t => { + t.mock.timers.enable({ apis: ['setTimeout'] }); + const requests = []; + const client = testClient(mode, req => { + requests.push(req); + if (requests.length > 1) return new Response(frame('terminal', 'session.thread_status_terminated')); + const content = frame('checkpoint', 'session.thread_status_idle') + 'id: partial\ndata: {"id":"not-yielded"}'; + let sent = false; + return new Response(new ReadableStream({ + pull(controller) { + if (!sent) { sent = true; controller.enqueue(encoder.encode(content)); } + else if (ending === 'transport') controller.error(new Error('disconnected')); + else controller.close(); + }, + })); + }, { defaultHeaders: { 'x-remove': 'default', 'last-event-id': 'default' } }); + const params = { last_event_id: 'initial', ...(mode === 'managed' ? { workspace_id: 'workspace', betas: ['beta-one'] } : {}) }; + const options = { headers: { 'last-event-id': 'override', 'x-test': 'kept', 'x-remove': null }, timeout: 7000, maxRetries: 0 }; + const stream = subscribe(client, mode, params, options); + options.headers['x-test'] = 'changed'; + params.last_event_id = 'changed'; + if (params.betas) params.betas.push('changed'); + const iterator = stream[Symbol.asyncIterator](); + assert.equal((await iterator.next()).value.type, 'session.thread_status_idle'); + const next = iterator.next(); + await retry(t); + assert.equal((await next).value.type, 'session.thread_status_terminated'); + assert.equal((await iterator.next()).done, true); + assert.equal(stream.lastEventID, 'terminal'); + assert.deepEqual(requests.map(req => req.headers.get('last-event-id')), ['override', 'checkpoint']); + for (const req of requests) { + const url = new URL(req.url); + assert(url.pathname.endsWith('/sessions/sess-one/threads/thread-one/stream')); + assert.equal(url.search, ''); + assert.equal(req.method, 'GET'); + assert.equal(req.headers.get('x-test'), 'kept'); + assert.equal(req.headers.get('x-remove'), null); + assert.equal(req.headers.get('x-qoder-timeout'), '7'); + if (mode === 'managed') { + assert.equal(req.headers.get('qoder-workspace-id'), 'workspace'); + assert.equal(req.headers.get('x-qoder-beta'), 'beta-one'); + } + } + }); + } + + for (const status of [400, 401, 403, 404, 409, 429, 503]) { + test(`${mode} thread: HTTP ${status} retry policy`, async t => { + t.mock.timers.enable({ apis: ['setTimeout'] }); + const cursors = []; + const client = testClient(mode, req => { + cursors.push(req.headers.get('last-event-id')); + return cursors.length === 1 + ? new Response(JSON.stringify({ error: { message: 'test failure' } }), { status }) + : new Response(frame('terminal', 'session.thread_status_terminated')); + }, { maxRetries: 0 }); + const stream = subscribe(client, mode, { last_event_id: 'initial' }); + const iterator = stream[Symbol.asyncIterator](); + const next = iterator.next(); + if (status === 429 || status === 503) { + await retry(t); + assert.equal((await next).value.type, 'session.thread_status_terminated'); + assert.equal((await iterator.next()).done, true); + assert.deepEqual(cursors, ['initial', 'initial']); + } else { + await assert.rejects(next, error => error.status === status); + assert.deepEqual(cursors, ['initial']); + } + }); + } + + for (const action of ['abort', 'close']) { + test(`${mode} thread: ${action} interrupts backoff without reconnecting`, async t => { + t.mock.timers.enable({ apis: ['setTimeout'] }); + let calls = 0; + const client = testClient(mode, () => { calls++; return new Response(''); }); + const controller = new AbortController(); + const stream = subscribe(client, mode, {}, { signal: controller.signal }); + const next = stream[Symbol.asyncIterator]().next(); + await new Promise(resolve => setImmediate(resolve)); + if (action === 'abort') { + controller.abort(); + await assert.rejects(next, sdk.APIUserAbortError); + } else { + await stream.close(); + assert.equal((await next).done, true); + } + t.mock.timers.tick(10_000); + assert.equal(calls, 1); + }); + } + + test(`${mode}: child termination does not end a Session subscription`, async () => { + const client = testClient(mode, () => new Response(frame('thread', 'session.thread_status_terminated') + frame('session', 'session.deleted'))); + const events = []; + for await (const event of client.sessions.events.resumableStream('sess-one')) events.push(event.type); + assert.deepEqual(events, ['session.thread_status_terminated', 'session.deleted']); + }); + + test(`${mode} thread: missing IDs fail before opening a connection`, () => { + const client = testClient(mode, () => { throw new Error('unexpected HTTP request'); }); + if (mode === 'forward') { + assert.throws(() => client.sessions.threads.events.resumableStream('', 'thread'), /session_id/); + assert.throws(() => client.sessions.threads.events.resumableStream('session', ''), /thread_id/); + } else { + assert.throws(() => client.sessions.threads.events.resumableStream('thread', { session_id: '' }), /session_id/); + assert.throws(() => client.sessions.threads.events.resumableStream('', { session_id: 'session' }), /thread_id/); + } + }); +} diff --git a/tests/types-smoke.ts b/tests/types-smoke.ts index 4ebc134..fcd0995 100644 --- a/tests/types-smoke.ts +++ b/tests/types-smoke.ts @@ -3,7 +3,7 @@ import { ForwardClient, ManagedClient, APIPromise, Stream, ResumableSessionEvent import Forward from 'qca-sdk/forward'; import Managed from 'qca-sdk/managed'; import type { TemplateUpdateParams, SessionEvent, ServiceAccountToken, ServiceAccountTokenRevokeResponse } from 'qca-sdk/forward'; -import type { AgentUpdateParams, ManagedAgentsStreamSessionEventsUnion, SessionEventStreamParams as ManagedSessionEventStreamParams } from 'qca-sdk/managed'; +import type { AgentUpdateParams, ManagedAgentsStreamSessionEventsUnion, ManagedAgentsStreamSessionThreadEventsUnion, SessionEventStreamParams as ManagedSessionEventStreamParams } from 'qca-sdk/managed'; const forward: ForwardClient = new Forward({ pat: async () => 'token', timeout: 5000 }); const managed: ManagedClient = new Managed(); @@ -51,6 +51,14 @@ const stream: APIPromise> = forward.sessions.events.streamE const resumable: ResumableSessionEventStream = forward.sessions.events.resumableStream('session', { last_event_id: 'event' }); const managedStreamParams: ManagedSessionEventStreamParams = { last_event_id: 'event', event_deltas: ['agent.message'] }; const managedResumable: ResumableSessionEventStream = managed.sessions.events.resumableStream('session', managedStreamParams); +const threadResumable: ResumableSessionEventStream = forward.sessions.threads.events.resumableStream('session', 'thread', { last_event_id: 'event' }); +const managedThreadResumable: ResumableSessionEventStream = managed.sessions.threads.events.resumableStream('thread', { session_id: 'session', last_event_id: 'event' }); +void threadResumable; +void managedThreadResumable; +// @ts-expect-error Managed requires the parent Session ID. +managed.sessions.threads.events.resumableStream('thread', { last_event_id: 'event' }); +// @ts-expect-error Forward requires a Thread ID. +forward.sessions.threads.events.resumableStream('session'); void stream; void resumable; void managedResumable;