ChatClient reports status ready before connect() adapter queue drains

TanStack AI version

0.30.0 (@tanstack/ai-client)

Framework/Library version

N/A, framework-agnostic (plain ChatClient, no UI framework)

Describe the bug and the steps to reproduce it

For a ConnectionAdapter that implements only connect() (no native subscribe/send), normalizeConnectionAdapter's legacy wrapper in connection-adapters.ts drains the adapter into an internal queue and waits for it via waitUntilSubscriberIdle(). That wait is bounded (16 microtask polls, then 32 macrotask polls) and returns even when the queue is not empty. streamResponse() in chat-client.ts then calls processor.finalizeStream() directly, commented as idempotent because RUN_FINISHED should already have triggered it. Under enough per run event volume this fires onStreamEnd and status ready before the real terminal event is processed.

Reproduction with the snippet below, sendMessage called once:

Below around 15 step pairs, status ready fires correctly, after the message.

From around 16 pairs on, status ready fires while the message is still assembling. The message still lands in getMessages() later, after ready already fired.

Expected: ready and onStreamEnd fire only after the connect wrapper's queue drains and the real RUN_FINISHED is processed.

Actual: the bounded poll in waitUntilSubscriberIdle gives up under backlog and finalizeStream() runs on partial state.

Your Minimal, Reproducible Example - (Sandbox Highly Recommended)

No sandbox, single file reproduction below, no dependency beyond @tanstack/ai-client.

const chatClient = new ChatClient({ 
 connection: { 
   connect: async function* () { 
     yield { type: 'RUN_STARTED', threadId: 't1', runId: 'r1' } 
     for (let i = 0; i < 20; i++) { 
       yield { type: 'STEP_STARTED', stepName: `s${i}` } 
       yield { type: 'STEP_FINISHED', stepName: `s${i}` } 
     } 
     yield { type: 'TEXT_MESSAGE_START', messageId: 'a1', role: 'assistant' } 
     yield { type: 'TEXT_MESSAGE_CONTENT', messageId: 'a1', delta: 'done' } 
     yield { type: 'TEXT_MESSAGE_END', messageId: 'a1' } 
     yield { type: 'RUN_FINISHED', threadId: 't1', runId: 'r1' } 
   }, 
 }, 
 onStatusChange: (s) => console.log('status', s), 
 onMessagesChange: (m) => console.log('messages', m.length), 
}) 
await chatClient.sendMessage('go') 

Terms & Code of Conduct

• I agree to follow this project's Code of Conduct
• I understand that if my bug cannot be reliable reproduced in a debuggable environment, it will probably not be fixed and this issue may even be closed.