From 286a8b168e1e4e536c75800d0b9d2523213f7991 Mon Sep 17 00:00:00 2001 From: imccyu <276526105+imccyu@users.noreply.github.com> Date: Fri, 7 Aug 2026 12:53:16 +0800 Subject: [PATCH] fix(connection): route fixture calls through remote semantics --- .../client/connection/src/client/fixture.ts | 266 +++++++++++++----- .../client/connection/src/client/index.ts | 7 +- packages/client/connection/src/client/rpc.ts | 12 - .../connection/tests/client-apply.spec.ts | 32 ++- 4 files changed, 234 insertions(+), 83 deletions(-) diff --git a/packages/client/connection/src/client/fixture.ts b/packages/client/connection/src/client/fixture.ts index e13c0a19f6..1a9841aece 100644 --- a/packages/client/connection/src/client/fixture.ts +++ b/packages/client/connection/src/client/fixture.ts @@ -36,6 +36,7 @@ import type { import type { RequestPayload, ResponseValue, RpcMethodMap } from '@deepseek-ai/dsh-host-apiproxy/api' import { AbstractApiClient, RpcId, SESSION_SEARCH_RESULT_LIMIT } from './api.ts' import { randomUuid } from './random-uuid.ts' +import type { ClientConnectionRpc } from '../rpc.ts' /** The fake carrier mints like a real one (business code never mints). */ function rpcRequest

(payload: P): RpcRequest

{ @@ -1329,6 +1330,16 @@ class FxInbox implements StreamConn { * @returns an ApiProxy backed entirely by in-memory state — no host process, no network. */ export function createFixtureApi(options: FixtureOptions = {}): ApiProxy { + return createFixtureWorld(options).api +} + +interface FixtureWorld { + readonly api: ApiProxy + readonly rpc: ClientConnectionRpc +} + +/** Build the fixture's legacy API and Remote RPC faces over one state graph. */ +function createFixtureWorld(options: FixtureOptions): FixtureWorld { // The resident fixture sessions all carry history, so none of them is blank. const sessions: SessionSummary[] = options.empty ? [] : [ { sessionId: sid('fx-alpha'), updatedAt: Date.now(), running: true, blank: false, cwd: '/tmp/fixture' }, @@ -1507,31 +1518,136 @@ export function createFixtureApi(options: FixtureOptions = {}): ApiProxy { return backscanGoal(log) as FxGoalProjection } - /** Shared CAS mutation path of the goal verbs (undefined next = invalid transition). */ - const fxMutateGoal = ( - request: RpcRequest<{ sessionId: SessionId; ref: { id: string; revision: number } }>, - ref: { id: string; revision: number }, + type FxGoalRef = { id: string; revision: number } + type FxGoalView = FxGoalProjection['goal'] & { + roundsStarted: number + createdAt: number + updatedAt: number + activation: 'armed' | 'disarmed' + } + + const goalFailure = (message: string): RpcResult => ({ + ok: false, + error: { code: 'internal', message, details: {} }, + }) + + const requireGoalSession = (id: SessionId): RpcResult | undefined => ( + summaryOf(id) === undefined + ? { ok: false, error: { code: 'session-not-found', message: `no session ${id}`, details: { sessionId: id } } } + : undefined + ) + + const goalView = (projection: FxGoalProjection): FxGoalView => ({ + ...projection.goal, + roundsStarted: projection.roundsStarted, + createdAt: projection.createdAt, + updatedAt: projection.updatedAt, + activation: projection.goal.phase === 'active' ? 'armed' : 'disarmed', + }) + + /** Canonical fixture implementation of the generated Goal Remote contract. */ + const goalRemotes = { + create(id: SessionId, request: { objective: string; maxGoalRounds?: number }): RpcResult<{ ref: FxGoalRef }> { + const missing = requireGoalSession(id) + if (missing !== undefined) return missing + const current = backscanGoal(logOf(id)) + if (current !== null && current.goal.phase !== 'complete') { + return goalFailure(`goal "${current.goal.id}" already exists`) + } + const now = Date.now() + const projection = appendGoalChange(id, { + kind: 'goal/change', version: 1, operation: 'create', + goal: { + id: `fx-goal-${logOf(id).length}`, + revision: 1, + objective: request.objective, + phase: 'active', + maxGoalRounds: request.maxGoalRounds ?? 256, + }, + roundsStarted: 0, createdAt: now, updatedAt: now, + }) + return { ok: true, value: { ref: { id: projection.goal.id, revision: projection.goal.revision } } } + }, + edit(id: SessionId, ref: FxGoalRef, request: { objective?: string; maxGoalRounds?: number }): RpcResult { + return mutateGoal(id, ref, current => ({ + ...current.goal, + revision: current.goal.revision + 1, + ...request.objective === undefined ? {} : { objective: request.objective }, + ...request.maxGoalRounds === undefined ? {} : { maxGoalRounds: request.maxGoalRounds }, + })) + }, + pause(id: SessionId, ref: FxGoalRef): RpcResult { + return mutateGoal(id, ref, current => ( + current.goal.phase === 'active' + ? { ...current.goal, revision: current.goal.revision + 1, phase: 'paused' } + : undefined + )) + }, + resume(id: SessionId, ref: FxGoalRef): RpcResult { + return mutateGoal(id, ref, current => ( + current.goal.phase === 'paused' || current.goal.phase === 'blocked' || current.goal.phase === 'active' + ? { ...current.goal, revision: current.goal.revision + 1, phase: 'active' } + : undefined + )) + }, + complete(id: SessionId, ref: FxGoalRef): RpcResult { + return mutateGoal(id, ref, current => ( + current.goal.phase === 'complete' + ? undefined + : { ...current.goal, revision: current.goal.revision + 1, phase: 'complete' } + )) + }, + clear(id: SessionId, ref: FxGoalRef): RpcResult { + const missing = requireGoalSession(id) + if (missing !== undefined) return missing + const current = backscanGoal(logOf(id)) + if (current === null || current.goal.id !== ref.id || current.goal.revision !== ref.revision) { + return goalFailure('stale or missing goal revision') + } + const tombstone = { id: current.goal.id, revision: current.goal.revision + 1 } + appendGoalChange(id, { + kind: 'goal/change', version: 1, operation: 'clear', cleared: tombstone, clearedAt: Date.now(), + }) + return { ok: true, value: tombstone } + }, + } + + /** Shared CAS mutation path behind the canonical Remote verbs. */ + function mutateGoal( + id: SessionId, + ref: FxGoalRef, next: (current: FxGoalProjection) => FxGoalProjection['goal'] | undefined, - ): Promise> => { - const missing = requireSession(request) + ): RpcResult { + const missing = requireGoalSession(id) if (missing !== undefined) return missing - const id = request.payload.sessionId const current = backscanGoal(logOf(id)) if (current === null || current.goal.id !== ref.id || current.goal.revision !== ref.revision) { - return err(request, { code: 'internal', message: 'stale or missing goal revision', details: { goalCode: 'GOAL_STALE_REVISION' } }) + return goalFailure('stale or missing goal revision') } const goal = next(current) if (goal === undefined) { - return err(request, { code: 'internal', message: `invalid goal transition from "${current.goal.phase}"`, details: { goalCode: 'GOAL_INVALID_TRANSITION' } }) + return goalFailure(`invalid goal transition from "${current.goal.phase}"`) } const projection = appendGoalChange(id, { kind: 'goal/change', version: 1, operation: goal.phase === current.goal.phase ? 'edit' : goal.phase === 'paused' ? 'pause' : goal.phase === 'active' ? 'resume' : 'complete', goal, roundsStarted: current.roundsStarted, createdAt: current.createdAt, updatedAt: Date.now(), }) - return ok(request, { ref: { id: projection.goal.id as never, revision: projection.goal.revision } }) + return { ok: true, value: goalView(projection) } } + const mapGoalResult = (result: RpcResult, map: (value: T) => U): RpcResult => ( + result.ok ? { ok: true, value: map(result.value) } : result + ) + + const goalRefResult = (result: RpcResult): RpcResult<{ ref: { id: never; revision: number } }> => ( + mapGoalResult(result, view => ({ ref: { id: view.id as never, revision: view.revision } })) + ) + + const legacyGoalResponse = (request: RpcRequest

, result: RpcResult): Promise> => ( + Promise.resolve({ rpcId: request.rpcId, result }) + ) + /** At most one in-flight replay per session; cancel clears it. */ const replays = new Map; finish(aborted: boolean): void }>() @@ -1777,7 +1893,7 @@ export function createFixtureApi(options: FixtureOptions = {}): ApiProxy { replays.set(id, { timer: setTimeout(tick, 80), finish }) } - return { + const api: ApiProxy = { sessions: { list: request => ok(request, { items: [...sessions].sort((a, b) => b.updatedAt - a.updatedAt) }), search: (request, signal) => { @@ -2334,60 +2450,44 @@ export function createFixtureApi(options: FixtureOptions = {}): ApiProxy { }, }, goals: { - // Mutation-only mirror of the host handlers: each verb CAS-checks the - // projected current goal, appends the whole-value change (the mux - // stream and projection frame ride the shared append path), and - // acknowledges with the new ref only. - create: (request) => { - const missing = requireSession(request) - if (missing !== undefined) return missing - const id = request.payload.sessionId - const current = backscanGoal(logOf(id)) - if (current !== null && current.goal.phase !== 'complete') { - return err(request, { code: 'internal', message: `goal "${current.goal.id}" already exists`, details: { goalCode: 'GOAL_ALREADY_EXISTS' } }) - } - const projection = appendGoalChange(id, { - kind: 'goal/change', version: 1, operation: 'create', - goal: { id: `fx-goal-${logOf(id).length}`, revision: 1, objective: request.payload.objective, phase: 'active', maxGoalRounds: request.payload.maxGoalRounds ?? 256 }, - roundsStarted: 0, createdAt: Date.now(), updatedAt: Date.now(), - }) - return ok(request, { ref: { id: projection.goal.id as never, revision: projection.goal.revision } }) - }, - edit: request => fxMutateGoal(request, request.payload.ref, current => ({ - ...current.goal, - revision: current.goal.revision + 1, - ...request.payload.objective === undefined ? {} : { objective: request.payload.objective }, - ...request.payload.maxGoalRounds === undefined ? {} : { maxGoalRounds: request.payload.maxGoalRounds }, - })), - pause: request => fxMutateGoal(request, request.payload.ref, current => ( - current.goal.phase === 'active' - ? { ...current.goal, revision: current.goal.revision + 1, phase: 'paused' } - : undefined - )), - resume: request => fxMutateGoal(request, request.payload.ref, current => ( - current.goal.phase === 'paused' || current.goal.phase === 'blocked' || current.goal.phase === 'active' - ? { ...current.goal, revision: current.goal.revision + 1, phase: 'active' } - : undefined - )), - complete: request => fxMutateGoal(request, request.payload.ref, current => ( - current.goal.phase === 'complete' - ? undefined - : { ...current.goal, revision: current.goal.revision + 1, phase: 'complete' } - )), - clear: (request) => { - const missing = requireSession(request) - if (missing !== undefined) return missing - const id = request.payload.sessionId - const current = backscanGoal(logOf(id)) - if (current === null || current.goal.id !== request.payload.ref.id || current.goal.revision !== request.payload.ref.revision) { - return err(request, { code: 'internal', message: 'stale or missing goal revision', details: { goalCode: 'GOAL_STALE_REVISION' } }) - } - appendGoalChange(id, { - kind: 'goal/change', version: 1, operation: 'clear', - cleared: { id: current.goal.id, revision: current.goal.revision + 1 }, clearedAt: Date.now(), - }) - return ok(request, { cleared: true as const }) - }, + // Compatibility face only: old API Proxy payloads and acknowledgements + // adapt to the canonical fixture Remote implementation above. + create: request => legacyGoalResponse( + request, + mapGoalResult( + goalRemotes.create(request.payload.sessionId, { + objective: request.payload.objective, + ...request.payload.maxGoalRounds === undefined ? {} : { maxGoalRounds: request.payload.maxGoalRounds }, + }), + value => ({ ref: { id: value.ref.id as never, revision: value.ref.revision } }), + ), + ), + edit: request => legacyGoalResponse( + request, + goalRefResult(goalRemotes.edit(request.payload.sessionId, request.payload.ref, { + ...request.payload.objective === undefined ? {} : { objective: request.payload.objective }, + ...request.payload.maxGoalRounds === undefined ? {} : { maxGoalRounds: request.payload.maxGoalRounds }, + })), + ), + pause: request => legacyGoalResponse( + request, + goalRefResult(goalRemotes.pause(request.payload.sessionId, request.payload.ref)), + ), + resume: request => legacyGoalResponse( + request, + goalRefResult(goalRemotes.resume(request.payload.sessionId, request.payload.ref)), + ), + complete: request => legacyGoalResponse( + request, + goalRefResult(goalRemotes.complete(request.payload.sessionId, request.payload.ref)), + ), + clear: request => legacyGoalResponse( + request, + mapGoalResult( + goalRemotes.clear(request.payload.sessionId, request.payload.ref), + () => ({ cleared: true as const }), + ), + ), }, events: { async *mux(_request, signal) { @@ -2548,6 +2648,36 @@ export function createFixtureApi(options: FixtureOptions = {}): ApiProxy { return Promise.resolve({ accepted: true }) }, } + + const rpc: ClientConnectionRpc = { + call(channel, endpoint, payload) { + if (channel !== '/api') { + return Promise.reject(new Error(`fixture connection RPC channel ${JSON.stringify(channel)} is unavailable`)) + } + const args = (payload as { + args: { + agentId: SessionId + ref?: { id: string; revision: number } + request?: { objective?: string; maxGoalRounds?: number } + } + }).args + const sessionId = args.agentId + switch (endpoint) { + case 'goals/create': return Promise.resolve(goalRemotes.create(sessionId, { + objective: args.request?.objective as string, + ...args.request?.maxGoalRounds === undefined ? {} : { maxGoalRounds: args.request.maxGoalRounds }, + })) + case 'goals/edit': return Promise.resolve(goalRemotes.edit(sessionId, args.ref as FxGoalRef, args.request ?? {})) + case 'goals/pause': return Promise.resolve(goalRemotes.pause(sessionId, args.ref as FxGoalRef)) + case 'goals/resume': return Promise.resolve(goalRemotes.resume(sessionId, args.ref as FxGoalRef)) + case 'goals/complete': return Promise.resolve(goalRemotes.complete(sessionId, args.ref as FxGoalRef)) + case 'goals/clear': return Promise.resolve(goalRemotes.clear(sessionId, args.ref as FxGoalRef)) + default: + return Promise.reject(new Error(`fixture connection RPC endpoint ${JSON.stringify(endpoint)} is unavailable`)) + } + }, + } + return { api, rpc } } /** @@ -2559,10 +2689,14 @@ export function createFixtureApi(options: FixtureOptions = {}): ApiProxy { */ export class FixtureApiClient extends AbstractApiClient { private readonly api: ApiProxy + /** Generic Remote caller backed by the same in-memory state as the legacy fixture API. */ + readonly rpc: ClientConnectionRpc constructor() { super() - this.api = createFixtureApi(fixtureOptionsFromLocation()) + const world = createFixtureWorld(fixtureOptionsFromLocation()) + this.api = world.api + this.rpc = world.rpc } protected doFetch(): Promise { diff --git a/packages/client/connection/src/client/index.ts b/packages/client/connection/src/client/index.ts index 521e54160e..c2a6668d46 100644 --- a/packages/client/connection/src/client/index.ts +++ b/packages/client/connection/src/client/index.ts @@ -8,7 +8,7 @@ import type { IApiClient } from './api.ts' import { ConnectionController, type ConnectionConfig, type ConnectionSinks, type ConnectionState } from './connection.ts' import { FixtureApiClient } from './fixture.ts' import { WebApiClient } from './web-api-client.ts' -import { createUnavailableConnectionRpc, createWebConnectionRpc } from './rpc.ts' +import { createWebConnectionRpc } from './rpc.ts' import { isLoopbackHostname } from '../loopback-hostname.ts' import type { ClientConnectionRpc } from '../rpc.ts' @@ -74,8 +74,9 @@ export interface ConnectionHandle { export function apply(ctx: Context): void { const pageLocation = typeof location === 'undefined' ? undefined : location const fixture = pageLocation !== undefined && new URLSearchParams(pageLocation.search).has('fixture') - const api: IApiClient = fixture ? new FixtureApiClient() : new WebApiClient() - const rpc = fixture ? createUnavailableConnectionRpc() : createWebConnectionRpc() + const fixtureClient = fixture ? new FixtureApiClient() : undefined + const api: IApiClient = fixtureClient ?? new WebApiClient() + const rpc = fixtureClient?.rpc ?? createWebConnectionRpc() let started = false const handle: ConnectionHandle = { api, diff --git a/packages/client/connection/src/client/rpc.ts b/packages/client/connection/src/client/rpc.ts index 7883f2a9d3..f8bacb1553 100644 --- a/packages/client/connection/src/client/rpc.ts +++ b/packages/client/connection/src/client/rpc.ts @@ -48,18 +48,6 @@ export function createWebConnectionRpc(): ClientConnectionRpc { } } -/** - * Create the fixture-mode caller, where no Host Remote registry exists. - * @returns caller that rejects every generic Remote invocation. - */ -export function createUnavailableConnectionRpc(): ClientConnectionRpc { - return { - call(channel, endpoint) { - return Promise.reject(new Error(`connection RPC ${channel}/${endpoint} is unavailable in fixture mode`)) - }, - } -} - function resolveBase(): string { const location = (globalThis as { location?: { origin?: string } }).location return location?.origin !== undefined && location.origin !== 'null' ? location.origin : INTERNAL_BASE diff --git a/packages/client/connection/tests/client-apply.spec.ts b/packages/client/connection/tests/client-apply.spec.ts index 41e8e9b0e2..9d9bbd2f26 100644 --- a/packages/client/connection/tests/client-apply.spec.ts +++ b/packages/client/connection/tests/client-apply.spec.ts @@ -285,9 +285,37 @@ describe('connection client apply', () => { } }) - it('keeps generic Remote calls unavailable in the client-only fixture', async () => { + it('carries Goal Remotes over the same state as the client-only fixture API', async () => { ;(globalThis as Win).location = { hostname: 'localhost', search: '?fixture' } const handle = await mount() - await expect(handle.rpc.call('/api', 'goals/create', {})).rejects.toThrow(/unavailable in fixture mode/) + const created = await handle.rpc.call('/api', 'goals/create', { + args: { agentId: 'fx-alpha', request: { objective: 'fixture remote' } }, + }) + expect(created).toMatchObject({ ok: true, value: { ref: { revision: 1 } } }) + if (!created.ok) throw new Error('fixture Goal create failed') + const ref = (created.value as { ref: { id: string; revision: number } }).ref + const edited = await handle.rpc.call('/api', 'goals/edit', { + args: { agentId: 'fx-alpha', ref, request: { objective: 'edited fixture remote' } }, + }) + expect(edited).toMatchObject({ ok: true, value: { objective: 'edited fixture remote', revision: 2 } }) + const editedRef = { id: ref.id, revision: 2 } + const paused = await handle.rpc.call('/api', 'goals/pause', { + args: { agentId: 'fx-alpha', ref: editedRef }, + }) + expect(paused).toMatchObject({ ok: true, value: { phase: 'paused', activation: 'disarmed', revision: 3 } }) + const resumed = await handle.rpc.call('/api', 'goals/resume', { + args: { agentId: 'fx-alpha', ref: { id: ref.id, revision: 3 } }, + }) + expect(resumed).toMatchObject({ ok: true, value: { phase: 'active', activation: 'armed', revision: 4 } }) + const completed = await handle.rpc.call('/api', 'goals/complete', { + args: { agentId: 'fx-alpha', ref: { id: ref.id, revision: 4 } }, + }) + expect(completed).toMatchObject({ ok: true, value: { phase: 'complete', activation: 'disarmed', revision: 5 } }) + await expect(handle.rpc.call('/api', 'goals/clear', { + args: { agentId: 'fx-alpha', ref: { id: ref.id, revision: 5 } }, + })).resolves.toEqual({ ok: true, value: { id: ref.id, revision: 6 } }) + await expect(handle.rpc.call('/other', 'goals/create', {})).rejects.toThrow(/channel.*unavailable/) + await expect(handle.rpc.call('/api', 'unknown/read', { args: { agentId: 'fx-alpha' } })) + .rejects.toThrow(/endpoint.*unavailable/) }) })