cleanup: one shot vs stream

This commit is contained in:
Cristina Poncela Cubeiro
2026-06-22 14:53:07 +02:00
parent a7b0138ef0
commit d79e3061ad
5 changed files with 29 additions and 27 deletions
+1 -1
View File
@@ -41,7 +41,7 @@ async function rpcStream(instanceId: string): Promise<void> {
await new Promise<void>((resolve, reject) => {
socket.once("connect", () => {
socket.write(encodeMessage({ type: "rpc", instanceId }));
socket.write(encodeMessage({ type: "rpc_stream", instanceId }));
resolve();
});
socket.once("error", reject);
+15 -14
View File
@@ -15,6 +15,7 @@ import type {
RpcBridgeResponse,
RpcReadyResponse,
RpcRequest,
RpcStreamRequest,
SpawnRequest,
SpawnResponse,
StatusRequest,
@@ -50,9 +51,8 @@ export async function handleIpcRequest(request: SpawnRequest): Promise<SpawnResp
export async function handleIpcRequest(request: ListRequest): Promise<ListResponse | ErrorResponse>;
export async function handleIpcRequest(request: StopRequest): Promise<StopResponse | ErrorResponse>;
export async function handleIpcRequest(request: StatusRequest): Promise<StatusResponse | ErrorResponse>;
export async function handleIpcRequest(
request: RpcRequest,
): Promise<RpcBridgeResponse | RpcReadyResponse | ErrorResponse>;
export async function handleIpcRequest(request: RpcRequest): Promise<RpcBridgeResponse | ErrorResponse>;
export async function handleIpcRequest(request: RpcStreamRequest): Promise<RpcReadyResponse | ErrorResponse>;
export async function handleIpcRequest(request: OrchestratorRequest): Promise<OrchestratorResponse>;
export async function handleIpcRequest(request: OrchestratorRequest): Promise<OrchestratorResponse> {
switch (request.type) {
@@ -103,17 +103,6 @@ export async function handleIpcRequest(request: OrchestratorRequest): Promise<Or
}
case "rpc": {
if (!request.command) {
const instance = supervisor.getInstance(request.instanceId);
if (!instance) {
return unknownInstanceError(request.instanceId);
}
return {
type: "rpc_ready",
ok: true,
instance: toInstanceSummary(instance),
};
}
const response = await supervisor.handleRpc(request.instanceId, request.command);
if (!response) {
return unknownInstanceError(request.instanceId);
@@ -125,6 +114,18 @@ export async function handleIpcRequest(request: OrchestratorRequest): Promise<Or
response,
};
}
case "rpc_stream": {
const instance = supervisor.getInstance(request.instanceId);
if (!instance) {
return unknownInstanceError(request.instanceId);
}
return {
type: "rpc_ready",
ok: true,
instance: toInstanceSummary(instance),
};
}
}
}
+9 -2
View File
@@ -32,7 +32,12 @@ export interface StatusRequest {
export interface RpcRequest {
type: "rpc";
instanceId: string;
command?: RpcCommand;
command: RpcCommand;
}
export interface RpcStreamRequest {
type: "rpc_stream";
instanceId: string;
}
export interface RequestMap {
@@ -41,6 +46,7 @@ export interface RequestMap {
stop: StopRequest;
status: StatusRequest;
rpc: RpcRequest;
rpc_stream: RpcStreamRequest;
}
export type OrchestratorRequest = RequestMap[keyof RequestMap];
@@ -101,7 +107,8 @@ export interface ResponseMap {
list: ListResponse;
stop: StopResponse;
status: StatusResponse;
rpc: RpcBridgeResponse | RpcReadyResponse;
rpc: RpcBridgeResponse;
rpc_stream: RpcReadyResponse;
}
export type OrchestratorResponse = ResponseMap[keyof ResponseMap] | ErrorResponse;
+4 -8
View File
@@ -13,6 +13,7 @@ import {
type RpcClientMessage,
type RpcReadyResponse,
type RpcRequest,
type RpcStreamRequest,
type SpawnRequest,
type SpawnResponse,
type StatusRequest,
@@ -26,13 +27,8 @@ export interface IpcRequestHandler {
(request: ListRequest): Promise<ListResponse | ErrorResponse> | ListResponse | ErrorResponse;
(request: StopRequest): Promise<StopResponse | ErrorResponse> | StopResponse | ErrorResponse;
(request: StatusRequest): Promise<StatusResponse | ErrorResponse> | StatusResponse | ErrorResponse;
(
request: RpcRequest,
):
| Promise<RpcBridgeResponse | RpcReadyResponse | ErrorResponse>
| RpcBridgeResponse
| RpcReadyResponse
| ErrorResponse;
(request: RpcRequest): Promise<RpcBridgeResponse | ErrorResponse> | RpcBridgeResponse | ErrorResponse;
(request: RpcStreamRequest): Promise<RpcReadyResponse | ErrorResponse> | RpcReadyResponse | ErrorResponse;
(request: OrchestratorRequest): Promise<OrchestratorResponse> | OrchestratorResponse;
attach(
instanceId: string,
@@ -69,7 +65,7 @@ export async function startIpcServer(handler: IpcRequestHandler): Promise<Server
try {
const request = parseRequestLine(line);
if (request.type === "rpc" && request.command === undefined) {
if (request.type === "rpc_stream") {
const response = await handler(request);
if (!response.ok || response.type !== "rpc_ready" || !response.instance) {
socket.end(encodeMessage(response));
-2
View File
@@ -94,7 +94,6 @@ export class OrchestratorSupervisor {
| {
handleRpc(command: RpcCommand): Promise<RpcResponse>;
handleUiResponse(response: RpcExtensionUIResponse): void;
setHostTheme(theme: unknown): void;
close(): void;
}
| undefined {
@@ -113,7 +112,6 @@ export class OrchestratorSupervisor {
handleUiResponse: (response) => {
live.rpc.handleUiResponse(response);
},
setHostTheme: (_theme) => {},
close: () => {
if (live.onUiRequest === onUiRequest) {
live.onUiRequest = undefined;