From 090a72a910a9631027768873f76fc4d19549b2e5 Mon Sep 17 00:00:00 2001 From: leyoonafr Date: Sat, 29 Aug 2026 23:34:24 +0800 Subject: [PATCH 1/4] feat: complete protocol endpoint dispatch (#5) --- apps/server/src/loopback-server.ts | 62 ++- apps/server/src/protocol-dispatch.test.ts | 609 ++++++++++++++++++++++ apps/server/src/protocol-dispatch.ts | 379 ++++++++++++++ apps/server/src/server.ts | 4 + 4 files changed, 1032 insertions(+), 22 deletions(-) create mode 100644 apps/server/src/protocol-dispatch.test.ts create mode 100644 apps/server/src/protocol-dispatch.ts diff --git a/apps/server/src/loopback-server.ts b/apps/server/src/loopback-server.ts index 6dc00a9..436a987 100644 --- a/apps/server/src/loopback-server.ts +++ b/apps/server/src/loopback-server.ts @@ -14,9 +14,14 @@ import { sseInvalidationSchema, type HealthResponse, type ProtocolError, - type SessionMetadata, } from '@codex-git/protocol'; +import { + createProtocolDispatcher, + type ProtocolDispatcher, + type ProtocolHandlers, +} from './protocol-dispatch.js'; + const healthResponse = { product: 'codex-git', status: 'ok', @@ -34,20 +39,6 @@ const endpointMethods = new Map([ ['session', 'GET'], ['snapshot', 'GET'], ]); -const sessionMetadata = { - protocolVersion: PROTOCOL_VERSION, - capabilities: { - branchSearch: false, - commands: false, - commitDrafts: false, - diff: false, - events: true, - nativeActions: false, - operationRecovery: false, - }, - limits: PROTOCOL_LIMITS, -} satisfies SessionMetadata; - export interface LoopbackAddress { readonly host: typeof loopbackHost; readonly port: number; @@ -65,6 +56,7 @@ export interface LoopbackServer { export interface LoopbackServerOptions { readonly allowedOrigins?: readonly string[]; + readonly handlers?: ProtocolHandlers; readonly randomBytes?: (length: number) => Uint8Array; } @@ -78,6 +70,7 @@ export async function startLoopbackServer( const token = Buffer.from(tokenBytes).toString('hex'); const instancePrefix = `/instance/${token}/v1`; const allowedOrigins = new Set(options.allowedOrigins ?? []); + const dispatcher = createProtocolDispatcher(options.handlers); const events = createEventBroker(); const server = createServer((request, response) => { void handleRequest( @@ -86,6 +79,7 @@ export async function startLoopbackServer( token, instancePrefix, allowedOrigins, + dispatcher, events, ).catch(() => { if (!response.headersSent) { @@ -138,6 +132,7 @@ async function handleRequest( token: string, instancePrefix: string, allowedOrigins: ReadonlySet, + dispatcher: ProtocolDispatcher, events: EventBroker, ): Promise { if (request.socket.remoteAddress !== loopbackHost) { @@ -234,6 +229,7 @@ async function handleRequest( return; } + let body: Buffer | undefined; if (expectedMethod === 'POST' || expectedMethod === 'PUT') { const mediaType = request.headers['content-type']?.split(';', 1)[0]?.trim(); if (mediaType !== 'application/json') { @@ -249,23 +245,26 @@ async function handleRequest( ); return; } - const body = await readBoundedBody( + const requestBody = await readBoundedBody( request, - sessionMetadata.limits.requestBodyBytes, + dispatcher.sessionMetadata.limits.requestBodyBytes, ); - if (body === null) { + if (requestBody === null) { sendProtocolError( response, 413, { code: 'body_too_large', - details: { limitBytes: sessionMetadata.limits.requestBodyBytes }, + details: { + limitBytes: dispatcher.sessionMetadata.limits.requestBodyBytes, + }, message: 'The request body exceeds the configured limit.', }, origin, ); return; } + body = requestBody; } if (endpoint === 'events') { @@ -274,7 +273,26 @@ async function handleRequest( } if (endpoint === 'session') { - sendJson(response, 200, sessionMetadata, origin); + sendJson(response, 200, dispatcher.sessionMetadata, origin); + return; + } + + try { + const dispatched = await dispatcher.dispatch(endpoint, body); + if (dispatched !== undefined) { + sendJson(response, dispatched.status, dispatched.value, origin); + return; + } + } catch { + sendProtocolError( + response, + 500, + { + code: 'internal_error', + message: 'The protocol request could not be completed.', + }, + origin, + ); return; } @@ -505,7 +523,7 @@ function sendJson( origin?: string, ): void { const body = JSON.stringify(value); - if (Buffer.byteLength(body) > sessionMetadata.limits.diffOutputBytes) { + if (Buffer.byteLength(body) > PROTOCOL_LIMITS.diffOutputBytes) { response.writeHead(507, { ...(origin === undefined ? {} @@ -516,7 +534,7 @@ function sendJson( JSON.stringify({ error: { code: 'output_too_large', - details: { limitBytes: sessionMetadata.limits.diffOutputBytes }, + details: { limitBytes: PROTOCOL_LIMITS.diffOutputBytes }, message: 'The protocol response exceeds the configured limit.', }, }), diff --git a/apps/server/src/protocol-dispatch.test.ts b/apps/server/src/protocol-dispatch.test.ts new file mode 100644 index 0000000..13564e6 --- /dev/null +++ b/apps/server/src/protocol-dispatch.test.ts @@ -0,0 +1,609 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; + +import { + branchSearchResultSchema, + commitDraftSchema, + PROTOCOL_VERSION_HEADER, + diffResultSchema, + operationReceiptSchema, + operationResultSchema, + repositorySnapshotSchema, +} from '@codex-git/protocol'; + +import { startLoopbackServer, type LoopbackServer } from './server.js'; + +const servers: LoopbackServer[] = []; +const headers = { + origin: 'null', + [PROTOCOL_VERSION_HEADER]: '1', +} as const; + +afterEach(async () => { + await Promise.all(servers.splice(0).map((server) => server.close())); +}); + +describe('protocol HTTP dispatch', () => { + it('routes snapshots through an installed handler without advertising absent handlers', async () => { + const snapshot = repositorySnapshotSchema.parse({ + repositoryId: 'repository_0123456789abcdef0123456789abcdef', + repositoryRevision: 4, + topologyRevision: 2, + refsRevision: 3, + refresh: { kind: 'current' }, + worktrees: [], + remotes: [], + operations: [], + }); + const handleSnapshot = vi.fn(() => snapshot); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { snapshot: handleSnapshot }, + }); + servers.push(server); + + const [sessionResponse, snapshotResponse] = await Promise.all([ + fetch(server.sessionUrl, { headers }), + fetch(endpointUrl(server, 'snapshot'), { headers }), + ]); + + expect({ + capabilities: ( + (await sessionResponse.json()) as { + capabilities: Record; + } + ).capabilities, + snapshot: await snapshotResponse.json(), + snapshotStatus: snapshotResponse.status, + }).toEqual({ + capabilities: { + branchSearch: false, + commands: false, + commitDrafts: false, + diff: false, + events: true, + nativeActions: false, + operationRecovery: false, + }, + snapshot, + snapshotStatus: 200, + }); + expect(handleSnapshot).toHaveBeenCalledOnce(); + }); + + it('rejects malformed diff requests before calling the installed handler', async () => { + const handleDiff = vi.fn(() => + diffResultSchema.parse({ + kind: 'binary' as const, + fileId: 'file_0123456789abcdef0123456789abcdef', + baseline: 'index_to_working_tree' as const, + byteCount: 42, + }), + ); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { diff: handleDiff }, + }); + servers.push(server); + + const response = await fetch(endpointUrl(server, 'diff'), { + method: 'POST', + headers: { ...headers, 'content-type': 'application/json' }, + body: JSON.stringify({ + fileId: 'file_0123456789abcdef0123456789abcdef', + path: '/Users/example/repository/private.txt', + }), + }); + const session = (await ( + await fetch(server.sessionUrl, { headers }) + ).json()) as { capabilities: { diff: boolean } }; + + expect({ + capability: session.capabilities.diff, + response: await response.json(), + status: response.status, + }).toEqual({ + capability: true, + response: { + error: { + code: 'invalid_payload', + details: expect.objectContaining({ issues: expect.any(Array) }), + message: 'The protocol payload is invalid.', + }, + }, + status: 400, + }); + expect(handleDiff).not.toHaveBeenCalled(); + }); + + it('does not serialize a handler response that fails runtime validation', async () => { + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { + diff: async () => + ({ + kind: 'text', + fileId: 'file_0123456789abcdef0123456789abcdef', + baseline: 'head_to_index', + content: 'token=fixture-handler-secret', + lineCount: 20_001, + }) as never, + }, + }); + servers.push(server); + + const response = await fetch(endpointUrl(server, 'diff'), { + method: 'POST', + headers: { ...headers, 'content-type': 'application/json' }, + body: JSON.stringify({ + fileId: 'file_0123456789abcdef0123456789abcdef', + }), + }); + const body = JSON.stringify(await response.json()); + + expect({ body: JSON.parse(body), status: response.status }).toEqual({ + body: { + error: { + code: 'internal_error', + message: 'The protocol handler returned an invalid response.', + }, + }, + status: 500, + }); + expect(body).not.toContain('fixture-handler-secret'); + }); + + it('rejects a valid diff response for a different opaque File ID', async () => { + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { + diff: async () => + diffResultSchema.parse({ + kind: 'binary', + fileId: 'file_bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb', + baseline: 'head_to_index', + byteCount: 24, + }), + }, + }); + servers.push(server); + + const response = await postJson(server, 'diff', { + fileId: 'file_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', + }); + + expect({ body: await response.json(), status: response.status }).toEqual({ + body: { + error: { + code: 'internal_error', + message: 'The protocol handler returned an invalid response.', + }, + }, + status: 500, + }); + }); + + it('dispatches branch search with validated opaque targets', async () => { + const result = branchSearchResultSchema.parse({ + refsRevision: 7, + candidates: [ + { + refId: 'ref_0123456789abcdef0123456789abcdef', + kind: 'local', + displayName: 'feature/protocol', + occupiedBy: null, + }, + ], + }); + const searchBranches = vi.fn(() => result); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { branchSearch: searchBranches }, + }); + servers.push(server); + + const response = await fetch(endpointUrl(server, 'branches'), { + method: 'POST', + headers: { ...headers, 'content-type': 'application/json' }, + body: JSON.stringify({ + worktreeId: 'worktree_0123456789abcdef0123456789abcdef', + query: 'feature', + }), + }); + const session = (await ( + await fetch(server.sessionUrl, { headers }) + ).json()) as { capabilities: { branchSearch: boolean } }; + + expect({ + capability: session.capabilities.branchSearch, + result: await response.json(), + status: response.status, + }).toEqual({ capability: true, result, status: 200 }); + expect(searchBranches).toHaveBeenCalledWith({ + worktreeId: 'worktree_0123456789abcdef0123456789abcdef', + query: 'feature', + }); + }); + + it('dispatches a validated Commit Draft update through the installed handler', async () => { + const draft = commitDraftSchema.parse({ + worktreeId: 'worktree_0123456789abcdef0123456789abcdef', + revision: 9, + text: 'Describe the protocol change', + }); + const updateDraft = vi.fn(() => draft); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { commitDrafts: updateDraft }, + }); + servers.push(server); + const request = { + worktreeId: 'worktree_0123456789abcdef0123456789abcdef', + expectedRevision: 8, + update: { kind: 'set', text: 'Describe the protocol change' }, + } as const; + + const response = await fetch(endpointUrl(server, 'draft'), { + method: 'PUT', + headers: { ...headers, 'content-type': 'application/json' }, + body: JSON.stringify(request), + }); + const session = (await ( + await fetch(server.sessionUrl, { headers }) + ).json()) as { capabilities: { commitDrafts: boolean } }; + + expect({ + capability: session.capabilities.commitDrafts, + result: await response.json(), + status: response.status, + }).toEqual({ capability: true, result: draft, status: 200 }); + expect(updateDraft).toHaveBeenCalledWith(request); + }); + + it('rejects a Commit Draft response for a different opaque Worktree ID', async () => { + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { + commitDrafts: async () => + commitDraftSchema.parse({ + worktreeId: 'worktree_bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb', + revision: 3, + text: 'Wrong Worktree draft', + }), + }, + }); + servers.push(server); + + const response = await fetch(endpointUrl(server, 'draft'), { + method: 'PUT', + headers: { ...headers, 'content-type': 'application/json' }, + body: JSON.stringify({ + worktreeId: 'worktree_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', + expectedRevision: 2, + update: { kind: 'set', text: 'Expected Worktree draft' }, + }), + }); + + expect({ body: await response.json(), status: response.status }).toEqual({ + body: { + error: { + code: 'internal_error', + message: 'The protocol handler returned an invalid response.', + }, + }, + status: 500, + }); + }); + + it('dispatches one validated Product Command and returns its receipt', async () => { + const request = { + clientCommandId: 'command_0123456789abcdef0123456789abcdef', + command: { + kind: 'refresh', + repositoryId: 'repository_0123456789abcdef0123456789abcdef', + }, + } as const; + const receipt = operationReceiptSchema.parse({ + operationId: 'operation_0123456789abcdef0123456789abcdef', + clientCommandId: request.clientCommandId, + disposition: 'accepted', + }); + const dispatchCommand = vi.fn(() => receipt); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { commands: dispatchCommand }, + }); + servers.push(server); + + const response = await postJson(server, 'commands', request); + const session = (await ( + await fetch(server.sessionUrl, { headers }) + ).json()) as { capabilities: { commands: boolean } }; + + expect({ + capability: session.capabilities.commands, + receipt: await response.json(), + status: response.status, + }).toEqual({ capability: true, receipt, status: 200 }); + expect(dispatchCommand).toHaveBeenCalledWith(request); + }); + + it('returns one operation for concurrent exact command retries', async () => { + const request = { + clientCommandId: 'command_11111111111111111111111111111111', + command: { + kind: 'refresh', + repositoryId: 'repository_0123456789abcdef0123456789abcdef', + }, + } as const; + const receipt = operationReceiptSchema.parse({ + operationId: 'operation_11111111111111111111111111111111', + clientCommandId: request.clientCommandId, + disposition: 'accepted', + }); + const dispatchCommand = vi.fn(async () => receipt); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { commands: dispatchCommand }, + }); + servers.push(server); + + const responses = await Promise.all([ + postJson(server, 'commands', request), + postJson(server, 'commands', { + clientCommandId: request.clientCommandId, + command: { + repositoryId: request.command.repositoryId, + kind: request.command.kind, + }, + }), + ]); + const bodies = await Promise.all( + responses.map((response) => response.json()), + ); + + expect(bodies).toEqual([receipt, { ...receipt, disposition: 'duplicate' }]); + expect(dispatchCommand).toHaveBeenCalledOnce(); + }); + + it('rejects a command ID reused for a different validated intent', async () => { + const clientCommandId = 'command_22222222222222222222222222222222'; + const receipt = operationReceiptSchema.parse({ + operationId: 'operation_22222222222222222222222222222222', + clientCommandId, + disposition: 'accepted', + }); + const dispatchCommand = vi.fn(() => receipt); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { commands: dispatchCommand }, + }); + servers.push(server); + + await postJson(server, 'commands', { + clientCommandId, + command: { + kind: 'refresh', + repositoryId: 'repository_11111111111111111111111111111111', + }, + }); + const collision = await postJson(server, 'commands', { + clientCommandId, + command: { + kind: 'refresh', + repositoryId: 'repository_22222222222222222222222222222222', + }, + }); + + expect({ body: await collision.json(), status: collision.status }).toEqual({ + body: { + error: { + code: 'command_id_collision', + message: + 'The client command ID was already used for another command.', + }, + }, + status: 409, + }); + expect(dispatchCommand).toHaveBeenCalledOnce(); + }); + + it('records a command before a synchronously failing handler can be retried', async () => { + const request = { + clientCommandId: 'command_77777777777777777777777777777777', + command: { + kind: 'refresh', + repositoryId: 'repository_77777777777777777777777777777777', + }, + } as const; + const dispatchCommand = vi.fn(() => { + throw new Error('fixture handler failure'); + }); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { commands: dispatchCommand }, + }); + servers.push(server); + + const first = await postJson(server, 'commands', request); + const retry = await postJson(server, 'commands', request); + + expect({ + allowOrigin: first.headers.get('access-control-allow-origin'), + first: await first.json(), + firstStatus: first.status, + retry: await retry.json(), + retryStatus: retry.status, + }).toEqual({ + allowOrigin: 'null', + first: { + error: { + code: 'internal_error', + message: 'The protocol request could not be completed.', + }, + }, + firstStatus: 500, + retry: { + error: { + code: 'duplicate_command', + message: 'The duplicate command cannot be replayed safely.', + }, + }, + retryStatus: 409, + }); + expect(dispatchCommand).toHaveBeenCalledOnce(); + }); + + it('routes operation recovery by opaque Operation ID', async () => { + const operationId = 'operation_33333333333333333333333333333333'; + const result = operationResultSchema.parse({ + kind: 'unknown_outcome', + operationId, + code: 'reconciliation_incomplete', + message: 'Repository state is still being reconciled.', + recoveryAvailable: true, + }); + const recoverOperation = vi.fn(() => result); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { operationRecovery: recoverOperation }, + }); + servers.push(server); + + const response = await postJson(server, 'operations', { operationId }); + const session = (await ( + await fetch(server.sessionUrl, { headers }) + ).json()) as { capabilities: { operationRecovery: boolean } }; + + expect({ + capability: session.capabilities.operationRecovery, + result: await response.json(), + status: response.status, + }).toEqual({ capability: true, result, status: 200 }); + expect(recoverOperation).toHaveBeenCalledWith(operationId); + }); + + it('rejects a recovery result for a different Operation ID', async () => { + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { + operationRecovery: async () => + operationResultSchema.parse({ + kind: 'succeeded', + operationId: 'operation_55555555555555555555555555555555', + result: { kind: 'no_change' }, + }), + }, + }); + servers.push(server); + + const response = await postJson(server, 'operations', { + operationId: 'operation_44444444444444444444444444444444', + }); + + expect({ body: await response.json(), status: response.status }).toEqual({ + body: { + error: { + code: 'internal_error', + message: 'The protocol handler returned an invalid response.', + }, + }, + status: 500, + }); + }); + + it('rejects a native action outside the server-issued target allow-list', async () => { + const targetId = 'native_66666666666666666666666666666666'; + const actionsForTarget = vi.fn((candidate: string) => + candidate === targetId ? (['copy_relative_path'] as const) : undefined, + ); + const perform = vi.fn(() => ({ kind: 'performed' as const })); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { nativeActions: { actionsForTarget, perform } }, + }); + servers.push(server); + + const response = await postJson(server, 'native-actions', { + kind: 'open_default_app', + targetId, + }); + const unissued = await postJson(server, 'native-actions', { + kind: 'copy_relative_path', + targetId: 'native_99999999999999999999999999999999', + }); + const session = (await ( + await fetch(server.sessionUrl, { headers }) + ).json()) as { capabilities: { nativeActions: boolean } }; + + expect({ + capability: session.capabilities.nativeActions, + result: await response.json(), + status: response.status, + unissued: await unissued.json(), + unissuedStatus: unissued.status, + }).toEqual({ + capability: true, + result: { + error: { + code: 'stale_target', + message: 'The native target is stale or does not allow this action.', + }, + }, + status: 409, + unissued: { + error: { + code: 'stale_target', + message: 'The native target is stale or does not allow this action.', + }, + }, + unissuedStatus: 409, + }); + expect(actionsForTarget).toHaveBeenCalledTimes(2); + expect(perform).not.toHaveBeenCalled(); + }); + + it('performs an allow-listed native action against its exact opaque target', async () => { + const targetId = 'native_88888888888888888888888888888888'; + const perform = vi.fn(() => ({ + kind: 'copy_text' as const, + text: 'src/protocol-dispatch.ts', + })); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { + nativeActions: { + actionsForTarget: () => ['copy_relative_path'], + perform, + }, + }, + }); + servers.push(server); + const request = { kind: 'copy_relative_path', targetId } as const; + + const response = await postJson(server, 'native-actions', request); + + expect({ result: await response.json(), status: response.status }).toEqual({ + result: { kind: 'copy_text', text: 'src/protocol-dispatch.ts' }, + status: 200, + }); + expect(perform).toHaveBeenCalledWith(request); + }); +}); + +function endpointUrl(server: LoopbackServer, endpoint: string): URL { + const url = new URL(server.sessionUrl); + url.pathname = url.pathname.replace(/\/session$/u, `/${endpoint}`); + return url; +} + +function postJson( + server: LoopbackServer, + endpoint: string, + value: unknown, +): Promise { + return fetch(endpointUrl(server, endpoint), { + method: 'POST', + headers: { ...headers, 'content-type': 'application/json' }, + body: JSON.stringify(value), + }); +} diff --git a/apps/server/src/protocol-dispatch.ts b/apps/server/src/protocol-dispatch.ts new file mode 100644 index 0000000..4e5b31e --- /dev/null +++ b/apps/server/src/protocol-dispatch.ts @@ -0,0 +1,379 @@ +import { createHash } from 'node:crypto'; + +import { + branchSearchRequestSchema, + branchSearchResultSchema, + commitDraftSchema, + commitDraftUpdateSchema, + commandEnvelopeSchema, + diffRequestSchema, + diffResultSchema, + operationReceiptSchema, + operationRecoveryRequestSchema, + operationResultSchema, + nativeActionRequestSchema, + nativeActionResultSchema, + parseProtocolPayload, + PROTOCOL_LIMITS, + PROTOCOL_VERSION, + type DiffRequest, + type DiffResult, + type FileId, + type BranchSearchRequest, + type BranchSearchResult, + type CommitDraft, + type CommitDraftUpdate, + type CommandEnvelope, + type RepositorySnapshot, + type OperationReceipt, + type OperationId, + type OperationResult, + type NativeActionKind, + type NativeActionRequest, + type NativeActionResult, + type NativeTargetId, + repositorySnapshotSchema, + type SessionMetadata, + type WorktreeId, +} from '@codex-git/protocol'; + +type Awaitable = Promise | T; + +export interface NativeActionHandler { + actionsForTarget( + targetId: NativeTargetId, + ): Awaitable; + perform(request: NativeActionRequest): Awaitable; +} + +export interface ProtocolHandlers { + readonly branchSearch?: ( + request: BranchSearchRequest, + ) => Awaitable; + readonly commitDrafts?: ( + request: CommitDraftUpdate, + ) => Awaitable; + readonly commands?: (request: CommandEnvelope) => Awaitable; + readonly operationRecovery?: ( + operationId: OperationId, + ) => Awaitable; + readonly nativeActions?: NativeActionHandler; + readonly diff?: (request: DiffRequest) => Awaitable; + readonly snapshot?: () => Awaitable; +} + +export interface ProtocolDispatchResponse { + readonly status: number; + readonly value: unknown; +} + +export interface ProtocolDispatcher { + readonly sessionMetadata: SessionMetadata; + dispatch( + endpoint: string, + body?: Uint8Array, + ): Promise; +} + +export function createProtocolDispatcher( + handlers: ProtocolHandlers = {}, +): ProtocolDispatcher { + const commandRecords = new Map< + string, + { + readonly fingerprint: string; + readonly response: Promise; + } + >(); + + return { + sessionMetadata: { + protocolVersion: PROTOCOL_VERSION, + capabilities: { + branchSearch: handlers.branchSearch !== undefined, + commands: handlers.commands !== undefined, + commitDrafts: handlers.commitDrafts !== undefined, + diff: handlers.diff !== undefined, + events: true, + nativeActions: handlers.nativeActions !== undefined, + operationRecovery: handlers.operationRecovery !== undefined, + }, + limits: PROTOCOL_LIMITS, + }, + async dispatch(endpoint, body) { + if (endpoint === 'snapshot' && handlers.snapshot !== undefined) { + return validatedResponse( + repositorySnapshotSchema, + await handlers.snapshot(), + ); + } + if (endpoint === 'diff' && handlers.diff !== undefined) { + const input = parseJsonBody(body); + if (!input.ok) return input.response; + const request = parseProtocolPayload(diffRequestSchema, input.value); + if (!request.ok) { + return { status: 400, value: { error: request.error } }; + } + return diffResponse( + request.value.fileId, + await handlers.diff(request.value), + ); + } + if (endpoint === 'branches' && handlers.branchSearch !== undefined) { + const input = parseJsonBody(body); + if (!input.ok) return input.response; + const request = parseProtocolPayload( + branchSearchRequestSchema, + input.value, + ); + if (!request.ok) { + return { status: 400, value: { error: request.error } }; + } + return validatedResponse( + branchSearchResultSchema, + await handlers.branchSearch(request.value), + ); + } + if (endpoint === 'draft' && handlers.commitDrafts !== undefined) { + const input = parseJsonBody(body); + if (!input.ok) return input.response; + const request = parseProtocolPayload( + commitDraftUpdateSchema, + input.value, + ); + if (!request.ok) { + return { status: 400, value: { error: request.error } }; + } + return commitDraftResponse( + request.value.worktreeId, + await handlers.commitDrafts(request.value), + ); + } + if (endpoint === 'commands' && handlers.commands !== undefined) { + const input = parseJsonBody(body); + if (!input.ok) return input.response; + const request = parseProtocolPayload( + commandEnvelopeSchema, + input.value, + ); + if (!request.ok) { + return { status: 400, value: { error: request.error } }; + } + const fingerprint = fingerprintCommand(request.value.command); + const existing = commandRecords.get(request.value.clientCommandId); + if (existing !== undefined) { + if (existing.fingerprint !== fingerprint) { + return commandCollisionResponse(); + } + let response: ProtocolDispatchResponse; + try { + response = await existing.response; + } catch { + return duplicateCommandResponse(); + } + const receipt = operationReceiptSchema.safeParse(response.value); + if (response.status !== 200 || !receipt.success) { + return duplicateCommandResponse(); + } + return { + status: 200, + value: { ...receipt.data, disposition: 'duplicate' }, + }; + } + const dispatchCommand = handlers.commands; + const response = Promise.resolve() + .then(() => dispatchCommand(request.value)) + .then((value) => acceptedCommandResponse(request.value, value)); + commandRecords.set(request.value.clientCommandId, { + fingerprint, + response, + }); + return response; + } + if ( + endpoint === 'operations' && + handlers.operationRecovery !== undefined + ) { + const input = parseJsonBody(body); + if (!input.ok) return input.response; + const request = parseProtocolPayload( + operationRecoveryRequestSchema, + input.value, + ); + if (!request.ok) { + return { status: 400, value: { error: request.error } }; + } + return operationRecoveryResponse( + request.value.operationId, + await handlers.operationRecovery(request.value.operationId), + ); + } + if ( + endpoint === 'native-actions' && + handlers.nativeActions !== undefined + ) { + const input = parseJsonBody(body); + if (!input.ok) return input.response; + const request = parseProtocolPayload( + nativeActionRequestSchema, + input.value, + ); + if (!request.ok) { + return { status: 400, value: { error: request.error } }; + } + const allowed = await handlers.nativeActions.actionsForTarget( + request.value.targetId, + ); + if (allowed === undefined || !allowed.includes(request.value.kind)) { + return staleNativeTargetResponse(); + } + return validatedResponse( + nativeActionResultSchema, + await handlers.nativeActions.perform(request.value), + ); + } + return undefined; + }, + }; +} + +function fingerprintCommand(command: CommandEnvelope['command']): string { + return createHash('sha256').update(JSON.stringify(command)).digest('hex'); +} + +function diffResponse( + fileId: FileId, + value: unknown, +): ProtocolDispatchResponse { + const result = diffResultSchema.safeParse(value); + if (!result.success || result.data.fileId !== fileId) { + return invalidHandlerResponse(); + } + return { status: 200, value: result.data }; +} + +function commitDraftResponse( + worktreeId: WorktreeId, + value: unknown, +): ProtocolDispatchResponse { + const draft = commitDraftSchema.safeParse(value); + if (!draft.success || draft.data.worktreeId !== worktreeId) { + return invalidHandlerResponse(); + } + return { status: 200, value: draft.data }; +} + +function staleNativeTargetResponse(): ProtocolDispatchResponse { + return { + status: 409, + value: { + error: { + code: 'stale_target', + message: 'The native target is stale or does not allow this action.', + }, + }, + }; +} + +function operationRecoveryResponse( + operationId: OperationId, + value: unknown, +): ProtocolDispatchResponse { + const result = operationResultSchema.safeParse(value); + if (!result.success || result.data.operationId !== operationId) { + return invalidHandlerResponse(); + } + return { status: 200, value: result.data }; +} + +function acceptedCommandResponse( + request: CommandEnvelope, + value: unknown, +): ProtocolDispatchResponse { + const receipt = operationReceiptSchema.safeParse(value); + if ( + !receipt.success || + receipt.data.clientCommandId !== request.clientCommandId || + receipt.data.disposition !== 'accepted' + ) { + return invalidHandlerResponse(); + } + return { status: 200, value: receipt.data }; +} + +function commandCollisionResponse(): ProtocolDispatchResponse { + return { + status: 409, + value: { + error: { + code: 'command_id_collision', + message: 'The client command ID was already used for another command.', + }, + }, + }; +} + +function duplicateCommandResponse(): ProtocolDispatchResponse { + return { + status: 409, + value: { + error: { + code: 'duplicate_command', + message: 'The duplicate command cannot be replayed safely.', + }, + }, + }; +} + +interface RuntimeSchema { + safeParse( + input: unknown, + ): { readonly data: T; readonly success: true } | { readonly success: false }; +} + +function validatedResponse( + schema: RuntimeSchema, + value: unknown, +): ProtocolDispatchResponse { + const parsed = schema.safeParse(value); + if (parsed.success) return { status: 200, value: parsed.data }; + return invalidHandlerResponse(); +} + +function invalidHandlerResponse(): ProtocolDispatchResponse { + return { + status: 500, + value: { + error: { + code: 'internal_error', + message: 'The protocol handler returned an invalid response.', + }, + }, + }; +} + +type JsonBodyResult = + | { readonly ok: true; readonly value: unknown } + | { readonly ok: false; readonly response: ProtocolDispatchResponse }; + +function parseJsonBody(body: Uint8Array | undefined): JsonBodyResult { + try { + return { + ok: true, + value: JSON.parse(new TextDecoder().decode(body)) as unknown, + }; + } catch { + return { + ok: false, + response: { + status: 400, + value: { + error: { + code: 'invalid_payload', + message: 'The protocol request body is not valid JSON.', + }, + }, + }, + }; + } +} diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index d0908eb..da8148b 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -1 +1,5 @@ export * from './loopback-server.js'; +export type { + NativeActionHandler, + ProtocolHandlers, +} from './protocol-dispatch.js'; From cf02166c62634206ea02b2c840e8873ccf569755 Mon Sep 17 00:00:00 2001 From: leyoonafr Date: Sun, 30 Aug 2026 00:36:15 +0800 Subject: [PATCH 2/4] fix: harden protocol dispatch review findings (#5) --- apps/server/src/loopback-server.ts | 43 ++-- apps/server/src/protocol-dispatch.test.ts | 265 +++++++++++++++++++--- apps/server/src/protocol-dispatch.ts | 105 +++++++-- 3 files changed, 344 insertions(+), 69 deletions(-) diff --git a/apps/server/src/loopback-server.ts b/apps/server/src/loopback-server.ts index 436a987..8b2989e 100644 --- a/apps/server/src/loopback-server.ts +++ b/apps/server/src/loopback-server.ts @@ -10,14 +10,18 @@ import { PROTOCOL_VERSION, PROTOCOL_VERSION_HEADER, PROTOCOL_LIMITS, - redactDiagnostic, + createDiagnosticRedactor, sseInvalidationSchema, type HealthResponse, + type DiagnosticRedactor, type ProtocolError, } from '@codex-git/protocol'; import { createProtocolDispatcher, + PROTOCOL_ENDPOINTS, + redactProtocolDiagnostics, + resolveProtocolEndpoint, type ProtocolDispatcher, type ProtocolHandlers, } from './protocol-dispatch.js'; @@ -28,17 +32,6 @@ const healthResponse = { } satisfies HealthResponse; const loopbackHost = '127.0.0.1' as const; -const endpointMethods = new Map([ - ['branches', 'POST'], - ['commands', 'POST'], - ['diff', 'POST'], - ['draft', 'PUT'], - ['events', 'GET'], - ['native-actions', 'POST'], - ['operations', 'POST'], - ['session', 'GET'], - ['snapshot', 'GET'], -]); export interface LoopbackAddress { readonly host: typeof loopbackHost; readonly port: number; @@ -70,6 +63,7 @@ export async function startLoopbackServer( const token = Buffer.from(tokenBytes).toString('hex'); const instancePrefix = `/instance/${token}/v1`; const allowedOrigins = new Set(options.allowedOrigins ?? []); + const redactDiagnostic = createDiagnosticRedactor({ secrets: [token] }); const dispatcher = createProtocolDispatcher(options.handlers); const events = createEventBroker(); const server = createServer((request, response) => { @@ -79,6 +73,7 @@ export async function startLoopbackServer( token, instancePrefix, allowedOrigins, + redactDiagnostic, dispatcher, events, ).catch(() => { @@ -132,6 +127,7 @@ async function handleRequest( token: string, instancePrefix: string, allowedOrigins: ReadonlySet, + redactDiagnostic: DiagnosticRedactor, dispatcher: ProtocolDispatcher, events: EventBroker, ): Promise { @@ -178,16 +174,18 @@ async function handleRequest( return; } - const endpoint = pathname.startsWith(`${instancePrefix}/`) + const endpointName = pathname.startsWith(`${instancePrefix}/`) ? pathname.slice(instancePrefix.length + 1) : ''; - const expectedMethod = endpointMethods.get(endpoint); + const endpoint = resolveProtocolEndpoint(endpointName); + const expectedMethod = + endpoint === undefined ? undefined : PROTOCOL_ENDPOINTS[endpoint].method; if (request.method === 'OPTIONS') { handlePreflight(request, response, origin, expectedMethod, allowedOrigins); return; } - if (endpoint !== 'events' && version !== String(PROTOCOL_VERSION)) { + if (endpointName !== 'events' && version !== String(PROTOCOL_VERSION)) { sendProtocolError(response, 426, { code: 'unsupported_protocol_version', details: { @@ -207,7 +205,7 @@ async function handleRequest( return; } - if (expectedMethod === undefined) { + if (endpoint === undefined) { sendProtocolError( response, 404, @@ -280,7 +278,13 @@ async function handleRequest( try { const dispatched = await dispatcher.dispatch(endpoint, body); if (dispatched !== undefined) { - sendJson(response, dispatched.status, dispatched.value, origin); + sendJson( + response, + dispatched.status, + redactProtocolDiagnostics(dispatched.value, redactDiagnostic), + origin, + PROTOCOL_ENDPOINTS[endpoint].responseBodyBytes, + ); return; } } catch { @@ -521,9 +525,10 @@ function sendJson( status: number, value: unknown, origin?: string, + limitBytes: number = PROTOCOL_LIMITS.diffOutputBytes, ): void { const body = JSON.stringify(value); - if (Buffer.byteLength(body) > PROTOCOL_LIMITS.diffOutputBytes) { + if (Buffer.byteLength(body) > limitBytes) { response.writeHead(507, { ...(origin === undefined ? {} @@ -534,7 +539,7 @@ function sendJson( JSON.stringify({ error: { code: 'output_too_large', - details: { limitBytes: PROTOCOL_LIMITS.diffOutputBytes }, + details: { limitBytes }, message: 'The protocol response exceeds the configured limit.', }, }), diff --git a/apps/server/src/protocol-dispatch.test.ts b/apps/server/src/protocol-dispatch.test.ts index 13564e6..5580089 100644 --- a/apps/server/src/protocol-dispatch.test.ts +++ b/apps/server/src/protocol-dispatch.test.ts @@ -3,6 +3,7 @@ import { afterEach, describe, expect, it, vi } from 'vitest'; import { branchSearchResultSchema, commitDraftSchema, + PROTOCOL_LIMITS, PROTOCOL_VERSION_HEADER, diffResultSchema, operationReceiptSchema, @@ -23,14 +24,36 @@ afterEach(async () => { }); describe('protocol HTTP dispatch', () => { - it('routes snapshots through an installed handler without advertising absent handlers', async () => { + it('routes snapshots with redacted diagnostics without advertising absent handlers', async () => { const snapshot = repositorySnapshotSchema.parse({ repositoryId: 'repository_0123456789abcdef0123456789abcdef', repositoryRevision: 4, topologyRevision: 2, refsRevision: 3, - refresh: { kind: 'current' }, - worktrees: [], + refresh: { + kind: 'stale', + message: 'Authorization: Bearer fixture-snapshot-token', + }, + worktrees: [ + { + worktreeId: 'worktree_0123456789abcdef0123456789abcdef', + worktreeRevision: 1, + generation: 'generation_0123456789abcdef0123456789abcdef', + freshness: { + kind: 'failed', + message: + 'remote=https://alice:fixture-password@example.com/repo.git', + }, + head: { kind: 'initial' }, + indexTree: null, + status: { + kind: 'unavailable', + reason: 'token=fixture-unavailable-secret', + }, + changes: [], + nativeTargets: [], + }, + ], remotes: [], operations: [], }); @@ -46,13 +69,14 @@ describe('protocol HTTP dispatch', () => { fetch(endpointUrl(server, 'snapshot'), { headers }), ]); + const snapshotBody = await snapshotResponse.json(); expect({ capabilities: ( (await sessionResponse.json()) as { capabilities: Record; } ).capabilities, - snapshot: await snapshotResponse.json(), + snapshotIsValid: repositorySnapshotSchema.safeParse(snapshotBody).success, snapshotStatus: snapshotResponse.status, }).toEqual({ capabilities: { @@ -64,9 +88,23 @@ describe('protocol HTTP dispatch', () => { nativeActions: false, operationRecovery: false, }, - snapshot, + snapshotIsValid: true, snapshotStatus: 200, }); + expect(snapshotBody).toMatchObject({ + refresh: { message: 'Authorization: [REDACTED]' }, + worktrees: [ + { + freshness: { + message: 'remote=https://[REDACTED]@example.com/repo.git', + }, + status: { reason: 'token=[REDACTED]' }, + }, + ], + }); + expect(JSON.stringify(snapshotBody)).not.toMatch( + /fixture-(?:snapshot-token|password|unavailable-secret)/u, + ); expect(handleSnapshot).toHaveBeenCalledOnce(); }); @@ -115,6 +153,39 @@ describe('protocol HTTP dispatch', () => { expect(handleDiff).not.toHaveBeenCalled(); }); + it('rejects malformed UTF-8 JSON bytes before calling a handler', async () => { + const searchBranches = vi.fn(() => + branchSearchResultSchema.parse({ refsRevision: 1, candidates: [] }), + ); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { branchSearch: searchBranches }, + }); + servers.push(server); + const prefix = new TextEncoder().encode( + `{"worktreeId":"worktree_0123456789abcdef0123456789abcdef","query":"`, + ); + const suffix = new TextEncoder().encode('"}'); + const body = Uint8Array.from([...prefix, 0xc3, 0x28, ...suffix]); + + const response = await fetch(endpointUrl(server, 'branches'), { + method: 'POST', + headers: { ...headers, 'content-type': 'application/json' }, + body, + }); + + expect({ body: await response.json(), status: response.status }).toEqual({ + body: { + error: { + code: 'invalid_payload', + message: 'The protocol request body is not valid JSON.', + }, + }, + status: 400, + }); + expect(searchBranches).not.toHaveBeenCalled(); + }); + it('does not serialize a handler response that fails runtime validation', async () => { const server = await startLoopbackServer({ allowedOrigins: ['null'], @@ -182,6 +253,36 @@ describe('protocol HTTP dispatch', () => { }); }); + it('returns a diff at the exact negotiated content limit', async () => { + const fileId = 'file_cccccccccccccccccccccccccccccccc'; + const content = 'x'.repeat(PROTOCOL_LIMITS.diffOutputBytes); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { + diff: () => + diffResultSchema.parse({ + kind: 'text', + fileId, + baseline: 'head_to_index', + content, + lineCount: 1, + }), + }, + }); + servers.push(server); + + const response = await postJson(server, 'diff', { fileId }); + const body = (await response.json()) as { content?: string }; + + expect({ + contentBytes: body.content?.length, + status: response.status, + }).toEqual({ + contentBytes: PROTOCOL_LIMITS.diffOutputBytes, + status: 200, + }); + }); + it('dispatches branch search with validated opaque targets', async () => { const result = branchSearchResultSchema.parse({ refsRevision: 7, @@ -455,16 +556,25 @@ describe('protocol HTTP dispatch', () => { it('routes operation recovery by opaque Operation ID', async () => { const operationId = 'operation_33333333333333333333333333333333'; + const launchToken = 'ab'.repeat(32); const result = operationResultSchema.parse({ - kind: 'unknown_outcome', + kind: 'partial_success', operationId, - code: 'reconciliation_incomplete', - message: 'Repository state is still being reconciled.', - recoveryAvailable: true, + message: `Recovery retained ${launchToken}`, + effects: [ + { kind: 'succeeded', label: 'origin' }, + { + kind: 'failed_known', + label: 'backup', + code: 'authentication', + message: 'Authorization: Bearer fixture-operation-token', + }, + ], }); const recoverOperation = vi.fn(() => result); const server = await startLoopbackServer({ allowedOrigins: ['null'], + randomBytes: (length) => new Uint8Array(length).fill(0xab), handlers: { operationRecovery: recoverOperation }, }); servers.push(server); @@ -474,11 +584,23 @@ describe('protocol HTTP dispatch', () => { await fetch(server.sessionUrl, { headers }) ).json()) as { capabilities: { operationRecovery: boolean } }; + const responseBody = await response.json(); expect({ capability: session.capabilities.operationRecovery, - result: await response.json(), + resultIsValid: operationResultSchema.safeParse(responseBody).success, status: response.status, - }).toEqual({ capability: true, result, status: 200 }); + }).toEqual({ capability: true, resultIsValid: true, status: 200 }); + expect(responseBody).toMatchObject({ + message: 'Recovery retained [REDACTED]', + effects: [ + { kind: 'succeeded' }, + { message: 'Authorization: [REDACTED]' }, + ], + }); + expect(JSON.stringify(responseBody)).not.toContain(launchToken); + expect(JSON.stringify(responseBody)).not.toContain( + 'fixture-operation-token', + ); expect(recoverOperation).toHaveBeenCalledWith(operationId); }); @@ -511,38 +633,28 @@ describe('protocol HTTP dispatch', () => { }); }); - it('rejects a native action outside the server-issued target allow-list', async () => { - const targetId = 'native_66666666666666666666666666666666'; - const actionsForTarget = vi.fn((candidate: string) => - candidate === targetId ? (['copy_relative_path'] as const) : undefined, - ); + it('rejects a fabricated native target that was not issued by the snapshot', async () => { + const issuedTargetId = 'native_66666666666666666666666666666666'; const perform = vi.fn(() => ({ kind: 'performed' as const })); const server = await startLoopbackServer({ allowedOrigins: ['null'], - handlers: { nativeActions: { actionsForTarget, perform } }, + handlers: { + snapshot: () => nativeSnapshot(issuedTargetId, ['copy_relative_path']), + nativeActions: perform, + }, }); servers.push(server); + await fetch(endpointUrl(server, 'snapshot'), { headers }); const response = await postJson(server, 'native-actions', { - kind: 'open_default_app', - targetId, - }); - const unissued = await postJson(server, 'native-actions', { kind: 'copy_relative_path', targetId: 'native_99999999999999999999999999999999', }); - const session = (await ( - await fetch(server.sessionUrl, { headers }) - ).json()) as { capabilities: { nativeActions: boolean } }; expect({ - capability: session.capabilities.nativeActions, result: await response.json(), status: response.status, - unissued: await unissued.json(), - unissuedStatus: unissued.status, }).toEqual({ - capability: true, result: { error: { code: 'stale_target', @@ -550,15 +662,37 @@ describe('protocol HTTP dispatch', () => { }, }, status: 409, - unissued: { + }); + expect(perform).not.toHaveBeenCalled(); + }); + + it('rejects an action that was not issued for the native target', async () => { + const targetId = 'native_77777777777777777777777777777777'; + const perform = vi.fn(() => ({ kind: 'performed' as const })); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { + snapshot: () => nativeSnapshot(targetId, ['copy_relative_path']), + nativeActions: perform, + }, + }); + servers.push(server); + + await fetch(endpointUrl(server, 'snapshot'), { headers }); + const response = await postJson(server, 'native-actions', { + kind: 'open_default_app', + targetId, + }); + + expect({ result: await response.json(), status: response.status }).toEqual({ + result: { error: { code: 'stale_target', message: 'The native target is stale or does not allow this action.', }, }, - unissuedStatus: 409, + status: 409, }); - expect(actionsForTarget).toHaveBeenCalledTimes(2); expect(perform).not.toHaveBeenCalled(); }); @@ -571,15 +705,14 @@ describe('protocol HTTP dispatch', () => { const server = await startLoopbackServer({ allowedOrigins: ['null'], handlers: { - nativeActions: { - actionsForTarget: () => ['copy_relative_path'], - perform, - }, + snapshot: () => nativeSnapshot(targetId, ['copy_relative_path']), + nativeActions: perform, }, }); servers.push(server); const request = { kind: 'copy_relative_path', targetId } as const; + await fetch(endpointUrl(server, 'snapshot'), { headers }); const response = await postJson(server, 'native-actions', request); expect({ result: await response.json(), status: response.status }).toEqual({ @@ -588,6 +721,40 @@ describe('protocol HTTP dispatch', () => { }); expect(perform).toHaveBeenCalledWith(request); }); + + it('leaves current-state validation and execution in one native handler call', async () => { + const targetId = 'native_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'; + let targetExists = true; + const perform = vi.fn(() => + targetExists + ? ({ kind: 'performed' } as const) + : ({ + kind: 'unavailable', + message: 'The target no longer exists.', + } as const), + ); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { + snapshot: () => nativeSnapshot(targetId, ['reveal_in_finder']), + nativeActions: perform, + }, + }); + servers.push(server); + + await fetch(endpointUrl(server, 'snapshot'), { headers }); + targetExists = false; + const response = await postJson(server, 'native-actions', { + kind: 'reveal_in_finder', + targetId, + }); + + expect({ result: await response.json(), status: response.status }).toEqual({ + result: { kind: 'unavailable', message: 'The target no longer exists.' }, + status: 200, + }); + expect(perform).toHaveBeenCalledOnce(); + }); }); function endpointUrl(server: LoopbackServer, endpoint: string): URL { @@ -607,3 +774,31 @@ function postJson( body: JSON.stringify(value), }); } + +function nativeSnapshot( + targetId: string, + actions: readonly string[], +): ReturnType { + return repositorySnapshotSchema.parse({ + repositoryId: 'repository_99999999999999999999999999999999', + repositoryRevision: 1, + topologyRevision: 1, + refsRevision: 1, + refresh: { kind: 'current' }, + worktrees: [ + { + worktreeId: 'worktree_99999999999999999999999999999999', + worktreeRevision: 1, + generation: 'generation_99999999999999999999999999999999', + freshness: { kind: 'current' }, + head: { kind: 'initial' }, + indexTree: null, + status: { kind: 'clean' }, + changes: [], + nativeTargets: [{ targetId, actions }], + }, + ], + remotes: [], + operations: [], + }); +} diff --git a/apps/server/src/protocol-dispatch.ts b/apps/server/src/protocol-dispatch.ts index 4e5b31e..9e0651f 100644 --- a/apps/server/src/protocol-dispatch.ts +++ b/apps/server/src/protocol-dispatch.ts @@ -35,16 +35,14 @@ import { repositorySnapshotSchema, type SessionMetadata, type WorktreeId, + type DiagnosticRedactor, } from '@codex-git/protocol'; type Awaitable = Promise | T; -export interface NativeActionHandler { - actionsForTarget( - targetId: NativeTargetId, - ): Awaitable; - perform(request: NativeActionRequest): Awaitable; -} +export type NativeActionHandler = ( + request: NativeActionRequest, +) => Awaitable; export interface ProtocolHandlers { readonly branchSearch?: ( @@ -67,14 +65,69 @@ export interface ProtocolDispatchResponse { readonly value: unknown; } +const defaultResponseBodyBytes = PROTOCOL_LIMITS.diffOutputBytes; +const diffResponseBodyBytes = PROTOCOL_LIMITS.diffOutputBytes * 6 + 16_384; + +export const PROTOCOL_ENDPOINTS = { + branches: { method: 'POST', responseBodyBytes: defaultResponseBodyBytes }, + commands: { method: 'POST', responseBodyBytes: defaultResponseBodyBytes }, + diff: { method: 'POST', responseBodyBytes: diffResponseBodyBytes }, + draft: { method: 'PUT', responseBodyBytes: defaultResponseBodyBytes }, + events: { method: 'GET', responseBodyBytes: defaultResponseBodyBytes }, + 'native-actions': { + method: 'POST', + responseBodyBytes: defaultResponseBodyBytes, + }, + operations: { method: 'POST', responseBodyBytes: defaultResponseBodyBytes }, + session: { method: 'GET', responseBodyBytes: defaultResponseBodyBytes }, + snapshot: { method: 'GET', responseBodyBytes: defaultResponseBodyBytes }, +} as const; + +export type ProtocolEndpoint = keyof typeof PROTOCOL_ENDPOINTS; + +export function resolveProtocolEndpoint( + value: string, +): ProtocolEndpoint | undefined { + return Object.hasOwn(PROTOCOL_ENDPOINTS, value) + ? (value as ProtocolEndpoint) + : undefined; +} + export interface ProtocolDispatcher { readonly sessionMetadata: SessionMetadata; dispatch( - endpoint: string, + endpoint: ProtocolEndpoint, body?: Uint8Array, ): Promise; } +const diagnosticKeys = new Set(['message', 'reason']); +const fatalUtf8Decoder = new TextDecoder('utf-8', { fatal: true }); + +export function redactProtocolDiagnostics( + value: unknown, + redact: DiagnosticRedactor, + key?: string, +): unknown { + if (typeof value === 'string') { + if (key === undefined || !diagnosticKeys.has(key)) return value; + const redacted = redact(value); + return redacted.length > value.length + ? redacted.slice(0, value.length) + : redacted; + } + if (Array.isArray(value)) { + return value.map((entry) => redactProtocolDiagnostics(entry, redact)); + } + if (typeof value !== 'object' || value === null) return value; + return Object.fromEntries( + Object.entries(value).map(([entryKey, entry]) => [ + entryKey, + redactProtocolDiagnostics(entry, redact, entryKey), + ]), + ); +} + export function createProtocolDispatcher( handlers: ProtocolHandlers = {}, ): ProtocolDispatcher { @@ -85,6 +138,7 @@ export function createProtocolDispatcher( readonly response: Promise; } >(); + let issuedNativeActions = new Map>(); return { sessionMetadata: { @@ -102,10 +156,12 @@ export function createProtocolDispatcher( }, async dispatch(endpoint, body) { if (endpoint === 'snapshot' && handlers.snapshot !== undefined) { - return validatedResponse( - repositorySnapshotSchema, + const snapshot = repositorySnapshotSchema.safeParse( await handlers.snapshot(), ); + if (!snapshot.success) return invalidHandlerResponse(); + issuedNativeActions = collectNativeActions(snapshot.data); + return { status: 200, value: snapshot.data }; } if (endpoint === 'diff' && handlers.diff !== undefined) { const input = parseJsonBody(body); @@ -221,15 +277,16 @@ export function createProtocolDispatcher( if (!request.ok) { return { status: 400, value: { error: request.error } }; } - const allowed = await handlers.nativeActions.actionsForTarget( - request.value.targetId, - ); - if (allowed === undefined || !allowed.includes(request.value.kind)) { + if ( + !issuedNativeActions + .get(request.value.targetId) + ?.has(request.value.kind) + ) { return staleNativeTargetResponse(); } return validatedResponse( nativeActionResultSchema, - await handlers.nativeActions.perform(request.value), + await handlers.nativeActions(request.value), ); } return undefined; @@ -237,6 +294,24 @@ export function createProtocolDispatcher( }; } +function collectNativeActions( + snapshot: RepositorySnapshot, +): Map> { + const issued = new Map>(); + for (const worktree of snapshot.worktrees) { + const descriptors = [ + ...worktree.nativeTargets, + ...worktree.changes.flatMap((change) => change.nativeTargets), + ]; + for (const descriptor of descriptors) { + const actions = issued.get(descriptor.targetId) ?? new Set(); + descriptor.actions.forEach((action) => actions.add(action)); + issued.set(descriptor.targetId, actions); + } + } + return issued; +} + function fingerprintCommand(command: CommandEnvelope['command']): string { return createHash('sha256').update(JSON.stringify(command)).digest('hex'); } @@ -360,7 +435,7 @@ function parseJsonBody(body: Uint8Array | undefined): JsonBodyResult { try { return { ok: true, - value: JSON.parse(new TextDecoder().decode(body)) as unknown, + value: JSON.parse(fatalUtf8Decoder.decode(body)) as unknown, }; } catch { return { From 0c7e304da13113b6cdf8b7b94d538c3a78e8877c Mon Sep 17 00:00:00 2001 From: leyoonafr Date: Sun, 30 Aug 2026 01:03:50 +0800 Subject: [PATCH 3/4] fix: close protocol response review gaps (#5) --- apps/server/src/protocol-dispatch.test.ts | 162 +++++++++++----------- apps/server/src/protocol-dispatch.ts | 21 +-- packages/protocol/src/schemas.ts | 5 +- 3 files changed, 96 insertions(+), 92 deletions(-) diff --git a/apps/server/src/protocol-dispatch.test.ts b/apps/server/src/protocol-dispatch.test.ts index 5580089..91fab2a 100644 --- a/apps/server/src/protocol-dispatch.test.ts +++ b/apps/server/src/protocol-dispatch.test.ts @@ -18,6 +18,12 @@ const headers = { origin: 'null', [PROTOCOL_VERSION_HEADER]: '1', } as const; +const staleTargetResponse = { + error: { + code: 'stale_target', + message: 'The native target is stale or does not allow this action.', + }, +}; afterEach(async () => { await Promise.all(servers.splice(0).map((server) => server.close())); @@ -50,11 +56,21 @@ describe('protocol HTTP dispatch', () => { kind: 'unavailable', reason: 'token=fixture-unavailable-secret', }, - changes: [], + changes: Array.from({ length: 2_000 }, (_, index) => ({ + baseline: 'index_to_working_tree', + displayPath: 'x'.repeat(4_096), + fileId: `file_${index.toString(16).padStart(32, '0')}`, + kind: 'change', + nativeTargets: [], + })), nativeTargets: [], }, ], - remotes: [], + remotes: ['a.co', 'a.co:1', '1.1.1.1', '[::1]'].map((host, index) => ({ + remoteId: `remote_${index.toString(16).padStart(32, '0')}`, + displayName: host, + host, + })), operations: [], }); const handleSnapshot = vi.fn(() => snapshot); @@ -70,15 +86,7 @@ describe('protocol HTTP dispatch', () => { ]); const snapshotBody = await snapshotResponse.json(); - expect({ - capabilities: ( - (await sessionResponse.json()) as { - capabilities: Record; - } - ).capabilities, - snapshotIsValid: repositorySnapshotSchema.safeParse(snapshotBody).success, - snapshotStatus: snapshotResponse.status, - }).toEqual({ + expect(await sessionResponse.json()).toMatchObject({ capabilities: { branchSearch: false, commands: false, @@ -88,9 +96,9 @@ describe('protocol HTTP dispatch', () => { nativeActions: false, operationRecovery: false, }, - snapshotIsValid: true, - snapshotStatus: 200, }); + expect(repositorySnapshotSchema.safeParse(snapshotBody).success).toBe(true); + expect(snapshotResponse.status).toBe(200); expect(snapshotBody).toMatchObject({ refresh: { message: 'Authorization: [REDACTED]' }, worktrees: [ @@ -105,7 +113,20 @@ describe('protocol HTTP dispatch', () => { expect(JSON.stringify(snapshotBody)).not.toMatch( /fixture-(?:snapshot-token|password|unavailable-secret)/u, ); - expect(handleSnapshot).toHaveBeenCalledOnce(); + + handleSnapshot.mockReturnValueOnce({ + ...snapshot, + remotes: [ + { ...snapshot.remotes[0]!, host: 'https://a:fixture@example.com' }, + ], + }); + const invalidHostResponse = await fetch(endpointUrl(server, 'snapshot'), { + headers, + }); + const invalidHostBody = + await expectInvalidHandlerResponse(invalidHostResponse); + expect(JSON.stringify(invalidHostBody)).not.toContain('a:fixture'); + expect(handleSnapshot).toHaveBeenCalledTimes(2); }); it('rejects malformed diff requests before calling the installed handler', async () => { @@ -209,17 +230,7 @@ describe('protocol HTTP dispatch', () => { fileId: 'file_0123456789abcdef0123456789abcdef', }), }); - const body = JSON.stringify(await response.json()); - - expect({ body: JSON.parse(body), status: response.status }).toEqual({ - body: { - error: { - code: 'internal_error', - message: 'The protocol handler returned an invalid response.', - }, - }, - status: 500, - }); + const body = JSON.stringify(await expectInvalidHandlerResponse(response)); expect(body).not.toContain('fixture-handler-secret'); }); @@ -242,15 +253,7 @@ describe('protocol HTTP dispatch', () => { fileId: 'file_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', }); - expect({ body: await response.json(), status: response.status }).toEqual({ - body: { - error: { - code: 'internal_error', - message: 'The protocol handler returned an invalid response.', - }, - }, - status: 500, - }); + await expectInvalidHandlerResponse(response); }); it('returns a diff at the exact negotiated content limit', async () => { @@ -286,14 +289,12 @@ describe('protocol HTTP dispatch', () => { it('dispatches branch search with validated opaque targets', async () => { const result = branchSearchResultSchema.parse({ refsRevision: 7, - candidates: [ - { - refId: 'ref_0123456789abcdef0123456789abcdef', - kind: 'local', - displayName: 'feature/protocol', - occupiedBy: null, - }, - ], + candidates: Array.from({ length: 5_000 }, (_, index) => ({ + refId: `ref_${index.toString(16).padStart(32, '0')}`, + kind: 'local' as const, + displayName: 'x'.repeat(1_024), + occupiedBy: null, + })), }); const searchBranches = vi.fn(() => result); const server = await startLoopbackServer({ @@ -314,11 +315,9 @@ describe('protocol HTTP dispatch', () => { await fetch(server.sessionUrl, { headers }) ).json()) as { capabilities: { branchSearch: boolean } }; - expect({ - capability: session.capabilities.branchSearch, - result: await response.json(), - status: response.status, - }).toEqual({ capability: true, result, status: 200 }); + expect(session.capabilities.branchSearch).toBe(true); + expect(await response.json()).toEqual(result); + expect(response.status).toBe(200); expect(searchBranches).toHaveBeenCalledWith({ worktreeId: 'worktree_0123456789abcdef0123456789abcdef', query: 'feature', @@ -384,15 +383,7 @@ describe('protocol HTTP dispatch', () => { }), }); - expect({ body: await response.json(), status: response.status }).toEqual({ - body: { - error: { - code: 'internal_error', - message: 'The protocol handler returned an invalid response.', - }, - }, - status: 500, - }); + await expectInvalidHandlerResponse(response); }); it('dispatches one validated Product Command and returns its receipt', async () => { @@ -622,15 +613,7 @@ describe('protocol HTTP dispatch', () => { operationId: 'operation_44444444444444444444444444444444', }); - expect({ body: await response.json(), status: response.status }).toEqual({ - body: { - error: { - code: 'internal_error', - message: 'The protocol handler returned an invalid response.', - }, - }, - status: 500, - }); + await expectInvalidHandlerResponse(response); }); it('rejects a fabricated native target that was not issued by the snapshot', async () => { @@ -651,16 +634,8 @@ describe('protocol HTTP dispatch', () => { targetId: 'native_99999999999999999999999999999999', }); - expect({ - result: await response.json(), - status: response.status, - }).toEqual({ - result: { - error: { - code: 'stale_target', - message: 'The native target is stale or does not allow this action.', - }, - }, + expect({ result: await response.json(), status: response.status }).toEqual({ + result: staleTargetResponse, status: 409, }); expect(perform).not.toHaveBeenCalled(); @@ -668,11 +643,13 @@ describe('protocol HTTP dispatch', () => { it('rejects an action that was not issued for the native target', async () => { const targetId = 'native_77777777777777777777777777777777'; + let duplicate = false; const perform = vi.fn(() => ({ kind: 'performed' as const })); const server = await startLoopbackServer({ allowedOrigins: ['null'], handlers: { - snapshot: () => nativeSnapshot(targetId, ['copy_relative_path']), + snapshot: () => + nativeSnapshot(targetId, ['copy_relative_path'], duplicate), nativeActions: perform, }, }); @@ -685,15 +662,16 @@ describe('protocol HTTP dispatch', () => { }); expect({ result: await response.json(), status: response.status }).toEqual({ - result: { - error: { - code: 'stale_target', - message: 'The native target is stale or does not allow this action.', - }, - }, + result: staleTargetResponse, status: 409, }); expect(perform).not.toHaveBeenCalled(); + + duplicate = true; + const duplicateResponse = await fetch(endpointUrl(server, 'snapshot'), { + headers, + }); + await expectInvalidHandlerResponse(duplicateResponse); }); it('performs an allow-listed native action against its exact opaque target', async () => { @@ -775,9 +753,24 @@ function postJson( }); } +async function expectInvalidHandlerResponse(response: Response) { + const body = await response.json(); + expect({ body, status: response.status }).toEqual({ + body: { + error: { + code: 'internal_error', + message: 'The protocol handler returned an invalid response.', + }, + }, + status: 500, + }); + return body; +} + function nativeSnapshot( targetId: string, actions: readonly string[], + duplicate = false, ): ReturnType { return repositorySnapshotSchema.parse({ repositoryId: 'repository_99999999999999999999999999999999', @@ -795,7 +788,10 @@ function nativeSnapshot( indexTree: null, status: { kind: 'clean' }, changes: [], - nativeTargets: [{ targetId, actions }], + nativeTargets: [ + { targetId, actions }, + ...(duplicate ? [{ targetId, actions }] : []), + ], }, ], remotes: [], diff --git a/apps/server/src/protocol-dispatch.ts b/apps/server/src/protocol-dispatch.ts index 9e0651f..dd77c0c 100644 --- a/apps/server/src/protocol-dispatch.ts +++ b/apps/server/src/protocol-dispatch.ts @@ -66,10 +66,14 @@ export interface ProtocolDispatchResponse { } const defaultResponseBodyBytes = PROTOCOL_LIMITS.diffOutputBytes; -const diffResponseBodyBytes = PROTOCOL_LIMITS.diffOutputBytes * 6 + 16_384; +const boundedEnvelope = (textBytes: number) => + defaultResponseBodyBytes + textBytes * 6; +const branchResponseBodyBytes = boundedEnvelope(5_000 * 1_024); +const diffResponseBodyBytes = boundedEnvelope(PROTOCOL_LIMITS.diffOutputBytes); +const snapshotResponseBodyBytes = boundedEnvelope(2_000 * 4_096); export const PROTOCOL_ENDPOINTS = { - branches: { method: 'POST', responseBodyBytes: defaultResponseBodyBytes }, + branches: { method: 'POST', responseBodyBytes: branchResponseBodyBytes }, commands: { method: 'POST', responseBodyBytes: defaultResponseBodyBytes }, diff: { method: 'POST', responseBodyBytes: diffResponseBodyBytes }, draft: { method: 'PUT', responseBodyBytes: defaultResponseBodyBytes }, @@ -80,7 +84,7 @@ export const PROTOCOL_ENDPOINTS = { }, operations: { method: 'POST', responseBodyBytes: defaultResponseBodyBytes }, session: { method: 'GET', responseBodyBytes: defaultResponseBodyBytes }, - snapshot: { method: 'GET', responseBodyBytes: defaultResponseBodyBytes }, + snapshot: { method: 'GET', responseBodyBytes: snapshotResponseBodyBytes }, } as const; export type ProtocolEndpoint = keyof typeof PROTOCOL_ENDPOINTS; @@ -160,7 +164,9 @@ export function createProtocolDispatcher( await handlers.snapshot(), ); if (!snapshot.success) return invalidHandlerResponse(); - issuedNativeActions = collectNativeActions(snapshot.data); + const nativeActions = collectNativeActions(snapshot.data); + if (nativeActions === undefined) return invalidHandlerResponse(); + issuedNativeActions = nativeActions; return { status: 200, value: snapshot.data }; } if (endpoint === 'diff' && handlers.diff !== undefined) { @@ -296,7 +302,7 @@ export function createProtocolDispatcher( function collectNativeActions( snapshot: RepositorySnapshot, -): Map> { +): Map> | undefined { const issued = new Map>(); for (const worktree of snapshot.worktrees) { const descriptors = [ @@ -304,9 +310,8 @@ function collectNativeActions( ...worktree.changes.flatMap((change) => change.nativeTargets), ]; for (const descriptor of descriptors) { - const actions = issued.get(descriptor.targetId) ?? new Set(); - descriptor.actions.forEach((action) => actions.add(action)); - issued.set(descriptor.targetId, actions); + if (issued.has(descriptor.targetId)) return undefined; + issued.set(descriptor.targetId, new Set(descriptor.actions)); } } return issued; diff --git a/packages/protocol/src/schemas.ts b/packages/protocol/src/schemas.ts index 56aeb06..764b809 100644 --- a/packages/protocol/src/schemas.ts +++ b/packages/protocol/src/schemas.ts @@ -196,10 +196,13 @@ export const changedFileSchema = z.discriminatedUnion('kind', [ }), ]); +const sanitizedRemoteHostPattern = + /^(?:\[[\da-f:.]+\]|[\da-z](?:[\da-z.-]*[\da-z])?)(?::\d{1,5})?$/iu; + export const remoteSummarySchema = z.strictObject({ remoteId: remoteIdSchema, displayName: z.string().min(1).max(256), - host: z.string().min(1).max(1_024), + host: z.string().min(1).max(1_024).regex(sanitizedRemoteHostPattern), }); export const worktreeSnapshotSchema = z.strictObject({ From e5debb5fca6b6f6f5b7c8f73c676fd907ec76f45 Mon Sep 17 00:00:00 2001 From: leyoonafr Date: Sun, 30 Aug 2026 01:11:17 +0800 Subject: [PATCH 4/4] fix: reconcile native targets and operation bounds (#5) --- apps/server/src/protocol-dispatch.test.ts | 73 +++++++++++------------ apps/server/src/protocol-dispatch.ts | 12 +++- 2 files changed, 44 insertions(+), 41 deletions(-) diff --git a/apps/server/src/protocol-dispatch.test.ts b/apps/server/src/protocol-dispatch.test.ts index 91fab2a..bd346d5 100644 --- a/apps/server/src/protocol-dispatch.test.ts +++ b/apps/server/src/protocol-dispatch.test.ts @@ -548,18 +548,20 @@ describe('protocol HTTP dispatch', () => { it('routes operation recovery by opaque Operation ID', async () => { const operationId = 'operation_33333333333333333333333333333333'; const launchToken = 'ab'.repeat(32); + const diagnostic = 'Authorization: Bearer fixture-operation-token'; + const padding = 'x'.repeat(8_192 - launchToken.length); const result = operationResultSchema.parse({ kind: 'partial_success', operationId, - message: `Recovery retained ${launchToken}`, + message: `${launchToken}${padding}`, effects: [ - { kind: 'succeeded', label: 'origin' }, - { + { kind: 'succeeded', label: 'x'.repeat(256) }, + ...Array.from({ length: 999 }, (_, index) => ({ kind: 'failed_known', - label: 'backup', + label: 'x'.repeat(256), code: 'authentication', - message: 'Authorization: Bearer fixture-operation-token', - }, + message: index === 0 ? diagnostic : 'x'.repeat(8_192), + })), ], }); const recoverOperation = vi.fn(() => result); @@ -575,23 +577,18 @@ describe('protocol HTTP dispatch', () => { await fetch(server.sessionUrl, { headers }) ).json()) as { capabilities: { operationRecovery: boolean } }; - const responseBody = await response.json(); - expect({ - capability: session.capabilities.operationRecovery, - resultIsValid: operationResultSchema.safeParse(responseBody).success, - status: response.status, - }).toEqual({ capability: true, resultIsValid: true, status: 200 }); - expect(responseBody).toMatchObject({ - message: 'Recovery retained [REDACTED]', - effects: [ - { kind: 'succeeded' }, - { message: 'Authorization: [REDACTED]' }, - ], - }); - expect(JSON.stringify(responseBody)).not.toContain(launchToken); - expect(JSON.stringify(responseBody)).not.toContain( - 'fixture-operation-token', - ); + const responseBody = operationResultSchema.parse(await response.json()); + expect(session.capabilities.operationRecovery).toBe(true); + expect(response.status).toBe(200); + if (responseBody.kind !== 'partial_success') throw new Error('unreachable'); + expect(responseBody.message).toBe(`[REDACTED]${padding}`); + expect(responseBody.effects).toHaveLength(1_000); + expect(responseBody.effects[1]).toMatchObject({ + message: 'Authorization: [REDACTED]', + }); + const serialized = JSON.stringify(responseBody); + expect(serialized).not.toContain(launchToken); + expect(serialized).not.toContain('fixture-operation-token'); expect(recoverOperation).toHaveBeenCalledWith(operationId); }); @@ -643,35 +640,35 @@ describe('protocol HTTP dispatch', () => { it('rejects an action that was not issued for the native target', async () => { const targetId = 'native_77777777777777777777777777777777'; - let duplicate = false; + const actions = ['copy_relative_path', 'reveal_in_finder'] as const; + let duplicate: readonly string[] | undefined; const perform = vi.fn(() => ({ kind: 'performed' as const })); const server = await startLoopbackServer({ allowedOrigins: ['null'], handlers: { - snapshot: () => - nativeSnapshot(targetId, ['copy_relative_path'], duplicate), + snapshot: () => nativeSnapshot(targetId, actions, duplicate), nativeActions: perform, }, }); servers.push(server); - await fetch(endpointUrl(server, 'snapshot'), { headers }); + const snapshotUrl = endpointUrl(server, 'snapshot'); + await fetch(snapshotUrl, { headers }); const response = await postJson(server, 'native-actions', { kind: 'open_default_app', targetId, }); - expect({ result: await response.json(), status: response.status }).toEqual({ - result: staleTargetResponse, - status: 409, - }); + expect(await response.json()).toEqual(staleTargetResponse); + expect(response.status).toBe(409); expect(perform).not.toHaveBeenCalled(); - duplicate = true; - const duplicateResponse = await fetch(endpointUrl(server, 'snapshot'), { - headers, - }); - await expectInvalidHandlerResponse(duplicateResponse); + duplicate = ['reveal_in_finder', 'copy_relative_path']; + const identicalResponse = await fetch(snapshotUrl, { headers }); + expect(identicalResponse.status).toBe(200); + duplicate = ['copy_relative_path']; + const conflictingResponse = await fetch(snapshotUrl, { headers }); + await expectInvalidHandlerResponse(conflictingResponse); }); it('performs an allow-listed native action against its exact opaque target', async () => { @@ -770,7 +767,7 @@ async function expectInvalidHandlerResponse(response: Response) { function nativeSnapshot( targetId: string, actions: readonly string[], - duplicate = false, + duplicate?: readonly string[], ): ReturnType { return repositorySnapshotSchema.parse({ repositoryId: 'repository_99999999999999999999999999999999', @@ -790,7 +787,7 @@ function nativeSnapshot( changes: [], nativeTargets: [ { targetId, actions }, - ...(duplicate ? [{ targetId, actions }] : []), + ...(duplicate ? [{ targetId, actions: duplicate }] : []), ], }, ], diff --git a/apps/server/src/protocol-dispatch.ts b/apps/server/src/protocol-dispatch.ts index dd77c0c..9326b07 100644 --- a/apps/server/src/protocol-dispatch.ts +++ b/apps/server/src/protocol-dispatch.ts @@ -70,6 +70,7 @@ const boundedEnvelope = (textBytes: number) => defaultResponseBodyBytes + textBytes * 6; const branchResponseBodyBytes = boundedEnvelope(5_000 * 1_024); const diffResponseBodyBytes = boundedEnvelope(PROTOCOL_LIMITS.diffOutputBytes); +const operationResponseBodyBytes = boundedEnvelope(1_000 * (8_192 + 256)); const snapshotResponseBodyBytes = boundedEnvelope(2_000 * 4_096); export const PROTOCOL_ENDPOINTS = { @@ -82,7 +83,7 @@ export const PROTOCOL_ENDPOINTS = { method: 'POST', responseBodyBytes: defaultResponseBodyBytes, }, - operations: { method: 'POST', responseBodyBytes: defaultResponseBodyBytes }, + operations: { method: 'POST', responseBodyBytes: operationResponseBodyBytes }, session: { method: 'GET', responseBodyBytes: defaultResponseBodyBytes }, snapshot: { method: 'GET', responseBodyBytes: snapshotResponseBodyBytes }, } as const; @@ -107,6 +108,8 @@ export interface ProtocolDispatcher { const diagnosticKeys = new Set(['message', 'reason']); const fatalUtf8Decoder = new TextDecoder('utf-8', { fatal: true }); +const actionKey = (actions: Set) => + JSON.stringify([...actions].sort()); export function redactProtocolDiagnostics( value: unknown, @@ -310,8 +313,11 @@ function collectNativeActions( ...worktree.changes.flatMap((change) => change.nativeTargets), ]; for (const descriptor of descriptors) { - if (issued.has(descriptor.targetId)) return undefined; - issued.set(descriptor.targetId, new Set(descriptor.actions)); + const actions = new Set(descriptor.actions); + const existing = issued.get(descriptor.targetId); + if (existing && actionKey(existing) !== actionKey(actions)) + return undefined; + issued.set(descriptor.targetId, actions); } } return issued;