Repository navigation
refactor(serve): extract createInMemoryChannel helper (#4156 A1) #4160
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
Merged
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit
Hold shift + click to select a range
d2fed2a
refactor(serve): extract createInMemoryChannel helper from httpAcpBri…
doudouOUC 374546d
refactor(serve): address Copilot review feedback on createInMemoryCha…
doudouOUC 83d36c0
refactor(serve): expose abort() teardown primitive on createInMemoryC…
doudouOUC File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,147 @@ | ||
| /** | ||
| * @license | ||
| * Copyright 2025 Qwen Team | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| import { describe, it, expect } from 'vitest'; | ||
| import type { AnyMessage } from '@agentclientprotocol/sdk'; | ||
| import { createInMemoryChannel } from './inMemoryChannel.js'; | ||
|
|
||
| /** | ||
| * Push one JSON-RPC notification onto a `Stream.writable`. The SDK's | ||
| * `ndJsonStream` encodes the message + appends `\n` so the matching | ||
| * decoder on the other side can frame it. | ||
| */ | ||
| async function send( | ||
| writable: WritableStream<AnyMessage>, | ||
| msg: AnyMessage, | ||
| ): Promise<void> { | ||
| const writer = writable.getWriter(); | ||
| try { | ||
| await writer.write(msg); | ||
| } finally { | ||
| writer.releaseLock(); | ||
| } | ||
| } | ||
|
|
||
| /** Read the next single message off a `Stream.readable`. */ | ||
| async function recvOne( | ||
| readable: ReadableStream<AnyMessage>, | ||
| ): Promise<AnyMessage> { | ||
| const reader = readable.getReader(); | ||
| try { | ||
| const { value, done } = await reader.read(); | ||
| if (done || !value) { | ||
| throw new Error('stream closed before a frame arrived'); | ||
| } | ||
| return value; | ||
| } finally { | ||
| reader.releaseLock(); | ||
| } | ||
| } | ||
|
|
||
| describe('createInMemoryChannel', () => { | ||
| it('round-trips a frame from client to agent', async () => { | ||
| const { clientStream, agentStream } = createInMemoryChannel(); | ||
| const sent: AnyMessage = { | ||
| jsonrpc: '2.0', | ||
| method: 'ping', | ||
| params: { n: 1 }, | ||
| }; | ||
| await send(clientStream.writable, sent); | ||
| const received = await recvOne(agentStream.readable); | ||
| expect(received).toEqual(sent); | ||
| }); | ||
|
|
||
| it('round-trips a frame from agent to client', async () => { | ||
| const { clientStream, agentStream } = createInMemoryChannel(); | ||
| const sent: AnyMessage = { | ||
| jsonrpc: '2.0', | ||
| method: 'pong', | ||
| params: { reply: true }, | ||
| }; | ||
| await send(agentStream.writable, sent); | ||
| const received = await recvOne(clientStream.readable); | ||
| expect(received).toEqual(sent); | ||
| }); | ||
|
|
||
| it('preserves order across multiple frames in one direction', async () => { | ||
| const { clientStream, agentStream } = createInMemoryChannel(); | ||
| const writer = clientStream.writable.getWriter(); | ||
| for (let i = 1; i <= 3; i++) { | ||
| await writer.write({ | ||
| jsonrpc: '2.0', | ||
| method: 'tick', | ||
| params: { i }, | ||
| }); | ||
| } | ||
| writer.releaseLock(); | ||
|
|
||
| const reader = agentStream.readable.getReader(); | ||
| const out: AnyMessage[] = []; | ||
| for (let i = 0; i < 3; i++) { | ||
| const { value, done } = await reader.read(); | ||
| if (done || !value) throw new Error('unexpected close'); | ||
| out.push(value); | ||
| } | ||
| reader.releaseLock(); | ||
|
|
||
| expect(out.map((m) => (m as { params: { i: number } }).params.i)).toEqual([ | ||
| 1, 2, 3, | ||
| ]); | ||
| }); | ||
|
|
||
| it('isolates client→agent direction (client write does not echo to client.readable)', async () => { | ||
| // Sanity check that the channel is truly paired. A buggy | ||
| // implementation that aliased `ab` to both ends would still pass | ||
| // the round-trip tests above for one frame but echo the writer's | ||
| // own messages back on its own readable. | ||
| const { clientStream, agentStream } = createInMemoryChannel(); | ||
| await send(clientStream.writable, { | ||
| jsonrpc: '2.0', | ||
| method: 'client-only', | ||
| }); | ||
|
|
||
| const onAgent = await recvOne(agentStream.readable); | ||
| expect((onAgent as { method: string }).method).toBe('client-only'); | ||
|
|
||
| // Client's own readable must NOT see it. Race a fresh read against | ||
| // a short timeout — if the channel is correctly paired the read | ||
| // never resolves on its own. | ||
| const reader = clientStream.readable.getReader(); | ||
| try { | ||
| const winner = await Promise.race([ | ||
| reader.read().then(() => 'leaked' as const), | ||
| new Promise<'isolated'>((res) => setTimeout(() => res('isolated'), 50)), | ||
| ]); | ||
| expect(winner).toBe('isolated'); | ||
| } finally { | ||
| reader.releaseLock(); | ||
| } | ||
| }); | ||
|
|
||
| it('isolates agent→client direction (agent write does not echo to agent.readable)', async () => { | ||
| // Symmetric counterpart of the previous test — closes the obvious | ||
| // "wired one direction correctly but not the other" failure mode. | ||
| const { clientStream, agentStream } = createInMemoryChannel(); | ||
| await send(agentStream.writable, { | ||
| jsonrpc: '2.0', | ||
| method: 'agent-only', | ||
| }); | ||
|
|
||
| const onClient = await recvOne(clientStream.readable); | ||
| expect((onClient as { method: string }).method).toBe('agent-only'); | ||
|
|
||
| const reader = agentStream.readable.getReader(); | ||
| try { | ||
| const winner = await Promise.race([ | ||
| reader.read().then(() => 'leaked' as const), | ||
| new Promise<'isolated'>((res) => setTimeout(() => res('isolated'), 50)), | ||
| ]); | ||
| expect(winner).toBe('isolated'); | ||
| } finally { | ||
| reader.releaseLock(); | ||
| } | ||
| }); | ||
| }); | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,48 @@ | ||
| /** | ||
| * @license | ||
| * Copyright 2025 Qwen Team | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| import { ndJsonStream, type Stream } from '@agentclientprotocol/sdk'; | ||
|
|
||
| /** | ||
| * Create a paired in-memory NDJSON channel: two `Stream`s connected | ||
| * back-to-back via two `TransformStream<Uint8Array, Uint8Array>` pairs. | ||
| * Whatever `clientStream.writable` writes appears on `agentStream.readable`, | ||
| * and vice versa. Each side is a full ACP `Stream` (via SDK `ndJsonStream`) | ||
| * so callers can hand them to `ClientSideConnection` / `AgentSideConnection` | ||
| * exactly as they would a real stdio pair. | ||
|
doudouOUC marked this conversation as resolved.
Resolved
|
||
| * | ||
| * Used today by Stage 1 tests (replaces 10 sites of inline boilerplate | ||
| * in `httpAcpBridge.test.ts`). Will also be consumed by the Stage 1.5b | ||
| * in-process bridge (issue #4156) when that lands, to wrap an in-process | ||
| * `QwenAgent` without spawning a `qwen --acp` child. | ||
| * | ||
| * The helper is intentionally bare — it returns only the stream pair, no | ||
| * lifecycle / teardown surface. Two reasons: | ||
| * | ||
| * 1. Consumer behavior diverges widely (stuck channel, crashable | ||
| * child simulation, no-op, real in-process termination). A | ||
| * one-size-fits-all `close()` would either pull test-fixture | ||
| * concerns into a production module or force a single shape on | ||
| * consumers that don't want it. | ||
| * | ||
| * 2. The SDK's `ndJsonStream` outer wrapper does not reliably | ||
| * propagate close on `Stream.writable` to the opposite | ||
| * `Stream.readable`. Consumers needing to simulate a child exit | ||
| * (like the inline `makeChannel` in `httpAcpBridge.test.ts`) hold | ||
| * onto their own underlying `TransformStream` references and | ||
| * close those directly. A `close()` on this helper would have to | ||
| * either expose the same internals or be silently incomplete. | ||
| */ | ||
| export function createInMemoryChannel(): { | ||
|
doudouOUC marked this conversation as resolved.
Resolved
|
||
| clientStream: Stream; | ||
| agentStream: Stream; | ||
| } { | ||
| const ab = new TransformStream<Uint8Array, Uint8Array>(); | ||
| const ba = new TransformStream<Uint8Array, Uint8Array>(); | ||
| const clientStream = ndJsonStream(ab.writable, ba.readable); | ||
| const agentStream = ndJsonStream(ba.writable, ab.readable); | ||
| return { clientStream, agentStream }; | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.