From 8ad0cde994d9a1267c7edbb4cfa802acad53eb5c Mon Sep 17 00:00:00 2001 From: ytahdn <1294726970@qq.com> Date: Tue, 23 Jun 2026 19:40:32 +0800 Subject: [PATCH] feat(cli): add extension operation polling (#5753) * feat(cli): add extension operation polling * fix(cli): harden extension operation polling --------- Co-authored-by: ytahdn --- packages/cli/src/serve/server.test.ts | 229 +++++++++++++++++- packages/cli/src/serve/server.ts | 168 +++++++++++-- .../sdk-typescript/src/daemon/DaemonClient.ts | 10 + packages/sdk-typescript/src/daemon/index.ts | 3 + packages/sdk-typescript/src/daemon/types.ts | 31 +++ .../test/unit/DaemonClient.test.ts | 27 +++ .../webui/src/daemon/workspace/actions.ts | 11 + packages/webui/src/daemon/workspace/types.ts | 4 + 8 files changed, 468 insertions(+), 15 deletions(-) diff --git a/packages/cli/src/serve/server.test.ts b/packages/cli/src/serve/server.test.ts index 2b5642b94c..cfecca3c65 100644 --- a/packages/cli/src/serve/server.test.ts +++ b/packages/cli/src/serve/server.test.ts @@ -2698,7 +2698,8 @@ describe('createServeApp', () => { }); expect(res.status).toBe(202); - expect(res.body).toEqual({ accepted: true }); + expect(res.body).toMatchObject({ accepted: true }); + expect(res.body.operationId).toEqual(expect.any(String)); await vi.waitFor(() => { expect(bridge.extensionEvents.at(-1)).toMatchObject({ status: 'installed', @@ -2722,11 +2723,196 @@ describe('createServeApp', () => { expect(requestSettingError).toContain( 'requires interactive configuration', ); + + const poll = await request(app) + .get( + `/workspace/extensions/operations/${encodeURIComponent( + res.body.operationId as string, + )}`, + ) + .set('Host', `127.0.0.1:${tokenOpts.port}`) + .set('Authorization', 'Bearer secret'); + expect(poll.status).toBe(200); + expect(poll.body).toMatchObject({ + v: 1, + operationId: res.body.operationId, + operation: 'install', + status: 'succeeded', + source: 'https://example.com/installed-ext', + result: { + status: 'installed', + source: 'https://example.com/installed-ext', + name: 'installed-ext', + version: '1.2.3', + refreshed: 1, + failed: 0, + }, + }); } finally { restore(); } }); + it('reports queued and running extension operation states', async () => { + let releaseInstall: (() => void) | undefined; + const installBlocker = new Promise((resolve) => { + releaseInstall = resolve; + }); + const restore = mockExtensionManagerMethods({ + installExtension: async () => { + await installBlocker; + return testExtension('installed-ext'); + }, + }); + try { + const tokenOpts: ServeOptions = { ...baseOpts, token: 'secret' }; + const bridge = fakeBridge({ knownClientIds: ['client-1'] }); + const app = createServeApp( + { ...tokenOpts, workspace: WS_BOUND }, + undefined, + { bridge }, + ); + + const first = await request(app) + .post('/workspace/extensions/install') + .set('Host', `127.0.0.1:${tokenOpts.port}`) + .set('Authorization', 'Bearer secret') + .set('X-Qwen-Client-Id', 'client-1') + .send({ source: 'https://example.com/first-ext', consent: true }); + expect(first.status).toBe(202); + + await vi.waitFor(async () => { + const poll = await request(app) + .get( + `/workspace/extensions/operations/${encodeURIComponent( + first.body.operationId as string, + )}`, + ) + .set('Host', `127.0.0.1:${tokenOpts.port}`) + .set('Authorization', 'Bearer secret'); + expect(poll.body.status).toBe('running'); + }); + + const second = await request(app) + .post('/workspace/extensions/install') + .set('Host', `127.0.0.1:${tokenOpts.port}`) + .set('Authorization', 'Bearer secret') + .set('X-Qwen-Client-Id', 'client-1') + .send({ source: 'https://example.com/second-ext', consent: true }); + expect(second.status).toBe(202); + + const queued = await request(app) + .get( + `/workspace/extensions/operations/${encodeURIComponent( + second.body.operationId as string, + )}`, + ) + .set('Host', `127.0.0.1:${tokenOpts.port}`) + .set('Authorization', 'Bearer secret'); + expect(queued.status).toBe(200); + expect(queued.body).toMatchObject({ + operationId: second.body.operationId, + operation: 'install', + status: 'queued', + source: 'https://example.com/second-ext', + }); + + releaseInstall!(); + await vi.waitFor(() => { + expect(bridge.extensionEvents.length).toBeGreaterThanOrEqual(2); + }); + } finally { + releaseInstall?.(); + restore(); + } + }); + + it('evicts the oldest terminal extension operations', async () => { + const restore = mockExtensionManagerMethods({ + async installExtension() { + return testExtension('installed-ext'); + }, + }); + try { + const tokenOpts: ServeOptions = { ...baseOpts, token: 'secret' }; + const bridge = fakeBridge({ knownClientIds: ['client-1'] }); + const app = createServeApp( + { ...tokenOpts, workspace: WS_BOUND }, + undefined, + { bridge }, + ); + const operationIds: string[] = []; + + for (let i = 0; i < 101; i += 1) { + const res = await request(app) + .post('/workspace/extensions/install') + .set('Host', `127.0.0.1:${tokenOpts.port}`) + .set('Authorization', 'Bearer secret') + .set('X-Qwen-Client-Id', 'client-1') + .send({ + source: `https://example.com/installed-ext-${i}`, + consent: true, + }); + expect(res.status).toBe(202); + operationIds.push(res.body.operationId as string); + await vi.waitFor(async () => { + const poll = await request(app) + .get( + `/workspace/extensions/operations/${encodeURIComponent( + res.body.operationId as string, + )}`, + ) + .set('Host', `127.0.0.1:${tokenOpts.port}`) + .set('Authorization', 'Bearer secret'); + expect(poll.body.status).toBe('succeeded'); + }); + } + + const evicted = await request(app) + .get( + `/workspace/extensions/operations/${encodeURIComponent( + operationIds[0]!, + )}`, + ) + .set('Host', `127.0.0.1:${tokenOpts.port}`) + .set('Authorization', 'Bearer secret'); + expect(evicted.status).toBe(404); + + const retained = await request(app) + .get( + `/workspace/extensions/operations/${encodeURIComponent( + operationIds.at(-1)!, + )}`, + ) + .set('Host', `127.0.0.1:${tokenOpts.port}`) + .set('Authorization', 'Bearer secret'); + expect(retained.status).toBe(200); + expect(retained.body.status).toBe('succeeded'); + } finally { + restore(); + } + }); + + it('returns 404 for unknown extension operation ids', async () => { + const tokenOpts: ServeOptions = { ...baseOpts, token: 'secret' }; + const bridge = fakeBridge({ knownClientIds: ['client-1'] }); + const app = createServeApp( + { ...tokenOpts, workspace: WS_BOUND }, + undefined, + { bridge }, + ); + + const res = await request(app) + .get('/workspace/extensions/operations/missing-operation') + .set('Host', `127.0.0.1:${tokenOpts.port}`) + .set('Authorization', 'Bearer secret'); + + expect(res.status).toBe(404); + expect(res.body).toMatchObject({ + code: 'extension_operation_not_found', + }); + }); + it('broadcasts a failed extension install with redacted error details', async () => { const restore = mockExtensionManagerMethods({ installExtension: async () => { @@ -2753,6 +2939,7 @@ describe('createServeApp', () => { }); expect(res.status).toBe(202); + expect(res.body.operationId).toEqual(expect.any(String)); await vi.waitFor(() => { expect(bridge.extensionEvents.at(-1)).toMatchObject({ status: 'failed', @@ -2763,6 +2950,24 @@ describe('createServeApp', () => { }); }); expect(bridge.extensionEvents.at(-1)?.error).not.toContain('token'); + + const poll = await request(app) + .get( + `/workspace/extensions/operations/${encodeURIComponent( + res.body.operationId as string, + )}`, + ) + .set('Host', `127.0.0.1:${tokenOpts.port}`) + .set('Authorization', 'Bearer secret'); + expect(poll.status).toBe(200); + expect(poll.body).toMatchObject({ + operationId: res.body.operationId, + operation: 'install', + status: 'failed', + source: 'https://example.com/private-ext', + error: 'https://***REDACTED***@example.com/private-ext failed', + }); + expect(poll.body.error).not.toContain('token'); } finally { restore(); } @@ -2797,6 +3002,7 @@ describe('createServeApp', () => { }); expect(res.status).toBe(202); + expect(res.body.operationId).toEqual(expect.any(String)); await vi.waitFor(() => { expect(bridge.extensionEvents.at(-1)).toMatchObject({ status: 'installed', @@ -2807,6 +3013,27 @@ describe('createServeApp', () => { error: 'refresh broke', }); }); + + const poll = await request(app) + .get( + `/workspace/extensions/operations/${encodeURIComponent( + res.body.operationId as string, + )}`, + ) + .set('Host', `127.0.0.1:${tokenOpts.port}`) + .set('Authorization', 'Bearer secret'); + expect(poll.status).toBe(200); + expect(poll.body).toMatchObject({ + operationId: res.body.operationId, + operation: 'install', + status: 'succeeded_with_refresh_error', + result: { + status: 'installed', + refreshed: 0, + failed: 1, + error: 'refresh broke', + }, + }); } finally { restore(); } diff --git a/packages/cli/src/serve/server.ts b/packages/cli/src/serve/server.ts index 4cba80e524..3d921e7a5e 100644 --- a/packages/cli/src/serve/server.ts +++ b/packages/cli/src/serve/server.ts @@ -1253,7 +1253,7 @@ export function createServeApp( extensionInstallQueue = next.catch(() => undefined); return next; }; - const EXTENSION_MUTATION_TIMEOUT_MS = 120_000; + const EXTENSION_MUTATION_TIMEOUT_MS = 10 * 60_000; const EXTENSION_REFRESH_TIMEOUT_MS = 30_000; const isExtensionQueueFullError = (err: unknown): boolean => err instanceof Error && err.message === 'Extension operation queue is full'; @@ -1427,6 +1427,65 @@ export function createServeApp( name?: string; version?: string; }; + type ExtensionOperationStatus = { + v: 1; + operationId: string; + operation: string; + status: + | 'queued' + | 'running' + | 'succeeded' + | 'succeeded_with_refresh_error' + | 'failed'; + createdAt: number; + updatedAt: number; + source?: string; + name?: string; + result?: ExtensionMutationEvent & { + refreshed?: number; + failed?: number; + error?: string; + }; + error?: string; + }; + const extensionOperations = new Map(); + const MAX_EXTENSION_OPERATION_HISTORY = 100; + const isTerminalExtensionOperation = ( + operation: ExtensionOperationStatus, + ): boolean => operation.status !== 'queued' && operation.status !== 'running'; + const redactExtensionOperationResult = ( + event: ExtensionMutationEvent, + ): ExtensionMutationEvent => ({ + ...event, + ...(event.source ? { source: redactUrlCredentials(event.source) } : {}), + }); + const rememberExtensionOperation = ( + operation: ExtensionOperationStatus, + ): void => { + extensionOperations.set(operation.operationId, operation); + while (extensionOperations.size > MAX_EXTENSION_OPERATION_HISTORY) { + let evicted = false; + for (const [id, storedOperation] of extensionOperations) { + if (!isTerminalExtensionOperation(storedOperation)) continue; + extensionOperations.delete(id); + evicted = true; + break; + } + if (!evicted) break; + } + }; + const updateExtensionOperation = ( + operationId: string, + patch: Partial>, + ): void => { + const current = extensionOperations.get(operationId); + if (!current) return; + extensionOperations.set(operationId, { + ...current, + ...patch, + updatedAt: Date.now(), + }); + }; const runQueuedExtensionMutation = ( operation: string, failureContext: { source?: string; name?: string }, @@ -1439,9 +1498,24 @@ export function createServeApp( sendExtensionQueueFull(res); return; } - res.status(202).json({ accepted: true }); + const operationId = crypto.randomUUID(); + const now = Date.now(); + rememberExtensionOperation({ + v: 1, + operationId, + operation, + status: 'queued', + createdAt: now, + updatedAt: now, + ...(failureContext.source + ? { source: redactUrlCredentials(failureContext.source) } + : {}), + ...(failureContext.name ? { name: failureContext.name } : {}), + }); + res.status(202).json({ accepted: true, operationId }); void enqueueExtensionInstall(async () => { try { + updateExtensionOperation(operationId, { status: 'running' }); const extensionManager = createExtensionManager(); await extensionManager.refreshCache(); const event = await withExtensionTimeout( @@ -1452,6 +1526,14 @@ export function createServeApp( extensionsStatusCache = undefined; try { const result = await bridge.refreshExtensionsForAllSessions(event); + updateExtensionOperation(operationId, { + status: 'succeeded', + result: { + ...redactExtensionOperationResult(event), + refreshed: result.refreshed, + failed: result.failed, + }, + }); writeStderrLine( `qwen serve: extensions ${operation}: refreshed ${result.refreshed} session(s), ${result.failed} failed`, ); @@ -1461,6 +1543,15 @@ export function createServeApp( ? refreshErr.message : String(refreshErr), ); + updateExtensionOperation(operationId, { + status: 'succeeded_with_refresh_error', + result: { + ...redactExtensionOperationResult(event), + refreshed: 0, + failed: 1, + error: message.slice(0, 500), + }, + }); try { bridge.broadcastExtensionsChanged({ ...event, @@ -1485,6 +1576,10 @@ export function createServeApp( const message = redactUrlCredentials( err instanceof Error ? err.message : String(err), ); + updateExtensionOperation(operationId, { + status: 'failed', + error: message.slice(0, 500), + }); try { bridge.broadcastExtensionsChanged({ status: 'failed', @@ -1514,11 +1609,16 @@ export function createServeApp( } } }).catch((err) => { + const message = redactUrlCredentials( + err instanceof Error ? err.message : String(err), + ); + updateExtensionOperation(operationId, { + status: 'failed', + error: message.slice(0, 500), + }); try { writeStderrLine( - `qwen serve: extensions ${operation}: queued task failed: ${ - err instanceof Error ? err.message : String(err) - }`, + `qwen serve: extensions ${operation}: queued task failed: ${message}`, ); } catch { // Last-resort guard for detached async work. @@ -2094,6 +2194,33 @@ export function createServeApp( } }); + app.get('/workspace/extensions/operations/:operationId', async (req, res) => { + try { + buildWorkspaceCtx( + req, + 'GET /workspace/extensions/operations/:operationId', + ); + const operationId = req.params['operationId']; + if (!operationId) { + res.status(400).json({ error: 'Missing extension operation id' }); + return; + } + const operation = extensionOperations.get(operationId); + if (!operation) { + res.status(404).json({ + error: `Extension operation "${operationId}" not found`, + code: 'extension_operation_not_found', + }); + return; + } + res.status(200).json(operation); + } catch (err) { + sendBridgeError(res, err, { + route: 'GET /workspace/extensions/operations/:operationId', + }); + } + }); + // POST /workspace/extensions/install — install an extension and refresh // all active sessions asynchronously. app.post( @@ -2147,9 +2274,17 @@ export function createServeApp( res.status(400).json({ error: '`registry` must be a string' }); return; } + const sourceValue = source; + const refValue = typeof ref === 'string' ? ref : undefined; + const autoUpdateValue = + typeof autoUpdate === 'boolean' ? autoUpdate : undefined; + const allowPreReleaseValue = + typeof allowPreRelease === 'boolean' ? allowPreRelease : undefined; + const registryValue = + typeof registry === 'string' ? registry : undefined; const registryUrl = - registry !== undefined - ? parseExtensionRegistryUrl(registry, res) + registryValue !== undefined + ? parseExtensionRegistryUrl(registryValue, res) : undefined; if (registryUrl === null) return; if (consent !== true) { @@ -2158,16 +2293,16 @@ export function createServeApp( }); return; } - if (!validateExtensionSourceHost(source, res)) { + if (!validateExtensionSourceHost(sourceValue, res)) { return; } runQueuedExtensionMutation( 'install', - { source }, + { source: sourceValue }, res, async (extensionManager) => { - const installMetadata = await parseInstallSource(source); + const installMetadata = await parseInstallSource(sourceValue); if ( installMetadata.type !== 'git' && @@ -2178,10 +2313,10 @@ export function createServeApp( 'Only GitHub, Git, and npm extension installs are supported over the daemon endpoint.', ); } - if (installMetadata.type === 'npm' && ref) { + if (installMetadata.type === 'npm' && refValue) { throw new Error('--ref is not applicable for npm extensions.'); } - if (installMetadata.type !== 'npm' && registry) { + if (installMetadata.type !== 'npm' && registryValue) { throw new Error( '--registry is only applicable for npm extensions.', ); @@ -2193,12 +2328,17 @@ export function createServeApp( installMetadata.registryUrl = registryUrl; } const extension = await extensionManager.installExtension( - { ...installMetadata, ref, autoUpdate, allowPreRelease }, + { + ...installMetadata, + ref: refValue, + autoUpdate: autoUpdateValue, + allowPreRelease: allowPreReleaseValue, + }, () => Promise.resolve(), ); return { status: 'installed', - source, + source: sourceValue, name: extension.name, version: extension.config.version, }; diff --git a/packages/sdk-typescript/src/daemon/DaemonClient.ts b/packages/sdk-typescript/src/daemon/DaemonClient.ts index 23c40ebf6b..47fc8922a9 100644 --- a/packages/sdk-typescript/src/daemon/DaemonClient.ts +++ b/packages/sdk-typescript/src/daemon/DaemonClient.ts @@ -87,6 +87,7 @@ import type { ExtensionMutationResponse, ExtensionInstallRequest, ExtensionInstallResponse, + ExtensionOperationStatus, ExtensionScopeRequest, ExtensionRefreshResponse, ExtensionUpdateCheckResponse, @@ -711,6 +712,15 @@ export class DaemonClient { ); } + async extensionOperationStatus( + operationId: string, + ): Promise { + return await this.jsonRequest( + `/workspace/extensions/operations/${encodeURIComponent(operationId)}`, + 'GET /workspace/extensions/operations/:operationId', + ); + } + async checkExtensionUpdates( clientId?: string, ): Promise { diff --git a/packages/sdk-typescript/src/daemon/index.ts b/packages/sdk-typescript/src/daemon/index.ts index 7d64e69fce..e1bbbfeafb 100644 --- a/packages/sdk-typescript/src/daemon/index.ts +++ b/packages/sdk-typescript/src/daemon/index.ts @@ -424,6 +424,9 @@ export type { ExtensionInstallRequest, ExtensionInstallResponse, ExtensionMutationResponse, + ExtensionOperationResult, + ExtensionOperationState, + ExtensionOperationStatus, ExtensionRefreshResponse, ExtensionScope, ExtensionScopeRequest, diff --git a/packages/sdk-typescript/src/daemon/types.ts b/packages/sdk-typescript/src/daemon/types.ts index 4016e0159b..25a7aaa9a8 100644 --- a/packages/sdk-typescript/src/daemon/types.ts +++ b/packages/sdk-typescript/src/daemon/types.ts @@ -1777,10 +1777,41 @@ export interface ExtensionInstallRequest { export interface ExtensionInstallResponse { accepted: true; + operationId: string; } export type ExtensionMutationResponse = ExtensionInstallResponse; +export type ExtensionOperationState = + | 'queued' + | 'running' + | 'succeeded' + | 'succeeded_with_refresh_error' + | 'failed'; + +export interface ExtensionOperationResult { + status: 'installed' | 'enabled' | 'disabled' | 'updated' | 'uninstalled'; + source?: string; + name?: string; + version?: string; + refreshed?: number; + failed?: number; + error?: string; +} + +export interface ExtensionOperationStatus { + v: 1; + operationId: string; + operation: string; + status: ExtensionOperationState; + createdAt: number; + updatedAt: number; + source?: string; + name?: string; + result?: ExtensionOperationResult; + error?: string; +} + export type ExtensionScope = 'user' | 'workspace'; export interface ExtensionScopeRequest { diff --git a/packages/sdk-typescript/test/unit/DaemonClient.test.ts b/packages/sdk-typescript/test/unit/DaemonClient.test.ts index aa6f747172..b8166743fd 100644 --- a/packages/sdk-typescript/test/unit/DaemonClient.test.ts +++ b/packages/sdk-typescript/test/unit/DaemonClient.test.ts @@ -2350,6 +2350,33 @@ describe('DaemonClient', () => { }); }); + describe('extension operations', () => { + it('GETs an extension operation status by id', async () => { + const { fetch, calls } = recordingFetch(() => + jsonResponse(200, { + v: 1, + operationId: 'op/1', + operation: 'install', + status: 'succeeded', + createdAt: 1, + updatedAt: 2, + }), + ); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + + const result = await client.extensionOperationStatus('op/1'); + + expect(calls[0]?.url).toBe( + 'http://daemon/workspace/extensions/operations/op%2F1', + ); + expect(calls[0]?.method).toBe('GET'); + expect(result).toMatchObject({ + operationId: 'op/1', + status: 'succeeded', + }); + }); + }); + describe('error coercion', () => { it('falls back to text body when the response is not JSON', async () => { const { fetch } = recordingFetch( diff --git a/packages/webui/src/daemon/workspace/actions.ts b/packages/webui/src/daemon/workspace/actions.ts index 3f05348bce..6d3e351b98 100644 --- a/packages/webui/src/daemon/workspace/actions.ts +++ b/packages/webui/src/daemon/workspace/actions.ts @@ -361,6 +361,17 @@ export function createDaemonWorkspaceActions({ ); }, + async extensionOperationStatus(operationId) { + const client = requireClient( + getClient, + 'Load extension operation failed', + ); + return withActionTimeout( + client.extensionOperationStatus(operationId), + 'Load extension operation timed out', + ); + }, + async checkExtensionUpdates(clientId) { const client = requireClient(getClient, 'Check extension updates failed'); return withActionTimeout( diff --git a/packages/webui/src/daemon/workspace/types.ts b/packages/webui/src/daemon/workspace/types.ts index b776c00c36..2d80e90446 100644 --- a/packages/webui/src/daemon/workspace/types.ts +++ b/packages/webui/src/daemon/workspace/types.ts @@ -19,6 +19,7 @@ import type { DaemonDeviceFlowStartResult, DaemonDeviceFlowState, ExtensionMutationResponse, + ExtensionOperationStatus, ExtensionRefreshResponse, ExtensionScopeRequest, ExtensionInstallRequest, @@ -227,6 +228,9 @@ export interface DaemonWorkspaceActions { params: ExtensionInstallRequest, clientId?: string, ): Promise; + extensionOperationStatus( + operationId: string, + ): Promise; checkExtensionUpdates( clientId?: string, ): Promise;