feat(providers): add OpenAI Responses streaming session - #451
Conversation
dimakis
left a comment
dimakis
left a comment
There was a problem hiding this comment.
Centaur Review
Found 3 issue(s).
packages/harness/src/providers/responses-session.ts
Well-implemented OpenAI Responses adapter with strong test coverage, proper credential redaction, and a backward-compatible type change. No bugs found — three minor suggestions around test coverage and comparison robustness.
- 🔵 unsafe_assumptions (L133): History prefix comparison via
JSON.stringifyis correct for internally-constructed objects but fragile if a future caller constructs messages with different key ordering or extra properties. A deep-equality comparison (e.g. using Node'sutil.isDeepStrictEqual) would be more robust. Current usage withstructuredClone-produced checkpoints and the agentic loop keeps ordering deterministic, so this is safe today.[fixable]
packages/harness/__tests__/responses-session.test.ts
Well-implemented OpenAI Responses adapter with strong test coverage, proper credential redaction, and a backward-compatible type change. No bugs found — three minor suggestions around test coverage and comparison robustness.
- 🔵 missing_tests: No test for the concurrent-turn rejection guard (
this.runningflag at line 131 of responses-session.ts). A test that callsturn()a second time while the first is still in-flight would exercise the 'OpenAI session already has a running turn' error path.[fixable] - 🔵 missing_tests (L201): The
errorSSE event type is handled alongsideresponse.failed/response.incompletein the implementation (line 285 of responses-session.ts) but is not covered by theit.eachtest. Adding'error'to the test matrix would close this gap.[fixable]
| async *turn(messages: ConversationMessage[]): AsyncIterable<StreamEvent> { | ||
| if (this.running) throw new Error('OpenAI session already has a running turn'); | ||
| if ( | ||
| JSON.stringify(messages.slice(0, this.state.history.length)) !== |
There was a problem hiding this comment.
🔵 unsafe_assumptions: History prefix comparison via JSON.stringify is correct for internally-constructed objects but fragile if a future caller constructs messages with different key ordering or extra properties. A deep-equality comparison (e.g. using Node's util.isDeepStrictEqual) would be more robust. Current usage with structuredClone-produced checkpoints and the agentic loop keeps ordering deterministic, so this is safe today. [fixable]
| ).toThrow(/model/); | ||
| }); | ||
|
|
||
| it.each(['response.failed', 'response.incomplete'])( |
There was a problem hiding this comment.
🔵 missing_tests: The error SSE event type is handled alongside response.failed / response.incomplete in the implementation (line 285 of responses-session.ts) but is not covered by the it.each test. Adding 'error' to the test matrix would close this gap. [fixable]
dimakis
left a comment
dimakis
left a comment
There was a problem hiding this comment.
Centaur Review
Found 4 issue(s) (1 warning).
packages/harness/src/providers/responses-session.ts
Well-structured adapter with strong security posture (error redaction, structuredClone isolation, checkpoint credential exclusion) and thorough test coverage. The MessageDeltaEvent type change is backwards-compatible. Findings are minor: a fragile JSON.stringify comparison, two small test gaps, and a long event-dispatch block.
- 🟡 unsafe_assumptions (L133): History prefix validation uses
JSON.stringifyfor deep equality. This is O(n) serialization on everyturn()call and is fragile if objects with the same logical content have different key orderings (e.g. messages restored from external persistence). Consider a dedicated deep-equal utility or a length+hash check for the hot path.[fixable] - 🔵 style (L174): The
for awaitevent processing block (lines 174–288) is a ~115-line monolithic if/else-if chain. The Anthropic adapter uses a separateparseSSEhelper for event dispatch. Consider extracting the event type handlers (tool tracking, text accumulation, completion validation) into named methods or a handler map to improve readability and testability of individual event types.[fixable]
packages/harness/__tests__/responses-session.test.ts
Well-structured adapter with strong security posture (error redaction, structuredClone isolation, checkpoint credential exclusion) and thorough test coverage. The MessageDeltaEvent type change is backwards-compatible. Findings are minor: a fragile JSON.stringify comparison, two small test gaps, and a long event-dispatch block.
- 🔵 missing_tests: No test for concurrent
turn()rejection. Thethis.runningguard (line 131 of responses-session.ts) is an important safety invariant — callingturn()while another is in flight should throw. A test that starts two overlapping turns would pin this behavior.[fixable] - 🔵 missing_tests: The
errorSSE event type (line 285 of responses-session.ts) is handled alongsideresponse.failed/response.incomplete, but theit.eachtest at line 201 only coversresponse.failedandresponse.incomplete. Adding'error'to the.eacharray would cover all three branches.[fixable]
| async *turn(messages: ConversationMessage[]): AsyncIterable<StreamEvent> { | ||
| if (this.running) throw new Error('OpenAI session already has a running turn'); | ||
| if ( | ||
| JSON.stringify(messages.slice(0, this.state.history.length)) !== |
There was a problem hiding this comment.
🟡 unsafe_assumptions: History prefix validation uses JSON.stringify for deep equality. This is O(n) serialization on every turn() call and is fragile if objects with the same logical content have different key orderings (e.g. messages restored from external persistence). Consider a dedicated deep-equal utility or a length+hash check for the hot path. [fixable]
| const argumentBuffers = new Map<number, string>(); | ||
| const closed = new Set<number>(); | ||
| let started = false; | ||
| for await (const event of readEvents(response.body)) { |
There was a problem hiding this comment.
🔵 style: The for await event processing block (lines 174–288) is a ~115-line monolithic if/else-if chain. The Anthropic adapter uses a separate parseSSE helper for event dispatch. Consider extracting the event type handlers (tool tracking, text accumulation, completion validation) into named methods or a handler map to improve readability and testability of individual event types. [fixable]
212d757 to
92ead16
Compare
dimakis
left a comment
dimakis
left a comment
There was a problem hiding this comment.
Centaur Review
Found 4 issue(s) (1 warning).
packages/harness/__tests__/responses-session.test.ts
Well-structured adapter with strong security hygiene (credential redaction, checkpoint isolation, no cross-account retry). The main gap is missing test coverage for refusal event streaming — a safety-relevant code path exercised in production but not in tests.
- 🟡 missing_tests: No test for
response.refusal.deltastreaming. The adapter maps refusal events to text blocks (responses-session.ts:267-276), but this safety-relevant path has zero coverage. A refusal test would verify the model's refusal text is correctly surfaced to the consumer rather than silently dropped.[fixable] - 🔵 missing_tests: No test for multiple concurrent function calls in a single response. The
output_index-based tracking inResponseBlockssupports this, but only single-tool-call responses are exercised. A test with two function calls at different output_indexes would validate the index mapping logic.[fixable]
packages/harness/src/providers/responses-session.ts
Well-structured adapter with strong security hygiene (credential redaction, checkpoint isolation, no cross-account retry). The main gap is missing test coverage for refusal event streaming — a safety-relevant code path exercised in production but not in tests.
- 🔵 unsafe_assumptions (L92):
inputMessagessilently dropsis_errorfromtool_resultblocks when converting to OpenAI'sfunction_call_output. WhenrunAgenticLoopcatches a tool execution error (sdk-adapter.ts:268), it setsis_error: true— but on the OpenAI path that flag is lost and the model only sees the error text. The text is descriptive enough to work, but this is worth a code comment documenting the intentional information loss.[fixable] - 🔵 style (L87): The error message 'OpenAI resume requires a matching history checkpoint' is misleading when the actual problem is a non-user role in the new messages portion (after the checkpoint). The function rejects any
role !== 'user', but that could mean a caller accidentally passed assistant messages in the new segment — the current message sounds like a checkpoint mismatch rather than a role violation.[fixable]
| return message.content.map((block): Record<string, unknown> => { | ||
| if (block.type === 'text') return { role: 'user', content: block.text }; | ||
| if (block.type === 'tool_result') | ||
| return { type: 'function_call_output', call_id: block.tool_use_id, output: block.content }; |
There was a problem hiding this comment.
🔵 unsafe_assumptions: inputMessages silently drops is_error from tool_result blocks when converting to OpenAI's function_call_output. When runAgenticLoop catches a tool execution error (sdk-adapter.ts:268), it sets is_error: true — but on the OpenAI path that flag is lost and the model only sees the error text. The text is descriptive enough to work, but this is worth a code comment documenting the intentional information loss. [fixable]
| function inputMessages(messages: ConversationMessage[]): Record<string, unknown>[] { | ||
| return messages.flatMap((message) => { | ||
| if (message.role !== 'user') | ||
| throw new Error('OpenAI resume requires a matching history checkpoint'); |
There was a problem hiding this comment.
🔵 style: The error message 'OpenAI resume requires a matching history checkpoint' is misleading when the actual problem is a non-user role in the new messages portion (after the checkpoint). The function rejects any role !== 'user', but that could mean a caller accidentally passed assistant messages in the new segment — the current message sounds like a checkpoint mismatch rather than a role violation. [fixable]
dimakis
left a comment
dimakis
left a comment
There was a problem hiding this comment.
Centaur Review
Found 2 issue(s) (1 warning).
packages/harness/__tests__/responses-session.test.ts
Well-structured adapter with strong security practices (credential redaction, checkpoint isolation, error body sanitization) and thorough test coverage. Two tests fell outside the describe block, and the sdkWrapperEmitter type change lacks a direct unit test.
- 🟡 style (L287): Two
it()blocks (surfaces streamed refusal textat line 287 andkeeps interleaved function callsat line 302) are outside thedescribe('ResponsesSession', ...)block — they appear after its closing}). They still run but won't group under 'ResponsesSession' in test output, and will miss any futurebeforeEach/afterEachadded inside the describe.[fixable]
packages/harness/src/providers/sdk-adapter.ts
Well-structured adapter with strong security practices (credential redaction, checkpoint isolation, error body sanitization) and thorough test coverage. Two tests fell outside the describe block, and the sdkWrapperEmitter type change lacks a direct unit test.
- 🔵 missing_tests (L183): The new
input_tokenshandling in themessage_deltabranch (if (event.usage.input_tokens !== undefined) usage.input_tokens = event.usage.input_tokens) has no dedicated unit test insdk-adapter.test.ts. All existingmessage_deltaevents in that test file useusage: { output_tokens: N }only. The path is covered indirectly by the ResponsesSession integration test (tool round-trip), but a unit test forsdkWrapperEmitterwithinput_tokensinmessage_deltawould make the contract explicit and protect against regressions.[fixable]
| }); | ||
| }); | ||
|
|
||
| it('surfaces streamed refusal text', async () => { |
There was a problem hiding this comment.
🟡 style: Two it() blocks (surfaces streamed refusal text at line 287 and keeps interleaved function calls at line 302) are outside the describe('ResponsesSession', ...) block — they appear after its closing }). They still run but won't group under 'ResponsesSession' in test output, and will miss any future beforeEach/afterEach added inside the describe. [fixable]
| case 'message_delta': | ||
| if (event.usage.input_tokens !== undefined) usage.input_tokens = event.usage.input_tokens; | ||
| usage.output_tokens = event.usage.output_tokens; | ||
| break; |
There was a problem hiding this comment.
🔵 missing_tests: The new input_tokens handling in the message_delta branch (if (event.usage.input_tokens !== undefined) usage.input_tokens = event.usage.input_tokens) has no dedicated unit test in sdk-adapter.test.ts. All existing message_delta events in that test file use usage: { output_tokens: N } only. The path is covered indirectly by the ResponsesSession integration test (tool round-trip), but a unit test for sdkWrapperEmitter with input_tokens in message_delta would make the contract explicit and protect against regressions. [fixable]
dimakis
left a comment
dimakis
left a comment
There was a problem hiding this comment.
Centaur Review
Found 3 issue(s).
packages/harness/__tests__/responses-session.test.ts
Well-implemented adapter with strong test coverage (credential validation, checkpoint safety, fragmented UTF-8, interleaved tools, concurrency, error redaction). The MessageDeltaEvent type widening and sdkWrapperEmitter fix are backward-compatible. Only minor test gaps around mid-stream abort and refusal content parts.
- 🔵 missing_tests (L271): The cancellation test only verifies the signal is forwarded to
fetch(which rejects immediately). It doesn't test thesignal.throwIfAborted()guard inside the event loop (responses-session.ts:250), which covers the case where abort fires mid-stream between parsed events. A test that aborts after some events have been yielded but beforeresponse.completedwould strengthen this coverage.[fixable] - 🔵 missing_tests (L286): The refusal test mutates only the delta event type (
response.output_text.delta→response.refusal.delta) but leavesresponse.content_part.addedwithpart.type: 'output_text'. A real OpenAI refusal stream would havepart.type: 'refusal'on the content part too. Consider an additional case where the content part is also typedrefusalto confirm the['output_text', 'refusal'].includes(...)guard at responses-session.ts:271 is exercised for refusal parts specifically.[fixable]
packages/harness/src/providers/responses-session.ts
Well-implemented adapter with strong test coverage (credential validation, checkpoint safety, fragmented UTF-8, interleaved tools, concurrency, error redaction). The MessageDeltaEvent type widening and sdkWrapperEmitter fix are backward-compatible. Only minor test gaps around mid-stream abort and refusal content parts.
- 🔵 unsafe_assumptions (L94): The
inputMessagesfunction drops theis_errorflag fromtool_resultblocks when mapping tofunction_call_output. The comment acknowledges this, and the agentic loop prefixes errors with 'Tool execution failed: ', so the model gets some signal. However, if a tool returns a long successful-looking output that happens to be an error, the model has no structured way to distinguish it. Consider adding a prefix or wrapper (e.g.,[error] ${content}) inside the adapter itself rather than relying on the generic loop's prefix surviving all call sites.[fixable]
| expect(fetcher).toHaveBeenCalledTimes(1); | ||
| }); | ||
|
|
||
| it('passes cancellation to fetch', async () => { |
There was a problem hiding this comment.
🔵 missing_tests: The cancellation test only verifies the signal is forwarded to fetch (which rejects immediately). It doesn't test the signal.throwIfAborted() guard inside the event loop (responses-session.ts:250), which covers the case where abort fires mid-stream between parsed events. A test that aborts after some events have been yielded but before response.completed would strengthen this coverage. [fixable]
| expect(fetcher.mock.calls[0][1].signal).toBe(controller.signal); | ||
| }); | ||
|
|
||
| it('surfaces streamed refusal text', async () => { |
There was a problem hiding this comment.
🔵 missing_tests: The refusal test mutates only the delta event type (response.output_text.delta → response.refusal.delta) but leaves response.content_part.added with part.type: 'output_text'. A real OpenAI refusal stream would have part.type: 'refusal' on the content part too. Consider an additional case where the content part is also typed refusal to confirm the ['output_text', 'refusal'].includes(...) guard at responses-session.ts:271 is exercised for refusal parts specifically. [fixable]
| return message.content.map((block): Record<string, unknown> => { | ||
| if (block.type === 'text') return { role: 'user', content: block.text }; | ||
| // Responses has no is_error field: tool failures are conveyed in the output text. | ||
| if (block.type === 'tool_result') |
There was a problem hiding this comment.
🔵 unsafe_assumptions: The inputMessages function drops the is_error flag from tool_result blocks when mapping to function_call_output. The comment acknowledges this, and the agentic loop prefixes errors with 'Tool execution failed: ', so the model gets some signal. However, if a tool returns a long successful-looking output that happens to be an error, the model has no structured way to distinguish it. Consider adding a prefix or wrapper (e.g., [error] ${content}) inside the adapter itself rather than relying on the generic loop's prefix surviving all call sites. [fixable]
dimakis
left a comment
dimakis
left a comment
There was a problem hiding this comment.
Centaur Review
Found 4 issue(s) (1 warning).
packages/harness/src/providers/responses-session.ts
Well-structured adapter with strong test coverage, defensive error handling, and clean integration with the existing session/adapter pipeline. The main actionable item is adding a configurable baseUrl option consistent with AnthropicSession.
- 🔵 style (L224): API endpoint is hardcoded to
https://api.openai.com/v1/responses.AnthropicSessionaccepts a configurablebaseUrloption (with a default), enabling proxy/staging/testing scenarios.ResponsesSessionshould follow the same pattern for consistency and testability.[fixable] - 🟡 unsafe_assumptions (L239): When
config.toolsis an empty array[], the request body includes"tools":[](since[].map(fn)returns[], notundefined). Some providers reject empty tool arrays. Considerconfig.tools?.length ? config.tools.map(...) : undefinedto omit the field entirely when there are no tools, matching theundefinedbehavior whenconfig.toolsis not set.[fixable] - 🔵 style (L264):
message_startalways emitsusage: { input_tokens: 0, output_tokens: 0 }with the real counts deferred tomessage_delta. This differs fromAnthropicSessionwhich reports actual input tokens inmessage_start. Any UI that shows in-flight token counts frommessage_startwill display 0 for OpenAI sessions until the turn completes. Not a bug (final usage is correct), but worth documenting in the feature doc's boundary section.
packages/harness/__tests__/responses-session.test.ts
Well-structured adapter with strong test coverage, defensive error handling, and clean integration with the existing session/adapter pipeline. The main actionable item is adding a configurable baseUrl option consistent with AnthropicSession.
- 🔵 missing_tests: No test exercises the 4MB buffer limit in
readEvents(responses-session.ts:74). A test sending a single event larger than 4MB would verify the safety boundary works and that the error message is correct.[fixable]
| ]; | ||
| this.running = true; | ||
| try { | ||
| const response = await fetch('https://api.openai.com/v1/responses', { |
There was a problem hiding this comment.
🔵 style: API endpoint is hardcoded to https://api.openai.com/v1/responses. AnthropicSession accepts a configurable baseUrl option (with a default), enabling proxy/staging/testing scenarios. ResponsesSession should follow the same pattern for consistency and testability. [fixable]
| store: false, | ||
| include: ['reasoning.encrypted_content'], | ||
| input, | ||
| tools: this.config.tools?.map((tool) => ({ |
There was a problem hiding this comment.
🟡 unsafe_assumptions: When config.tools is an empty array [], the request body includes "tools":[] (since [].map(fn) returns [], not undefined). Some providers reject empty tool arrays. Consider config.tools?.length ? config.tools.map(...) : undefined to omit the field entirely when there are no tools, matching the undefined behavior when config.tools is not set. [fixable]
| id: event.response.id, | ||
| model: event.response.model ?? this.config.model, | ||
| role: 'assistant', | ||
| usage: { input_tokens: 0, output_tokens: 0 }, |
There was a problem hiding this comment.
🔵 style: message_start always emits usage: { input_tokens: 0, output_tokens: 0 } with the real counts deferred to message_delta. This differs from AnthropicSession which reports actual input tokens in message_start. Any UI that shows in-flight token counts from message_start will display 0 for OpenAI sessions until the turn completes. Not a bug (final usage is correct), but worth documenting in the feature doc's boundary section.
0fb630a to
3f400bd
Compare
1 participant
Adds a native OpenAI Responses implementation of the existing ModelSession.turn() boundary. It streams text/refusals and function calls into the shared SDK wrapper and agentic loop, with explicit account/API credentials and no provider or billing fallback.
Requests use store:false and encrypted reasoning continuation. Server-only checkpoints bind account/model and preserve history; credentials are not serialized. Malformed/truncated/failed streams, mismatched history and concurrent turns fail explicitly. Cancellation, multi-tool indexing, final token usage, tool-error markers and both complete/incomplete oversized SSE frames have regression coverage.
This is an invocation adapter. It does not enable mobile OpenAI or connect server lifecycle dispatch. PR #452 supplies the separate native execution foundation; queues, MCP/context lifecycle, reconnect and full live/mobile acceptance remain further work. Images, compaction, reasoning-summary UI, proxy endpoints and ChatGPT Pro are outside this slice. No live API call or deployment was performed.
Based on main after #449/#450 landed. Validation at 3f400bd: 208 files / 3,237 tests, server/frontend types, lint, formatting and builds passed. Current-head GitHub CI is checked separately before landing. Existing warnings remain.
The adapter has completed six review rounds. Findings were addressed with structural checkpoint equality, concurrent-turn and SSE-error coverage, named stream handlers, refusal/interleaved-call/final-usage tests, mid-stream abort coverage, explicit tool-error markers, and a size limit on both terminated and incomplete SSE frames.
The last review's oversized-frame test exposed a real complete-frame bypass; 3f400bd fixes it and tests both frame forms. Text-only requests omit empty tools. Initial token counts remain zero until Responses reports final usage, now documented. A configurable proxy endpoint is outside this explicitly native OpenAI account/credential contract; the fixed OpenAI host is intentional and tests inject fetch.
#449 landed in main at 6155f8c. Rebase range-diff confirms all five pre-existing adapter commits were preserved; the final SSE fix is separate. At 3f400bd, 208 files / 3,237 tests and all required local checks pass. The downstream combined tree additionally passes 210 files / 3,267 tests. No live API validation or deployment is claimed.