fix(vscode): prevent early SSE replay loss
This commit is contained in:
@@ -185,7 +185,7 @@ export class AgentManagerPanelProvider {
|
|||||||
private async _startSseProxy(message: BridgeRequest): Promise<BridgeResponse> {
|
private async _startSseProxy(message: BridgeRequest): Promise<BridgeResponse> {
|
||||||
const { id, type, payload } = message;
|
const { id, type, payload } = message;
|
||||||
|
|
||||||
const { path, headers } = (payload || {}) as { path?: string; headers?: Record<string, string> };
|
const { path, headers, streamId: requestedStreamId } = (payload || {}) as { path?: string; headers?: Record<string, string>; streamId?: string };
|
||||||
const normalizedPath = typeof path === 'string' && path.trim().length > 0 ? path.trim() : '/event';
|
const normalizedPath = typeof path === 'string' && path.trim().length > 0 ? path.trim() : '/event';
|
||||||
|
|
||||||
if (!this._openCodeManager) {
|
if (!this._openCodeManager) {
|
||||||
@@ -197,8 +197,11 @@ export class AgentManagerPanelProvider {
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
const streamId = `sse_${++this._sseCounter}_${Date.now()}`;
|
const streamId = typeof requestedStreamId === 'string' && /^sse_webview_\d+_\d+$/.test(requestedStreamId)
|
||||||
|
? requestedStreamId
|
||||||
|
: `sse_${++this._sseCounter}_${Date.now()}`;
|
||||||
const controller = new AbortController();
|
const controller = new AbortController();
|
||||||
|
this._sseStreams.set(streamId, controller);
|
||||||
|
|
||||||
try {
|
try {
|
||||||
const start = await openSseProxy({
|
const start = await openSseProxy({
|
||||||
@@ -211,8 +214,6 @@ export class AgentManagerPanelProvider {
|
|||||||
},
|
},
|
||||||
});
|
});
|
||||||
|
|
||||||
this._sseStreams.set(streamId, controller);
|
|
||||||
|
|
||||||
start.run
|
start.run
|
||||||
.then(() => {
|
.then(() => {
|
||||||
this._panel?.webview.postMessage({ type: 'api:sse:end', streamId });
|
this._panel?.webview.postMessage({ type: 'api:sse:end', streamId });
|
||||||
@@ -238,6 +239,7 @@ export class AgentManagerPanelProvider {
|
|||||||
},
|
},
|
||||||
};
|
};
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
|
this._sseStreams.delete(streamId);
|
||||||
const message = error instanceof Error ? error.message : String(error);
|
const message = error instanceof Error ? error.message : String(error);
|
||||||
return {
|
return {
|
||||||
id,
|
id,
|
||||||
|
|||||||
@@ -536,7 +536,7 @@ export class ChatViewProvider implements vscode.WebviewViewProvider {
|
|||||||
private async _startSseProxy(message: BridgeRequest): Promise<BridgeResponse> {
|
private async _startSseProxy(message: BridgeRequest): Promise<BridgeResponse> {
|
||||||
const { id, type, payload } = message;
|
const { id, type, payload } = message;
|
||||||
|
|
||||||
const { path, headers } = (payload || {}) as { path?: string; headers?: Record<string, string> };
|
const { path, headers, streamId: requestedStreamId } = (payload || {}) as { path?: string; headers?: Record<string, string>; streamId?: string };
|
||||||
const normalizedPath = typeof path === 'string' && path.trim().length > 0 ? path.trim() : '/event';
|
const normalizedPath = typeof path === 'string' && path.trim().length > 0 ? path.trim() : '/event';
|
||||||
|
|
||||||
if (!this._openCodeManager) {
|
if (!this._openCodeManager) {
|
||||||
@@ -548,8 +548,11 @@ export class ChatViewProvider implements vscode.WebviewViewProvider {
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
const streamId = `sse_${++this._sseCounter}_${Date.now()}`;
|
const streamId = typeof requestedStreamId === 'string' && /^sse_webview_\d+_\d+$/.test(requestedStreamId)
|
||||||
|
? requestedStreamId
|
||||||
|
: `sse_${++this._sseCounter}_${Date.now()}`;
|
||||||
const controller = new AbortController();
|
const controller = new AbortController();
|
||||||
|
this._sseStreams.set(streamId, { controller, view: this._view });
|
||||||
|
|
||||||
try {
|
try {
|
||||||
const start = await openSseProxy({
|
const start = await openSseProxy({
|
||||||
@@ -562,8 +565,6 @@ export class ChatViewProvider implements vscode.WebviewViewProvider {
|
|||||||
},
|
},
|
||||||
});
|
});
|
||||||
|
|
||||||
this._sseStreams.set(streamId, { controller, view: this._view });
|
|
||||||
|
|
||||||
start.run
|
start.run
|
||||||
.then(() => {
|
.then(() => {
|
||||||
this._view?.webview.postMessage({ type: 'api:sse:end', streamId });
|
this._view?.webview.postMessage({ type: 'api:sse:end', streamId });
|
||||||
@@ -589,6 +590,7 @@ export class ChatViewProvider implements vscode.WebviewViewProvider {
|
|||||||
},
|
},
|
||||||
};
|
};
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
|
this._sseStreams.delete(streamId);
|
||||||
const message = error instanceof Error ? error.message : String(error);
|
const message = error instanceof Error ? error.message : String(error);
|
||||||
return {
|
return {
|
||||||
id,
|
id,
|
||||||
|
|||||||
@@ -47,6 +47,7 @@ The webview CSP permits `blob:` only for `worker-src` so shared UI parsers can r
|
|||||||
- `bridge-proxy-runtime.ts`
|
- `bridge-proxy-runtime.ts`
|
||||||
- Proxy route handlers (`api:proxy`, `api:session:message`) with injected helper dependencies.
|
- Proxy route handlers (`api:proxy`, `api:session:message`) with injected helper dependencies.
|
||||||
- SSE routes are intentionally excluded from the generic proxy and use `sseProxy.ts`, whose upstream-only stall watchdog closes a quiet OpenCode stream so the webview can reconnect instead of trusting an open but silent response.
|
- SSE routes are intentionally excluded from the generic proxy and use `sseProxy.ts`, whose upstream-only stall watchdog closes a quiet OpenCode stream so the webview can reconnect instead of trusting an open but silent response.
|
||||||
|
- The webview allocates each SSE stream ID and installs its listener before requesting the upstream stream, so immediate OpenCode replay events cannot race the bridge start response.
|
||||||
|
|
||||||
- `bridge-config-runtime.ts`
|
- `bridge-config-runtime.ts`
|
||||||
- Config and skills message handlers (`api:config/*`).
|
- Config and skills message handlers (`api:config/*`).
|
||||||
|
|||||||
@@ -415,7 +415,7 @@ export class SessionEditorPanelProvider {
|
|||||||
private async _startSseProxy(message: BridgeRequest, entry: SessionPanelState): Promise<BridgeResponse> {
|
private async _startSseProxy(message: BridgeRequest, entry: SessionPanelState): Promise<BridgeResponse> {
|
||||||
const { id, type, payload } = message;
|
const { id, type, payload } = message;
|
||||||
|
|
||||||
const { path, headers } = (payload || {}) as { path?: string; headers?: Record<string, string> };
|
const { path, headers, streamId: requestedStreamId } = (payload || {}) as { path?: string; headers?: Record<string, string>; streamId?: string };
|
||||||
const normalizedPath = typeof path === 'string' && path.trim().length > 0 ? path.trim() : '/event';
|
const normalizedPath = typeof path === 'string' && path.trim().length > 0 ? path.trim() : '/event';
|
||||||
|
|
||||||
if (!this._openCodeManager) {
|
if (!this._openCodeManager) {
|
||||||
@@ -427,8 +427,11 @@ export class SessionEditorPanelProvider {
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
const streamId = `sse_${++this._sseCounter}_${Date.now()}`;
|
const streamId = typeof requestedStreamId === 'string' && /^sse_webview_\d+_\d+$/.test(requestedStreamId)
|
||||||
|
? requestedStreamId
|
||||||
|
: `sse_${++this._sseCounter}_${Date.now()}`;
|
||||||
const controller = new AbortController();
|
const controller = new AbortController();
|
||||||
|
entry.sseStreams.set(streamId, controller);
|
||||||
|
|
||||||
try {
|
try {
|
||||||
const start = await openSseProxy({
|
const start = await openSseProxy({
|
||||||
@@ -442,8 +445,6 @@ export class SessionEditorPanelProvider {
|
|||||||
},
|
},
|
||||||
});
|
});
|
||||||
|
|
||||||
entry.sseStreams.set(streamId, controller);
|
|
||||||
|
|
||||||
start.run
|
start.run
|
||||||
.then(() => {
|
.then(() => {
|
||||||
entry.panel?.webview?.postMessage({ type: 'api:sse:end', streamId });
|
entry.panel?.webview?.postMessage({ type: 'api:sse:end', streamId });
|
||||||
@@ -469,6 +470,7 @@ export class SessionEditorPanelProvider {
|
|||||||
},
|
},
|
||||||
};
|
};
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
|
entry.sseStreams.delete(streamId);
|
||||||
const messageText = error instanceof Error ? error.message : String(error);
|
const messageText = error instanceof Error ? error.message : String(error);
|
||||||
return {
|
return {
|
||||||
id,
|
id,
|
||||||
|
|||||||
@@ -21,7 +21,7 @@ describe('VS Code webview bridge requests', () => {
|
|||||||
}),
|
}),
|
||||||
});
|
});
|
||||||
|
|
||||||
const { sendBridgeMessageWithOptions } = await import('./bridge');
|
const { sendBridgeMessageWithOptions, startSseProxy } = await import('./bridge');
|
||||||
const controller = new AbortController();
|
const controller = new AbortController();
|
||||||
controller.abort();
|
controller.abort();
|
||||||
|
|
||||||
@@ -36,6 +36,19 @@ describe('VS Code webview bridge requests', () => {
|
|||||||
assert.ok(result instanceof DOMException);
|
assert.ok(result instanceof DOMException);
|
||||||
assert.equal(result.name, 'AbortError');
|
assert.equal(result.name, 'AbortError');
|
||||||
assert.equal(messages.length, 0);
|
assert.equal(messages.length, 0);
|
||||||
|
|
||||||
|
const startPromise = startSseProxy({ path: '/global/event', streamId: 'sse_webview_1_1' });
|
||||||
|
const request = messages[0] as { id: string; payload?: { streamId?: string } };
|
||||||
|
assert.equal(request.payload?.streamId, 'sse_webview_1_1');
|
||||||
|
globalThis.window.dispatchEvent(new MessageEvent('message', {
|
||||||
|
data: {
|
||||||
|
id: request.id,
|
||||||
|
type: 'api:sse:start',
|
||||||
|
success: true,
|
||||||
|
data: { status: 200, headers: {}, streamId: 'sse_webview_1_1' },
|
||||||
|
},
|
||||||
|
}));
|
||||||
|
assert.equal((await startPromise).streamId, 'sse_webview_1_1');
|
||||||
} finally {
|
} finally {
|
||||||
Object.defineProperty(globalThis, 'window', { configurable: true, value: originalWindow });
|
Object.defineProperty(globalThis, 'window', { configurable: true, value: originalWindow });
|
||||||
Object.defineProperty(globalThis, 'acquireVsCodeApi', { configurable: true, value: originalAcquire });
|
Object.defineProperty(globalThis, 'acquireVsCodeApi', { configurable: true, value: originalAcquire });
|
||||||
|
|||||||
@@ -189,6 +189,7 @@ export type ProxiedSseStartResponse = {
|
|||||||
export async function startSseProxy(options: {
|
export async function startSseProxy(options: {
|
||||||
path: string;
|
path: string;
|
||||||
headers?: Record<string, string>;
|
headers?: Record<string, string>;
|
||||||
|
streamId?: string;
|
||||||
}): Promise<ProxiedSseStartResponse> {
|
}): Promise<ProxiedSseStartResponse> {
|
||||||
return sendBridgeMessage<ProxiedSseStartResponse>('api:sse:start', options);
|
return sendBridgeMessage<ProxiedSseStartResponse>('api:sse:start', options);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1122,6 +1122,7 @@ const handleLocalApiRequest = async (input: RequestInfo | URL, url: URL, init: R
|
|||||||
};
|
};
|
||||||
|
|
||||||
const originalFetch = window.fetch.bind(window);
|
const originalFetch = window.fetch.bind(window);
|
||||||
|
let sseStreamCounter = 0;
|
||||||
window.fetch = async (input: RequestInfo | URL, init?: RequestInit) => {
|
window.fetch = async (input: RequestInfo | URL, init?: RequestInit) => {
|
||||||
const targetUrl = typeof input === 'string' || input instanceof URL ? normalizeUrl(input) : normalizeUrl((input as Request).url);
|
const targetUrl = typeof input === 'string' || input instanceof URL ? normalizeUrl(input) : normalizeUrl((input as Request).url);
|
||||||
const method = (init?.method || (input instanceof Request ? input.method : 'GET')).toUpperCase();
|
const method = (init?.method || (input instanceof Request ? input.method : 'GET')).toUpperCase();
|
||||||
@@ -1160,12 +1161,10 @@ window.fetch = async (input: RequestInfo | URL, init?: RequestInit) => {
|
|||||||
const headers = { ...headersFromRequest, ...headersFromInit };
|
const headers = { ...headersFromRequest, ...headersFromInit };
|
||||||
|
|
||||||
if (isSseApiPath(targetUrl.pathname)) {
|
if (isSseApiPath(targetUrl.pathname)) {
|
||||||
const start = await vscodeStreamPerfMeasure('vscode.webview.sse_start_ms', () => startSseProxy({ path: suffixPath, headers }));
|
// Install the listener before the extension opens the upstream stream. A
|
||||||
if (!start.streamId) {
|
// reconnect can replay an event immediately, before the start response
|
||||||
return new Response(null, { status: start.status || 503, headers: start.headers || {} });
|
// has crossed the VS Code bridge.
|
||||||
}
|
const streamId = `sse_webview_${Date.now()}_${++sseStreamCounter}`;
|
||||||
|
|
||||||
const streamId = start.streamId;
|
|
||||||
const signal = (input instanceof Request ? input.signal : init?.signal) as AbortSignal | undefined;
|
const signal = (input instanceof Request ? input.signal : init?.signal) as AbortSignal | undefined;
|
||||||
const encoder = new TextEncoder();
|
const encoder = new TextEncoder();
|
||||||
let unsubscribe: (() => void) | null = null;
|
let unsubscribe: (() => void) | null = null;
|
||||||
@@ -1224,6 +1223,18 @@ window.fetch = async (input: RequestInfo | URL, init?: RequestInit) => {
|
|||||||
},
|
},
|
||||||
});
|
});
|
||||||
|
|
||||||
|
let start;
|
||||||
|
try {
|
||||||
|
start = await vscodeStreamPerfMeasure('vscode.webview.sse_start_ms', () => startSseProxy({ path: suffixPath, headers, streamId }));
|
||||||
|
} catch (error) {
|
||||||
|
await stream.cancel();
|
||||||
|
throw error;
|
||||||
|
}
|
||||||
|
if (!start.streamId) {
|
||||||
|
void stream.cancel();
|
||||||
|
return new Response(null, { status: start.status || 503, headers: start.headers || {} });
|
||||||
|
}
|
||||||
|
|
||||||
return new Response(stream, { status: start.status || 200, headers: start.headers || { 'content-type': 'text/event-stream' } });
|
return new Response(stream, { status: start.status || 200, headers: start.headers || { 'content-type': 'text/event-stream' } });
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user