diff --git a/packages/experimental/cua/src/native-client.ts b/packages/experimental/cua/src/native-client.ts new file mode 100644 index 0000000000..5e3849f6dc --- /dev/null +++ b/packages/experimental/cua/src/native-client.ts @@ -0,0 +1,159 @@ +/** Bounded JSON-line transport to one owned native helper process. */ +import { randomUUID } from 'node:crypto' +import type { SubprocessHandle, SubprocessRuntime } from '@deepseek-ai/dsh-subprocess' +import type { JsonValue } from '@deepseek-ai/dsh-util-values' +import { z } from 'zod' + +const Reply = z.union([ + z.object({ id: z.string(), result: z.json() }).strict(), + z.object({ id: z.string(), error: z.object({ code: z.string().min(1), message: z.string() }).strict() }).strict(), +]) + +/** Native refusal or transport failure with a stable machine-readable code. */ +export class NativeError extends Error { + /** + * @param code - Native error code or a transport-owned NATIVE_* code. + * @param message - Actionable diagnostic, without its code prefix. + * @param options - Original failure when available. + */ + constructor(readonly code: string, message: string, options?: ErrorOptions) { + super(`${code}: ${message}`, options) + this.name = 'NativeError' + } +} + +/** Resolved helper executable and bounded process/response configuration. */ +export interface NativeClientConfig { + helperPath: string + cwd: string + timeoutMs: number + maxRequestBytes: number + maxResponseBytes: number + graceMs: number +} + +/** Serial helper calls retain native handles until helper disposal or replacement. */ +export class NativeClient { + private readonly handle: SubprocessHandle + private readonly config: NativeClientConfig + private tail = Buffer.alloc(0) + private failure: NativeError | undefined + private disposal: Promise | undefined + private active: { id: string; resolve: (value: JsonValue) => void; reject: (error: NativeError) => void } | undefined + + constructor(owner: { subprocess: Pick }, config: NativeClientConfig) { + this.config = config + this.handle = owner.subprocess.spawn({ + argv: [config.helperPath], cwd: config.cwd, graceMs: config.graceMs, + stdio: { stdin: 'pipe', stdout: 'pipe', stderr: { maxBytes: 8192 } }, + }) + this.handle.stdout?.on('data', (chunk: Buffer) => { this.receive(chunk) }) + this.handle.stdout?.on('error', (error: Error) => { this.fail(new NativeError('NATIVE_IO', 'Helper stdout failed', { cause: error })) }) + this.handle.stdout?.on('end', () => { this.fail(new NativeError('NATIVE_EOF', 'Helper output ended; acquire fresh application handles')) }) + this.handle.stdout?.on('close', () => { this.fail(new NativeError('NATIVE_EOF', 'Helper output closed; acquire fresh application handles')) }) + this.handle.stdin?.on('error', (error: Error) => { this.fail(new NativeError('NATIVE_IO', 'Helper stdin failed', { cause: error })) }) + void this.handle.done.then( + () => { this.fail(new NativeError('NATIVE_EXITED', 'Helper exited; acquire fresh application handles')) }, + (error: unknown) => { this.fail(new NativeError('NATIVE_EXITED', 'Helper process failed', { cause: error })) }, + ) + } + + /** + * Send one operation. Native refusals preserve the helper; transport failures invalidate it. + * @param method - Helper operation name. + * @param params - JSON operation parameters. + * @param signal - Cancellation kills this helper and invalidates its retained handles. + * @returns The validated helper result, after failure cleanup when needed. + */ + async request(method: string, params: Record, signal: AbortSignal): Promise { + if (signal.aborted) throw new NativeError('NATIVE_CANCELLED', 'Operation was cancelled before dispatch') + if (this.disposed) { + await this.dispose() + throw new NativeError('NATIVE_CLOSED', 'Helper is closed; acquire fresh application handles', { cause: this.failure }) + } + if (this.active) throw new NativeError('NATIVE_BUSY', 'Await native operations sequentially') + const id = randomUUID() + const frame = `${JSON.stringify({ id, method, params })}\n` + if (Buffer.byteLength(frame) > this.config.maxRequestBytes) { + throw new NativeError('NATIVE_REQUEST_LIMIT', 'Request exceeded the configured byte limit before dispatch') + } + const outcome = Promise.withResolvers() + this.active = { id, ...outcome } + const abort = (): void => { this.fail(new NativeError('NATIVE_CANCELLED', 'Operation cancelled; inspect fresh state before retrying input')) } + signal.addEventListener('abort', abort, { once: true }) + const timer = setTimeout(() => { + this.fail(new NativeError('NATIVE_TIMEOUT', `Operation exceeded ${this.config.timeoutMs}ms`)) + }, this.config.timeoutMs) + try { + if (!this.handle.stdin || !this.handle.stdout) { + this.fail(new NativeError('NATIVE_IO', 'Helper did not expose the requested protocol pipes')) + } else { + try { + this.handle.stdin.write(frame, (error) => { + if (error) this.fail(new NativeError('NATIVE_IO', 'Writing a helper request failed', { cause: error })) + }) + } catch (error) { + this.fail(new NativeError('NATIVE_IO', 'Writing a helper request failed', { cause: error })) + } + } + const result = await outcome.promise + if (this.failure) throw this.failure + return result + } finally { + clearTimeout(timer) + signal.removeEventListener('abort', abort) + if (this.failure !== undefined) await this.dispose() + } + } + + /** Whether this helper has lost its retained native state. */ + get disposed(): boolean { return this.failure !== undefined } + + /** Stop pending work and await the entire managed process range before releasing streams. */ + dispose(): Promise { + this.fail(new NativeError('NATIVE_DISPOSED', 'Helper disposed')) + this.disposal ??= (async () => { + try { + const [, range] = await Promise.allSettled([this.handle.done, this.handle.waitForExit()]) + if (range.status === 'rejected') throw range.reason + } finally { + this.handle.stdout?.destroy() + this.handle.stdin?.destroy() + } + })() + return this.disposal + } + + private fail(error: NativeError): void { + if (this.failure) return + this.failure = error + this.tail = Buffer.alloc(0) + this.active?.reject(error) + this.active = undefined + this.handle.terminate() + } + + private receive(chunk: Buffer): void { + if (this.failure) return + if (this.tail.length + chunk.length > this.config.maxResponseBytes) { + this.fail(new NativeError('NATIVE_RESPONSE_LIMIT', 'Helper response exceeded the configured byte limit')) + return + } + this.tail = Buffer.concat([this.tail, chunk]) + const newline = this.tail.indexOf(10) + if (newline < 0) return + try { + if (newline !== this.tail.length - 1) throw new Error('One request must produce exactly one response frame') + const line = new TextDecoder('utf-8', { fatal: true }).decode(this.tail.subarray(0, newline)) + const reply = Reply.parse(JSON.parse(line)) + const active = this.active + if (!active || reply.id !== active.id) throw new Error('Helper reply has no matching request') + this.tail = Buffer.alloc(0) + this.active = undefined + if ('error' in reply) active.reject(new NativeError(reply.error.code, reply.error.message)) + else active.resolve(reply.result) + } catch (error) { + this.fail(new NativeError('NATIVE_PROTOCOL', 'Helper returned an invalid response; acquire fresh application handles', { cause: error })) + } + } +} diff --git a/packages/experimental/cua/tests/native-client.spec.ts b/packages/experimental/cua/tests/native-client.spec.ts new file mode 100644 index 0000000000..b1e0d7eb13 --- /dev/null +++ b/packages/experimental/cua/tests/native-client.spec.ts @@ -0,0 +1,277 @@ +/** Controlled process streams exercise the platform-independent native protocol client. */ +import { PassThrough, Writable } from 'node:stream' +import type { SubprocessHandle, SubprocessOutcome } from '@deepseek-ai/dsh-subprocess' +import { describe, expect, it, onTestFinished, vi } from 'vitest' +import { z } from 'zod' +import { NativeClient, NativeError } from '../src/native-client.ts' + +const Request = z.object({ id: z.string(), method: z.string(), params: z.record(z.string(), z.json()) }) + +function fixture(options: { + maxRequestBytes?: number + maxResponseBytes?: number + timeoutMs?: number + stdin?: boolean + stdout?: boolean + writeError?: Error +} = {}) { + const requests: z.infer[] = [] + const stdout = new PassThrough() + const done = Promise.withResolvers() + const gone = Promise.withResolvers() + const stdin = new Writable({ write(chunk, _encoding, callback) { + requests.push(Request.parse(JSON.parse(String(chunk)))) + callback(options.writeError) + } }) + const terminated: PromiseWithResolvers = Promise.withResolvers() + const terminate = vi.fn(() => { terminated.resolve() }) + const waitForExit = vi.fn(() => gone.promise) + const handle: SubprocessHandle = { + stdin: options.stdin === false ? undefined : stdin, + stdout: options.stdout === false ? undefined : stdout, + stderr: undefined, control: undefined, collected: {}, + done: done.promise, terminate, waitForExit, + } + const spawn = vi.fn(() => handle) + const client = new NativeClient({ subprocess: { spawn } }, { + helperPath: '/fixture/helper', cwd: '/fixture', timeoutMs: options.timeoutMs ?? 60_000, + maxRequestBytes: options.maxRequestBytes ?? 1024, + maxResponseBytes: options.maxResponseBytes ?? 1024, graceMs: 10, + }) + function finish() { done.resolve({ exitCode: 0, signal: null }); gone.resolve(true) } + function reply(value: unknown) { stdout.write(Buffer.from(JSON.stringify(value) + '\n')) } + function id(index = requests.length - 1) { + const request = requests[index] + if (!request) throw new Error('No native request was dispatched') + return request.id + } + onTestFinished(async () => { + finish() + await client.dispose().catch(() => undefined) + stdout.destroy() + stdin.destroy() + }) + return { client, requests, stdout, stdin, spawn, terminate, terminated: terminated.promise, waitForExit, done, gone, finish, reply, id } +} + +describe('native helper transport', () => { + it('accepts split UTF-8 replies and preserves native refusal codes without replacing the helper', async () => { + const f = fixture() + const first = f.client.request('listApps', {}, new AbortController().signal) + const bytes = Buffer.from(JSON.stringify({ id: f.id(), result: [{ name: '编辑器' }] }) + '\n') + for (const byte of bytes) f.stdout.write(Buffer.from([byte])) + expect(await first).toEqual([{ name: '编辑器' }]) + const second = f.client.request('click', { appHandle: 'app-one' }, new AbortController().signal) + const rejected = expect(second).rejects.toMatchObject({ name: 'NativeError', code: 'STALE_SNAPSHOT', message: 'STALE_SNAPSHOT: Refresh observation' }) + f.reply({ id: f.id(), error: { code: 'STALE_SNAPSHOT', message: 'Refresh observation' } }) + await rejected + expect(f.client.disposed).toBe(false) + expect(f.terminate).not.toHaveBeenCalled() + const third = f.client.request('getCapabilities', {}, new AbortController().signal) + f.reply({ id: f.id(), result: null }) + expect(await third).toBeNull() + expect(f.spawn).toHaveBeenCalledTimes(1) + }) + + it.each(['malformed', 'wrong-id', 'both', 'neither', 'empty-code', 'extra-field', 'invalid-utf8', 'extra-frame'] as const)('invalidates the helper on a %s response before releasing the caller', async (kind) => { + const f = fixture() + const pending = f.client.request('listApps', {}, new AbortController().signal) + const rejected = expect(pending).rejects.toMatchObject({ code: 'NATIVE_PROTOCOL' }) + if (kind === 'malformed') f.stdout.write('{broken}\n') + else if (kind === 'wrong-id') f.reply({ id: 'not-active', result: [] }) + else if (kind === 'both') f.reply({ id: f.id(), result: [], error: { code: 'failure', message: 'bad' } }) + else if (kind === 'neither') f.reply({ id: f.id() }) + else if (kind === 'empty-code') f.reply({ id: f.id(), error: { code: '', message: 'bad' } }) + else if (kind === 'extra-field') f.reply({ id: f.id(), result: [], extra: true }) + else if (kind === 'invalid-utf8') f.stdout.write(Buffer.concat([Buffer.from(`{"id":"${f.id()}","result":"`), Buffer.from([0xff]), Buffer.from('"}\n')])) + else f.stdout.write(JSON.stringify({ id: f.id(), result: [] }) + '\n{}\n') + expect(f.client.disposed).toBe(true) + expect(f.terminate).toHaveBeenCalledTimes(1) + f.finish() + await rejected + await expect(f.client.request('listApps', {}, new AbortController().signal)).rejects.toMatchObject({ code: 'NATIVE_CLOSED' }) + expect(f.requests).toHaveLength(1) + }) + + it('counts UTF-8 bytes and the newline against the response limit', async () => { + const f = fixture({ maxResponseBytes: 128 }) + const first = f.client.request('listApps', {}, new AbortController().signal) + const base = JSON.stringify({ id: f.id(), result: '' }) + '\n' + const value = 'x'.repeat(128 - Buffer.byteLength(base)) + f.reply({ id: f.id(), result: value }) + expect(await first).toBe(value) + const second = f.client.request('listApps', {}, new AbortController().signal) + const rejected = expect(second).rejects.toMatchObject({ code: 'NATIVE_RESPONSE_LIMIT' }) + const bytes = Buffer.from(JSON.stringify({ id: f.id(), result: value.slice(1) + 'é' }) + '\n') + expect(bytes).toHaveLength(129) + f.stdout.write(bytes.subarray(0, 90)) + f.stdout.write(bytes.subarray(90)) + f.finish() + await rejected + }) + + it('bounds request bytes before dispatch and keeps acquired handles after a rejected request', async () => { + const f = fixture({ maxRequestBytes: 128 }) + const envelope = JSON.stringify({ id: '0'.repeat(36), method: 'typeText', params: { text: '' } }) + '\n' + const text = 'x'.repeat(128 - Buffer.byteLength(envelope)) + const first = f.client.request('typeText', { text }, new AbortController().signal) + f.reply({ id: f.id(), result: { dispatched: true } }) + await expect(first).resolves.toEqual({ dispatched: true }) + await expect(f.client.request('typeText', { text: text.slice(1) + 'é' }, new AbortController().signal)).rejects.toMatchObject({ code: 'NATIVE_REQUEST_LIMIT' }) + expect(f.requests).toHaveLength(1) + expect(f.client.disposed).toBe(false) + expect(f.terminate).not.toHaveBeenCalled() + const next = f.client.request('listApps', {}, new AbortController().signal) + f.reply({ id: f.id(), result: [] }) + await expect(next).resolves.toEqual([]) + }) + + it('rejects an oversized unterminated response before retaining another chunk', async () => { + const f = fixture({ maxResponseBytes: 64 }) + const pending = f.client.request('listApps', {}, new AbortController().signal) + const rejected = expect(pending).rejects.toMatchObject({ code: 'NATIVE_RESPONSE_LIMIT' }) + f.stdout.write('x'.repeat(65)) + f.stdout.write('late output') + f.finish() + await rejected + expect(f.terminate).toHaveBeenCalledTimes(1) + }) + + it('cancellation waits for the complete managed process range', async () => { + const f = fixture() + const abort = new AbortController() + const pending = f.client.request('getState', {}, abort.signal) + let settled = false + const observed = pending.then(() => { settled = true }, (error: unknown) => { settled = true; expect(error).toMatchObject({ code: 'NATIVE_CANCELLED' }) }) + abort.abort() + f.done.resolve({ exitCode: 0, signal: null }) + await Promise.resolve() + await Promise.resolve() + expect(f.terminate).toHaveBeenCalledTimes(1) + expect(settled).toBe(false) + f.gone.resolve(true) + await observed + expect(f.stdout.destroyed).toBe(true) + expect(f.stdin.destroyed).toBe(true) + }) + + it('rejects a pre-cancelled operation without writing or invalidating acquired handles', async () => { + const f = fixture() + await expect(f.client.request('click', {}, AbortSignal.abort())).rejects.toMatchObject({ code: 'NATIVE_CANCELLED' }) + expect(f.requests).toEqual([]) + expect(f.client.disposed).toBe(false) + }) + + it('refuses overlapping operations without corrupting the active request', async () => { + const f = fixture() + const first = f.client.request('getState', {}, new AbortController().signal) + await expect(f.client.request('click', {}, new AbortController().signal)).rejects.toMatchObject({ code: 'NATIVE_BUSY' }) + expect(f.requests).toHaveLength(1) + f.reply({ id: f.id(), result: { snapshotId: 'one' } }) + expect(await first).toEqual({ snapshotId: 'one' }) + }) + + it('cancellation after a reply but before result delivery still invalidates the helper', async () => { + const f = fixture() + const abort = new AbortController() + const pending = f.client.request('click', {}, abort.signal) + const rejected = expect(pending).rejects.toMatchObject({ code: 'NATIVE_CANCELLED' }) + f.reply({ id: f.id(), result: { dispatched: true } }) + abort.abort() + f.finish() + await rejected + }) + + it('expires the configured operation deadline and waits for shutdown', async () => { + vi.useFakeTimers() + try { + const f = fixture({ timeoutMs: 25 }) + const pending = f.client.request('getState', {}, new AbortController().signal) + const rejected = expect(pending).rejects.toMatchObject({ code: 'NATIVE_TIMEOUT' }) + await vi.advanceTimersByTimeAsync(25) + expect(f.client.disposed).toBe(true) + f.finish() + await rejected + expect(vi.getTimerCount()).toBe(0) + } finally { vi.useRealTimers() } + }) + + it.each(['stdout-error', 'stdin-error', 'stdout-end', 'stdout-close', 'process-exit', 'process-rejection'] as const)('contains %s while an operation is pending', async (kind) => { + const f = fixture() + const pending = f.client.request('getState', {}, new AbortController().signal) + const code = kind.endsWith('-error') ? 'NATIVE_IO' : kind.startsWith('stdout-') ? 'NATIVE_EOF' : 'NATIVE_EXITED' + const rejected = expect(pending).rejects.toMatchObject({ code }) + if (kind === 'stdout-error') f.stdout.emit('error', new Error('read failed')) + else if (kind === 'stdin-error') f.stdin.emit('error', new Error('write failed')) + else if (kind === 'stdout-end') f.stdout.emit('end') + else if (kind === 'stdout-close') f.stdout.emit('close') + else if (kind === 'process-exit') f.done.resolve({ exitCode: 1, signal: null }) + else f.done.reject(new Error('spawn failed')) + f.gone.resolve(true) + if (!kind.startsWith('process-')) f.done.resolve({ exitCode: 1, signal: null }) + await rejected + expect(f.terminate).toHaveBeenCalledTimes(1) + }) + + it.each(['throw', 'callback'] as const)('invalidates uncertain dispatch after a write %s failure', async (kind) => { + const failure = new Error('pipe failed') + const f = fixture(kind === 'callback' ? { writeError: failure } : {}) + if (kind === 'throw') vi.spyOn(f.stdin, 'write').mockImplementation(() => { throw failure }) + const pending = f.client.request('click', {}, new AbortController().signal) + const rejected = expect(pending).rejects.toMatchObject({ code: 'NATIVE_IO', cause: failure }) + await f.terminated + f.finish() + await rejected + expect(f.client.disposed).toBe(true) + expect(f.terminate).toHaveBeenCalledTimes(1) + }) + + it.each([{ stdin: false }, { stdout: false }, { stdin: false, stdout: false }])('cleans up an incomplete process handle %j', async (options) => { + const f = fixture(options) + const pending = f.client.request('listApps', {}, new AbortController().signal) + const rejected = expect(pending).rejects.toMatchObject({ code: 'NATIVE_IO' }) + f.finish() + await rejected + expect(f.requests).toEqual([]) + }) + + it('rejects unsolicited replies and shares one quiescent disposal across callers', async () => { + const f = fixture() + f.reply({ id: 'unsolicited', result: [] }) + const first = f.client.dispose() + const second = f.client.dispose() + expect(second).toBe(first) + expect(f.waitForExit).toHaveBeenCalledTimes(1) + f.finish() + await first + expect(() => { + f.stdout.emit('error', new Error('late read error')) + f.stdin.emit('error', new Error('late write error')) + }).not.toThrow() + expect(f.terminate).toHaveBeenCalledTimes(1) + }) + + it('rejects pending work on disposal and releases streams even when range observation fails', async () => { + const f = fixture() + const pending = f.client.request('getState', {}, new AbortController().signal) + const failure = new Error('cannot observe range') + const rejected = expect(pending).rejects.toBe(failure) + const disposing = f.client.dispose() + const disposalRejected = expect(disposing).rejects.toBe(failure) + f.gone.reject(failure) + await Promise.resolve() + await Promise.resolve() + expect(f.stdout.destroyed).toBe(false) + f.done.resolve({ exitCode: 0, signal: null }) + await disposalRejected + await rejected + expect(f.stdout.destroyed).toBe(true) + expect(f.stdin.destroyed).toBe(true) + }) + + it('keeps native codes available to callers and in the model-visible diagnostic', () => { + const error = new NativeError('PERMISSION_DENIED', 'Enable Accessibility') + expect(error.code).toBe('PERMISSION_DENIED') + expect(error.message).toBe('PERMISSION_DENIED: Enable Accessibility') + }) +})