alezmad/claudemesh
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(cli): correlation ID refactor — resolver Maps with _reqId + FIFO fallback

Browse Source 
Replace all 22 resolver Array<fn> patterns with Map<reqId, {resolve, timer}>.
Outgoing messages now include _reqId; on response the broker's echoed _reqId
is used for exact matching, with FIFO fallback for brokers that don't echo it.
Add makeReqId() helper and resolveFromMap() utility. Error propagation block
updated to iterate Maps and pop the oldest entry across all queues.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Alejandro Gutiérrez
2026-04-07 14:25:51 +01:00
parent 4ce1034dcd
commit 8f925d9a9e
1 changed files with 253 additions and 299 deletions
Download Patch File Download Diff File Expand all files Collapse all files

512
apps/cli/src/ws/client.ts
View File

@@ -75,15 +75,15 @@ export class BrokerClient {
private outbound: Array<() => void> = []; // closures that send once ws is open private outbound: Array<() => void> = []; // closures that send once ws is open
private pushHandlers = new Set<PushHandler>(); private pushHandlers = new Set<PushHandler>();
private pushBuffer: InboundPush[] = []; private pushBuffer: InboundPush[] = [];
private listPeersResolvers: Array<(peers: PeerInfo[]) => void> = []; private listPeersResolvers = new Map<string, { resolve: (peers: PeerInfo[]) => void; timer: NodeJS.Timeout }>();
private stateResolvers: Array<(result: { key: string; value: unknown; updatedBy: string; updatedAt: string } | null) => void> = []; private stateResolvers = new Map<string, { resolve: (result: { key: string; value: unknown; updatedBy: string; updatedAt: string } | null) => void; timer: NodeJS.Timeout }>();
private stateListResolvers: Array<(entries: Array<{ key: string; value: unknown; updatedBy: string; updatedAt: string }>) => void> = []; private stateListResolvers = new Map<string, { resolve: (entries: Array<{ key: string; value: unknown; updatedBy: string; updatedAt: string }>) => void; timer: NodeJS.Timeout }>();
private memoryStoreResolvers: Array<(id: string | null) => void> = []; private memoryStoreResolvers = new Map<string, { resolve: (id: string | null) => void; timer: NodeJS.Timeout }>();
private memoryRecallResolvers: Array<(memories: Array<{ id: string; content: string; tags: string[]; rememberedBy: string; rememberedAt: string }>) => void> = []; private memoryRecallResolvers = new Map<string, { resolve: (memories: Array<{ id: string; content: string; tags: string[]; rememberedBy: string; rememberedAt: string }>) => void; timer: NodeJS.Timeout }>();
private stateChangeHandlers = new Set<(change: { key: string; value: unknown; updatedBy: string }) => void>(); private stateChangeHandlers = new Set<(change: { key: string; value: unknown; updatedBy: string }) => void>();
private sessionPubkey: string | null = null; private sessionPubkey: string | null = null;
private sessionSecretKey: string | null = null; private sessionSecretKey: string | null = null;
private grantFileAccessResolvers: Array<(ok: boolean) => void> = []; private grantFileAccessResolvers = new Map<string, { resolve: (ok: boolean) => void; timer: NodeJS.Timeout }>();
private closed = false; private closed = false;
private reconnectAttempt = 0; private reconnectAttempt = 0;
private helloTimer: NodeJS.Timeout | null = null; private helloTimer: NodeJS.Timeout | null = null;
@@ -116,6 +116,10 @@ export class BrokerClient {
/** Session secret key hex (null before first connection). */ /** Session secret key hex (null before first connection). */
getSessionSecretKey(): string | null { return this.sessionSecretKey; } getSessionSecretKey(): string | null { return this.sessionSecretKey; }
private makeReqId(): string {
return Math.random().toString(36).slice(2) + Date.now().toString(36);
}
/** Open WS, send hello, resolve when hello_ack received. */ /** Open WS, send hello, resolve when hello_ack received. */
async connect(): Promise<void> { async connect(): Promise<void> {
if (this.closed) throw new Error("client is closed"); if (this.closed) throw new Error("client is closed");
@@ -305,16 +309,11 @@ export class BrokerClient {
async listPeers(): Promise<PeerInfo[]> { async listPeers(): Promise<PeerInfo[]> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return []; if (!this.ws || this.ws.readyState !== this.ws.OPEN) return [];
return new Promise((resolve) => { return new Promise((resolve) => {
this.listPeersResolvers.push(resolve); const reqId = this.makeReqId();
this.ws!.send(JSON.stringify({ type: "list_peers" })); this.listPeersResolvers.set(reqId, { resolve, timer: setTimeout(() => {
// Timeout after 5s — return empty list rather than hang. if (this.listPeersResolvers.delete(reqId)) resolve([]);
setTimeout(() => { }, 5_000) });
const idx = this.listPeersResolvers.indexOf(resolve); this.ws!.send(JSON.stringify({ type: "list_peers", _reqId: reqId }));
if (idx !== -1) {
this.listPeersResolvers.splice(idx, 1);
resolve([]);
}
}, 5_000);
}); });
} }
@@ -348,15 +347,11 @@ export class BrokerClient {
async getState(key: string): Promise<{ key: string; value: unknown; updatedBy: string; updatedAt: string } | null> { async getState(key: string): Promise<{ key: string; value: unknown; updatedBy: string; updatedAt: string } | null> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return null; if (!this.ws || this.ws.readyState !== this.ws.OPEN) return null;
return new Promise((resolve) => { return new Promise((resolve) => {
this.stateResolvers.push(resolve); const reqId = this.makeReqId();
this.ws!.send(JSON.stringify({ type: "get_state", key })); this.stateResolvers.set(reqId, { resolve, timer: setTimeout(() => {
setTimeout(() => { if (this.stateResolvers.delete(reqId)) resolve(null);
const idx = this.stateResolvers.indexOf(resolve); }, 5_000) });
if (idx !== -1) { this.ws!.send(JSON.stringify({ type: "get_state", key, _reqId: reqId }));
this.stateResolvers.splice(idx, 1);
resolve(null);
}
}, 5_000);
}); });
} }
@@ -364,15 +359,11 @@ export class BrokerClient {
async listState(): Promise<Array<{ key: string; value: unknown; updatedBy: string; updatedAt: string }>> { async listState(): Promise<Array<{ key: string; value: unknown; updatedBy: string; updatedAt: string }>> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return []; if (!this.ws || this.ws.readyState !== this.ws.OPEN) return [];
return new Promise((resolve) => { return new Promise((resolve) => {
this.stateListResolvers.push(resolve); const reqId = this.makeReqId();
this.ws!.send(JSON.stringify({ type: "list_state" })); this.stateListResolvers.set(reqId, { resolve, timer: setTimeout(() => {
setTimeout(() => { if (this.stateListResolvers.delete(reqId)) resolve([]);
const idx = this.stateListResolvers.indexOf(resolve); }, 5_000) });
if (idx !== -1) { this.ws!.send(JSON.stringify({ type: "list_state", _reqId: reqId }));
this.stateListResolvers.splice(idx, 1);
resolve([]);
}
}, 5_000);
}); });
} }
@@ -382,15 +373,11 @@ export class BrokerClient {
async remember(content: string, tags?: string[]): Promise<string | null> { async remember(content: string, tags?: string[]): Promise<string | null> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return null; if (!this.ws || this.ws.readyState !== this.ws.OPEN) return null;
return new Promise((resolve) => { return new Promise((resolve) => {
this.memoryStoreResolvers.push(resolve); const reqId = this.makeReqId();
this.ws!.send(JSON.stringify({ type: "remember", content, tags })); this.memoryStoreResolvers.set(reqId, { resolve, timer: setTimeout(() => {
setTimeout(() => { if (this.memoryStoreResolvers.delete(reqId)) resolve(null);
const idx = this.memoryStoreResolvers.indexOf(resolve); }, 5_000) });
if (idx !== -1) { this.ws!.send(JSON.stringify({ type: "remember", content, tags, _reqId: reqId }));
this.memoryStoreResolvers.splice(idx, 1);
resolve(null);
}
}, 5_000);
}); });
} }
@@ -398,15 +385,11 @@ export class BrokerClient {
async recall(query: string): Promise<Array<{ id: string; content: string; tags: string[]; rememberedBy: string; rememberedAt: string }>> { async recall(query: string): Promise<Array<{ id: string; content: string; tags: string[]; rememberedBy: string; rememberedAt: string }>> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return []; if (!this.ws || this.ws.readyState !== this.ws.OPEN) return [];
return new Promise((resolve) => { return new Promise((resolve) => {
this.memoryRecallResolvers.push(resolve); const reqId = this.makeReqId();
this.ws!.send(JSON.stringify({ type: "recall", query })); this.memoryRecallResolvers.set(reqId, { resolve, timer: setTimeout(() => {
setTimeout(() => { if (this.memoryRecallResolvers.delete(reqId)) resolve([]);
const idx = this.memoryRecallResolvers.indexOf(resolve); }, 5_000) });
if (idx !== -1) { this.ws!.send(JSON.stringify({ type: "recall", query, _reqId: reqId }));
this.memoryRecallResolvers.splice(idx, 1);
resolve([]);
}
}, 5_000);
}); });
} }
@@ -417,33 +400,32 @@ export class BrokerClient {
} }
/** Check delivery status of a sent message. */ /** Check delivery status of a sent message. */
private messageStatusResolvers: Array<(result: { messageId: string; targetSpec: string; delivered: boolean; deliveredAt: string | null; recipients: Array<{ name: string; pubkey: string; status: string }> } | null) => void> = []; private messageStatusResolvers = new Map<string, { resolve: (result: { messageId: string; targetSpec: string; delivered: boolean; deliveredAt: string | null; recipients: Array<{ name: string; pubkey: string; status: string }> } | null) => void; timer: NodeJS.Timeout }>();
private fileUrlResolvers: Array<(result: { url: string; name: string; encrypted?: boolean; sealedKey?: string } | null) => void> = []; private fileUrlResolvers = new Map<string, { resolve: (result: { url: string; name: string; encrypted?: boolean; sealedKey?: string } | null) => void; timer: NodeJS.Timeout }>();
private fileListResolvers: Array<(files: Array<{ id: string; name: string; size: number; tags: string[]; uploadedBy: string; uploadedAt: string; persistent: boolean }>) => void> = []; private fileListResolvers = new Map<string, { resolve: (files: Array<{ id: string; name: string; size: number; tags: string[]; uploadedBy: string; uploadedAt: string; persistent: boolean }>) => void; timer: NodeJS.Timeout }>();
private fileStatusResolvers: Array<(accesses: Array<{ peerName: string; accessedAt: string }>) => void> = []; private fileStatusResolvers = new Map<string, { resolve: (accesses: Array<{ peerName: string; accessedAt: string }>) => void; timer: NodeJS.Timeout }>();
private vectorStoredResolvers: Array<(id: string | null) => void> = []; private vectorStoredResolvers = new Map<string, { resolve: (id: string | null) => void; timer: NodeJS.Timeout }>();
private vectorResultsResolvers: Array<(results: Array<{ id: string; text: string; score: number; metadata?: Record<string, unknown> }>) => void> = []; private vectorResultsResolvers = new Map<string, { resolve: (results: Array<{ id: string; text: string; score: number; metadata?: Record<string, unknown> }>) => void; timer: NodeJS.Timeout }>();
private collectionListResolvers: Array<(collections: string[]) => void> = []; private collectionListResolvers = new Map<string, { resolve: (collections: string[]) => void; timer: NodeJS.Timeout }>();
private graphResultResolvers: Array<(rows: Array<Record<string, unknown>>) => void> = []; private graphResultResolvers = new Map<string, { resolve: (rows: Array<Record<string, unknown>>) => void; timer: NodeJS.Timeout }>();
private contextListResolvers: Array<(contexts: Array<{ peerName: string; summary: string; tags: string[]; updatedAt: string }>) => void> = []; private contextListResolvers = new Map<string, { resolve: (contexts: Array<{ peerName: string; summary: string; tags: string[]; updatedAt: string }>) => void; timer: NodeJS.Timeout }>();
private contextResultsResolvers: Array<(contexts: Array<{ peerName: string; summary: string; filesRead: string[]; keyFindings: string[]; tags: string[]; updatedAt: string }>) => void> = []; private contextResultsResolvers = new Map<string, { resolve: (contexts: Array<{ peerName: string; summary: string; filesRead: string[]; keyFindings: string[]; tags: string[]; updatedAt: string }>) => void; timer: NodeJS.Timeout }>();
private taskCreatedResolvers: Array<(id: string | null) => void> = []; private taskCreatedResolvers = new Map<string, { resolve: (id: string | null) => void; timer: NodeJS.Timeout }>();
private taskListResolvers: Array<(tasks: Array<{ id: string; title: string; assignee: string; status: string; priority: string; createdBy: string }>) => void> = []; private taskListResolvers = new Map<string, { resolve: (tasks: Array<{ id: string; title: string; assignee: string; status: string; priority: string; createdBy: string }>) => void; timer: NodeJS.Timeout }>();
private meshQueryResolvers: Array<(result: { columns: string[]; rows: Array<Record<string, unknown>>; rowCount: number } | null) => void> = []; private meshQueryResolvers = new Map<string, { resolve: (result: { columns: string[]; rows: Array<Record<string, unknown>>; rowCount: number } | null) => void; timer: NodeJS.Timeout }>();
private meshSchemaResolvers: Array<(tables: Array<{ name: string; columns: Array<{ name: string; type: string; nullable: boolean }> }>) => void> = []; private meshSchemaResolvers = new Map<string, { resolve: (tables: Array<{ name: string; columns: Array<{ name: string; type: string; nullable: boolean }> }>) => void; timer: NodeJS.Timeout }>();
private streamCreatedResolvers: Array<(id: string | null) => void> = []; private streamCreatedResolvers = new Map<string, { resolve: (id: string | null) => void; timer: NodeJS.Timeout }>();
private streamListResolvers: Array<(streams: Array<{ id: string; name: string; createdBy: string; subscriberCount: number }>) => void> = []; private streamListResolvers = new Map<string, { resolve: (streams: Array<{ id: string; name: string; createdBy: string; subscriberCount: number }>) => void; timer: NodeJS.Timeout }>();
private streamDataHandlers = new Set<(data: { stream: string; data: unknown; publishedBy: string }) => void>(); private streamDataHandlers = new Set<(data: { stream: string; data: unknown; publishedBy: string }) => void>();
async messageStatus(messageId: string): Promise<{ messageId: string; targetSpec: string; delivered: boolean; deliveredAt: string | null; recipients: Array<{ name: string; pubkey: string; status: string }> } | null> { async messageStatus(messageId: string): Promise<{ messageId: string; targetSpec: string; delivered: boolean; deliveredAt: string | null; recipients: Array<{ name: string; pubkey: string; status: string }> } | null> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return null; if (!this.ws || this.ws.readyState !== this.ws.OPEN) return null;
return new Promise((resolve) => { return new Promise((resolve) => {
this.messageStatusResolvers.push(resolve); const reqId = this.makeReqId();
this.ws!.send(JSON.stringify({ type: "message_status", messageId })); this.messageStatusResolvers.set(reqId, { resolve, timer: setTimeout(() => {
setTimeout(() => { if (this.messageStatusResolvers.delete(reqId)) resolve(null);
const idx = this.messageStatusResolvers.indexOf(resolve); }, 5_000) });
if (idx !== -1) { this.messageStatusResolvers.splice(idx, 1); resolve(null); } this.ws!.send(JSON.stringify({ type: "message_status", messageId, _reqId: reqId }));
}, 5_000);
}); });
} }
@@ -453,15 +435,11 @@ export class BrokerClient {
async getFile(fileId: string): Promise<{ url: string; name: string; encrypted?: boolean; sealedKey?: string } | null> { async getFile(fileId: string): Promise<{ url: string; name: string; encrypted?: boolean; sealedKey?: string } | null> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return null; if (!this.ws || this.ws.readyState !== this.ws.OPEN) return null;
return new Promise((resolve) => { return new Promise((resolve) => {
this.fileUrlResolvers.push(resolve); const reqId = this.makeReqId();
this.ws!.send(JSON.stringify({ type: "get_file", fileId })); this.fileUrlResolvers.set(reqId, { resolve, timer: setTimeout(() => {
setTimeout(() => { if (this.fileUrlResolvers.delete(reqId)) resolve(null);
const idx = this.fileUrlResolvers.indexOf(resolve); }, 5_000) });
if (idx !== -1) { this.ws!.send(JSON.stringify({ type: "get_file", fileId, _reqId: reqId }));
this.fileUrlResolvers.splice(idx, 1);
resolve(null);
}
}, 5_000);
}); });
} }
@@ -469,15 +447,11 @@ export class BrokerClient {
async listFiles(query?: string, from?: string): Promise<Array<{ id: string; name: string; size: number; tags: string[]; uploadedBy: string; uploadedAt: string; persistent: boolean }>> { async listFiles(query?: string, from?: string): Promise<Array<{ id: string; name: string; size: number; tags: string[]; uploadedBy: string; uploadedAt: string; persistent: boolean }>> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return []; if (!this.ws || this.ws.readyState !== this.ws.OPEN) return [];
return new Promise((resolve) => { return new Promise((resolve) => {
this.fileListResolvers.push(resolve); const reqId = this.makeReqId();
this.ws!.send(JSON.stringify({ type: "list_files", query, from })); this.fileListResolvers.set(reqId, { resolve, timer: setTimeout(() => {
setTimeout(() => { if (this.fileListResolvers.delete(reqId)) resolve([]);
const idx = this.fileListResolvers.indexOf(resolve); }, 5_000) });
if (idx !== -1) { this.ws!.send(JSON.stringify({ type: "list_files", query, from, _reqId: reqId }));
this.fileListResolvers.splice(idx, 1);
resolve([]);
}
}, 5_000);
}); });
} }
@@ -485,15 +459,11 @@ export class BrokerClient {
async fileStatus(fileId: string): Promise<Array<{ peerName: string; accessedAt: string }>> { async fileStatus(fileId: string): Promise<Array<{ peerName: string; accessedAt: string }>> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return []; if (!this.ws || this.ws.readyState !== this.ws.OPEN) return [];
return new Promise((resolve) => { return new Promise((resolve) => {
this.fileStatusResolvers.push(resolve); const reqId = this.makeReqId();
this.ws!.send(JSON.stringify({ type: "file_status", fileId })); this.fileStatusResolvers.set(reqId, { resolve, timer: setTimeout(() => {
setTimeout(() => { if (this.fileStatusResolvers.delete(reqId)) resolve([]);
const idx = this.fileStatusResolvers.indexOf(resolve); }, 5_000) });
if (idx !== -1) { this.ws!.send(JSON.stringify({ type: "file_status", fileId, _reqId: reqId }));
this.fileStatusResolvers.splice(idx, 1);
resolve([]);
}
}, 5_000);
}); });
} }
@@ -547,13 +517,11 @@ export class BrokerClient {
async grantFileAccess(fileId: string, peerPubkey: string, sealedKey: string): Promise<boolean> { async grantFileAccess(fileId: string, peerPubkey: string, sealedKey: string): Promise<boolean> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return false; if (!this.ws || this.ws.readyState !== this.ws.OPEN) return false;
return new Promise((resolve) => { return new Promise((resolve) => {
const resolvers = this.grantFileAccessResolvers; const reqId = this.makeReqId();
resolvers.push(resolve); this.grantFileAccessResolvers.set(reqId, { resolve, timer: setTimeout(() => {
this.ws!.send(JSON.stringify({ type: "grant_file_access", fileId, peerPubkey, sealedKey })); if (this.grantFileAccessResolvers.delete(reqId)) resolve(false);
setTimeout(() => { }, 5_000) });
const idx = resolvers.indexOf(resolve); this.ws!.send(JSON.stringify({ type: "grant_file_access", fileId, peerPubkey, sealedKey, _reqId: reqId }));
if (idx !== -1) { resolvers.splice(idx, 1); resolve(false); }
}, 5_000);
}); });
} }
@@ -563,12 +531,11 @@ export class BrokerClient {
async vectorStore(collection: string, text: string, metadata?: Record<string, unknown>): Promise<string | null> { async vectorStore(collection: string, text: string, metadata?: Record<string, unknown>): Promise<string | null> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return null; if (!this.ws || this.ws.readyState !== this.ws.OPEN) return null;
return new Promise((resolve) => { return new Promise((resolve) => {
this.vectorStoredResolvers.push(resolve); const reqId = this.makeReqId();
this.ws!.send(JSON.stringify({ type: "vector_store", collection, text, metadata })); this.vectorStoredResolvers.set(reqId, { resolve, timer: setTimeout(() => {
setTimeout(() => { if (this.vectorStoredResolvers.delete(reqId)) resolve(null);
const idx = this.vectorStoredResolvers.indexOf(resolve); }, 5_000) });
if (idx !== -1) { this.vectorStoredResolvers.splice(idx, 1); resolve(null); } this.ws!.send(JSON.stringify({ type: "vector_store", collection, text, metadata, _reqId: reqId }));
}, 5_000);
}); });
} }
@@ -576,12 +543,11 @@ export class BrokerClient {
async vectorSearch(collection: string, query: string, limit?: number): Promise<Array<{ id: string; text: string; score: number; metadata?: Record<string, unknown> }>> { async vectorSearch(collection: string, query: string, limit?: number): Promise<Array<{ id: string; text: string; score: number; metadata?: Record<string, unknown> }>> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return []; if (!this.ws || this.ws.readyState !== this.ws.OPEN) return [];
return new Promise((resolve) => { return new Promise((resolve) => {
this.vectorResultsResolvers.push(resolve); const reqId = this.makeReqId();
this.ws!.send(JSON.stringify({ type: "vector_search", collection, query, limit })); this.vectorResultsResolvers.set(reqId, { resolve, timer: setTimeout(() => {
setTimeout(() => { if (this.vectorResultsResolvers.delete(reqId)) resolve([]);
const idx = this.vectorResultsResolvers.indexOf(resolve); }, 5_000) });
if (idx !== -1) { this.vectorResultsResolvers.splice(idx, 1); resolve([]); } this.ws!.send(JSON.stringify({ type: "vector_search", collection, query, limit, _reqId: reqId }));
}, 5_000);
}); });
} }
@@ -595,12 +561,11 @@ export class BrokerClient {
async listCollections(): Promise<string[]> { async listCollections(): Promise<string[]> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return []; if (!this.ws || this.ws.readyState !== this.ws.OPEN) return [];
return new Promise((resolve) => { return new Promise((resolve) => {
this.collectionListResolvers.push(resolve); const reqId = this.makeReqId();
this.ws!.send(JSON.stringify({ type: "list_collections" })); this.collectionListResolvers.set(reqId, { resolve, timer: setTimeout(() => {
setTimeout(() => { if (this.collectionListResolvers.delete(reqId)) resolve([]);
const idx = this.collectionListResolvers.indexOf(resolve); }, 5_000) });
if (idx !== -1) { this.collectionListResolvers.splice(idx, 1); resolve([]); } this.ws!.send(JSON.stringify({ type: "list_collections", _reqId: reqId }));
}, 5_000);
}); });
} }
@@ -610,12 +575,11 @@ export class BrokerClient {
async graphQuery(cypher: string): Promise<Array<Record<string, unknown>>> { async graphQuery(cypher: string): Promise<Array<Record<string, unknown>>> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return []; if (!this.ws || this.ws.readyState !== this.ws.OPEN) return [];
return new Promise((resolve) => { return new Promise((resolve) => {
this.graphResultResolvers.push(resolve); const reqId = this.makeReqId();
this.ws!.send(JSON.stringify({ type: "graph_query", cypher })); this.graphResultResolvers.set(reqId, { resolve, timer: setTimeout(() => {
setTimeout(() => { if (this.graphResultResolvers.delete(reqId)) resolve([]);
const idx = this.graphResultResolvers.indexOf(resolve); }, 5_000) });
if (idx !== -1) { this.graphResultResolvers.splice(idx, 1); resolve([]); } this.ws!.send(JSON.stringify({ type: "graph_query", cypher, _reqId: reqId }));
}, 5_000);
}); });
} }
@@ -623,12 +587,11 @@ export class BrokerClient {
async graphExecute(cypher: string): Promise<Array<Record<string, unknown>>> { async graphExecute(cypher: string): Promise<Array<Record<string, unknown>>> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return []; if (!this.ws || this.ws.readyState !== this.ws.OPEN) return [];
return new Promise((resolve) => { return new Promise((resolve) => {
this.graphResultResolvers.push(resolve); const reqId = this.makeReqId();
this.ws!.send(JSON.stringify({ type: "graph_execute", cypher })); this.graphResultResolvers.set(reqId, { resolve, timer: setTimeout(() => {
setTimeout(() => { if (this.graphResultResolvers.delete(reqId)) resolve([]);
const idx = this.graphResultResolvers.indexOf(resolve); }, 5_000) });
if (idx !== -1) { this.graphResultResolvers.splice(idx, 1); resolve([]); } this.ws!.send(JSON.stringify({ type: "graph_execute", cypher, _reqId: reqId }));
}, 5_000);
}); });
} }
@@ -644,12 +607,11 @@ export class BrokerClient {
async getContext(query: string): Promise<Array<{ peerName: string; summary: string; filesRead: string[]; keyFindings: string[]; tags: string[]; updatedAt: string }>> { async getContext(query: string): Promise<Array<{ peerName: string; summary: string; filesRead: string[]; keyFindings: string[]; tags: string[]; updatedAt: string }>> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return []; if (!this.ws || this.ws.readyState !== this.ws.OPEN) return [];
return new Promise((resolve) => { return new Promise((resolve) => {
this.contextResultsResolvers.push(resolve); const reqId = this.makeReqId();
this.ws!.send(JSON.stringify({ type: "get_context", query })); this.contextResultsResolvers.set(reqId, { resolve, timer: setTimeout(() => {
setTimeout(() => { if (this.contextResultsResolvers.delete(reqId)) resolve([]);
const idx = this.contextResultsResolvers.indexOf(resolve); }, 5_000) });
if (idx !== -1) { this.contextResultsResolvers.splice(idx, 1); resolve([]); } this.ws!.send(JSON.stringify({ type: "get_context", query, _reqId: reqId }));
}, 5_000);
}); });
} }
@@ -657,12 +619,11 @@ export class BrokerClient {
async listContexts(): Promise<Array<{ peerName: string; summary: string; tags: string[]; updatedAt: string }>> { async listContexts(): Promise<Array<{ peerName: string; summary: string; tags: string[]; updatedAt: string }>> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return []; if (!this.ws || this.ws.readyState !== this.ws.OPEN) return [];
return new Promise((resolve) => { return new Promise((resolve) => {
this.contextListResolvers.push(resolve); const reqId = this.makeReqId();
this.ws!.send(JSON.stringify({ type: "list_contexts" })); this.contextListResolvers.set(reqId, { resolve, timer: setTimeout(() => {
setTimeout(() => { if (this.contextListResolvers.delete(reqId)) resolve([]);
const idx = this.contextListResolvers.indexOf(resolve); }, 5_000) });
if (idx !== -1) { this.contextListResolvers.splice(idx, 1); resolve([]); } this.ws!.send(JSON.stringify({ type: "list_contexts", _reqId: reqId }));
}, 5_000);
}); });
} }
@@ -672,12 +633,11 @@ export class BrokerClient {
async createTask(title: string, assignee?: string, priority?: string, tags?: string[]): Promise<string | null> { async createTask(title: string, assignee?: string, priority?: string, tags?: string[]): Promise<string | null> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return null; if (!this.ws || this.ws.readyState !== this.ws.OPEN) return null;
return new Promise((resolve) => { return new Promise((resolve) => {
this.taskCreatedResolvers.push(resolve); const reqId = this.makeReqId();
this.ws!.send(JSON.stringify({ type: "create_task", title, assignee, priority, tags })); this.taskCreatedResolvers.set(reqId, { resolve, timer: setTimeout(() => {
setTimeout(() => { if (this.taskCreatedResolvers.delete(reqId)) resolve(null);
const idx = this.taskCreatedResolvers.indexOf(resolve); }, 5_000) });
if (idx !== -1) { this.taskCreatedResolvers.splice(idx, 1); resolve(null); } this.ws!.send(JSON.stringify({ type: "create_task", title, assignee, priority, tags, _reqId: reqId }));
}, 5_000);
}); });
} }
@@ -697,12 +657,11 @@ export class BrokerClient {
async listTasks(status?: string, assignee?: string): Promise<Array<{ id: string; title: string; assignee: string; status: string; priority: string; createdBy: string }>> { async listTasks(status?: string, assignee?: string): Promise<Array<{ id: string; title: string; assignee: string; status: string; priority: string; createdBy: string }>> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return []; if (!this.ws || this.ws.readyState !== this.ws.OPEN) return [];
return new Promise((resolve) => { return new Promise((resolve) => {
this.taskListResolvers.push(resolve); const reqId = this.makeReqId();
this.ws!.send(JSON.stringify({ type: "list_tasks", status, assignee })); this.taskListResolvers.set(reqId, { resolve, timer: setTimeout(() => {
setTimeout(() => { if (this.taskListResolvers.delete(reqId)) resolve([]);
const idx = this.taskListResolvers.indexOf(resolve); }, 5_000) });
if (idx !== -1) { this.taskListResolvers.splice(idx, 1); resolve([]); } this.ws!.send(JSON.stringify({ type: "list_tasks", status, assignee, _reqId: reqId }));
}, 5_000);
}); });
} }
@@ -712,12 +671,11 @@ export class BrokerClient {
async meshQuery(sql: string): Promise<{ columns: string[]; rows: Array<Record<string, unknown>>; rowCount: number } | null> { async meshQuery(sql: string): Promise<{ columns: string[]; rows: Array<Record<string, unknown>>; rowCount: number } | null> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return null; if (!this.ws || this.ws.readyState !== this.ws.OPEN) return null;
return new Promise((resolve) => { return new Promise((resolve) => {
this.meshQueryResolvers.push(resolve); const reqId = this.makeReqId();
this.ws!.send(JSON.stringify({ type: "mesh_query", sql })); this.meshQueryResolvers.set(reqId, { resolve, timer: setTimeout(() => {
setTimeout(() => { if (this.meshQueryResolvers.delete(reqId)) resolve(null);
const idx = this.meshQueryResolvers.indexOf(resolve); }, 5_000) });
if (idx !== -1) { this.meshQueryResolvers.splice(idx, 1); resolve(null); } this.ws!.send(JSON.stringify({ type: "mesh_query", sql, _reqId: reqId }));
}, 5_000);
}); });
} }
@@ -731,12 +689,11 @@ export class BrokerClient {
async meshSchema(): Promise<Array<{ name: string; columns: Array<{ name: string; type: string; nullable: boolean }> }>> { async meshSchema(): Promise<Array<{ name: string; columns: Array<{ name: string; type: string; nullable: boolean }> }>> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return []; if (!this.ws || this.ws.readyState !== this.ws.OPEN) return [];
return new Promise((resolve) => { return new Promise((resolve) => {
this.meshSchemaResolvers.push(resolve); const reqId = this.makeReqId();
this.ws!.send(JSON.stringify({ type: "mesh_schema" })); this.meshSchemaResolvers.set(reqId, { resolve, timer: setTimeout(() => {
setTimeout(() => { if (this.meshSchemaResolvers.delete(reqId)) resolve([]);
const idx = this.meshSchemaResolvers.indexOf(resolve); }, 5_000) });
if (idx !== -1) { this.meshSchemaResolvers.splice(idx, 1); resolve([]); } this.ws!.send(JSON.stringify({ type: "mesh_schema", _reqId: reqId }));
}, 5_000);
}); });
} }
@@ -746,12 +703,11 @@ export class BrokerClient {
async createStream(name: string): Promise<string | null> { async createStream(name: string): Promise<string | null> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return null; if (!this.ws || this.ws.readyState !== this.ws.OPEN) return null;
return new Promise((resolve) => { return new Promise((resolve) => {
this.streamCreatedResolvers.push(resolve); const reqId = this.makeReqId();
this.ws!.send(JSON.stringify({ type: "create_stream", name })); this.streamCreatedResolvers.set(reqId, { resolve, timer: setTimeout(() => {
setTimeout(() => { if (this.streamCreatedResolvers.delete(reqId)) resolve(null);
const idx = this.streamCreatedResolvers.indexOf(resolve); }, 5_000) });
if (idx !== -1) { this.streamCreatedResolvers.splice(idx, 1); resolve(null); } this.ws!.send(JSON.stringify({ type: "create_stream", name, _reqId: reqId }));
}, 5_000);
}); });
} }
@@ -777,12 +733,11 @@ export class BrokerClient {
async listStreams(): Promise<Array<{ id: string; name: string; createdBy: string; subscriberCount: number }>> { async listStreams(): Promise<Array<{ id: string; name: string; createdBy: string; subscriberCount: number }>> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return []; if (!this.ws || this.ws.readyState !== this.ws.OPEN) return [];
return new Promise((resolve) => { return new Promise((resolve) => {
this.streamListResolvers.push(resolve); const reqId = this.makeReqId();
this.ws!.send(JSON.stringify({ type: "list_streams" })); this.streamListResolvers.set(reqId, { resolve, timer: setTimeout(() => {
setTimeout(() => { if (this.streamListResolvers.delete(reqId)) resolve([]);
const idx = this.streamListResolvers.indexOf(resolve); }, 5_000) });
if (idx !== -1) { this.streamListResolvers.splice(idx, 1); resolve([]); } this.ws!.send(JSON.stringify({ type: "list_streams", _reqId: reqId }));
}, 5_000);
}); });
} }
@@ -799,17 +754,16 @@ export class BrokerClient {
} }
// --- Mesh info --- // --- Mesh info ---
private meshInfoResolvers: Array<(result: Record<string, unknown> | null) => void> = []; private meshInfoResolvers = new Map<string, { resolve: (result: Record<string, unknown> | null) => void; timer: NodeJS.Timeout }>();
async meshInfo(): Promise<Record<string, unknown> | null> { async meshInfo(): Promise<Record<string, unknown> | null> {
if (!this.ws || this.ws.readyState !== this.ws.OPEN) return null; if (!this.ws || this.ws.readyState !== this.ws.OPEN) return null;
return new Promise((resolve) => { return new Promise((resolve) => {
this.meshInfoResolvers.push(resolve); const reqId = this.makeReqId();
this.ws!.send(JSON.stringify({ type: "mesh_info" })); this.meshInfoResolvers.set(reqId, { resolve, timer: setTimeout(() => {
setTimeout(() => { if (this.meshInfoResolvers.delete(reqId)) resolve(null);
const idx = this.meshInfoResolvers.indexOf(resolve); }, 5_000) });
if (idx !== -1) { this.meshInfoResolvers.splice(idx, 1); resolve(null); } this.ws!.send(JSON.stringify({ type: "mesh_info", _reqId: reqId }));
}, 5_000);
}); });
} }
@@ -829,7 +783,33 @@ export class BrokerClient {
// --- Internals --- // --- Internals ---
private resolveFromMap<T>(
map: Map<string, { resolve: (v: T) => void; timer: NodeJS.Timeout }>,
reqId: string | undefined,
value: T,
): boolean {
let entry = reqId ? map.get(reqId) : undefined;
if (!entry) {
// Fallback: oldest pending (FIFO, for brokers that don't echo _reqId)
const first = map.entries().next().value as [string, { resolve: (v: T) => void; timer: NodeJS.Timeout }] | undefined;
if (first) {
entry = first[1];
map.delete(first[0]);
}
} else {
map.delete(reqId!);
}
if (entry) {
clearTimeout(entry.timer);
entry.resolve(value);
return true;
}
return false;
}
private handleServerMessage(msg: Record<string, unknown>): void { private handleServerMessage(msg: Record<string, unknown>): void {
const msgReqId = msg._reqId as string | undefined;
if (msg.type === "ack") { if (msg.type === "ack") {
const pending = this.pendingSends.get(String(msg.id ?? "")); const pending = this.pendingSends.get(String(msg.id ?? ""));
if (pending) { if (pending) {
@@ -843,8 +823,7 @@ export class BrokerClient {
} }
if (msg.type === "peers_list") { if (msg.type === "peers_list") {
const peers = (msg.peers as PeerInfo[]) ?? []; const peers = (msg.peers as PeerInfo[]) ?? [];
const resolver = this.listPeersResolvers.shift(); this.resolveFromMap(this.listPeersResolvers, msgReqId, peers);
if (resolver) resolver(peers);
return; return;
} }
if (msg.type === "push") { if (msg.type === "push") {
@@ -922,25 +901,21 @@ export class BrokerClient {
// both, it would be consumed here by the next pending get_state resolver, // both, it would be consumed here by the next pending get_state resolver,
// returning the wrong value (cross-contamination). The broker's set_state // returning the wrong value (cross-contamination). The broker's set_state
// handler was fixed to omit state_result; only get_state sends it. // handler was fixed to omit state_result; only get_state sends it.
const resolver = this.stateResolvers.shift();
if (resolver) {
if (msg.key) { if (msg.key) {
resolver({ this.resolveFromMap(this.stateResolvers, msgReqId, {
key: String(msg.key), key: String(msg.key),
value: msg.value, value: msg.value,
updatedBy: String(msg.updatedBy ?? ""), updatedBy: String(msg.updatedBy ?? ""),
updatedAt: String(msg.updatedAt ?? ""), updatedAt: String(msg.updatedAt ?? ""),
}); });
} else { } else {
resolver(null); this.resolveFromMap(this.stateResolvers, msgReqId, null);
}
} }
return; return;
} }
if (msg.type === "state_list") { if (msg.type === "state_list") {
const entries = (msg.entries as Array<{ key: string; value: unknown; updatedBy: string; updatedAt: string }>) ?? []; const entries = (msg.entries as Array<{ key: string; value: unknown; updatedBy: string; updatedAt: string }>) ?? [];
const resolver = this.stateListResolvers.shift(); this.resolveFromMap(this.stateListResolvers, msgReqId, entries);
if (resolver) resolver(entries);
return; return;
} }
if (msg.type === "state_change") { if (msg.type === "state_change") {
@@ -955,131 +930,108 @@ export class BrokerClient {
return; return;
} }
if (msg.type === "memory_stored") { if (msg.type === "memory_stored") {
const resolver = this.memoryStoreResolvers.shift(); this.resolveFromMap(this.memoryStoreResolvers, msgReqId, msg.id ? String(msg.id) : null);
if (resolver) resolver(msg.id ? String(msg.id) : null);
return; return;
} }
if (msg.type === "memory_results") { if (msg.type === "memory_results") {
const memories = (msg.memories as Array<{ id: string; content: string; tags: string[]; rememberedBy: string; rememberedAt: string }>) ?? []; const memories = (msg.memories as Array<{ id: string; content: string; tags: string[]; rememberedBy: string; rememberedAt: string }>) ?? [];
const resolver = this.memoryRecallResolvers.shift(); this.resolveFromMap(this.memoryRecallResolvers, msgReqId, memories);
if (resolver) resolver(memories);
return; return;
} }
if (msg.type === "message_status_result") { if (msg.type === "message_status_result") {
const resolver = this.messageStatusResolvers.shift(); this.resolveFromMap(this.messageStatusResolvers, msgReqId, msg as any);
if (resolver) resolver(msg as any);
return; return;
} }
if (msg.type === "file_url") { if (msg.type === "file_url") {
const resolver = this.fileUrlResolvers.shift();
if (resolver) {
if (msg.url) { if (msg.url) {
resolver({ this.resolveFromMap(this.fileUrlResolvers, msgReqId, {
url: String(msg.url), url: String(msg.url),
name: String(msg.name ?? ""), name: String(msg.name ?? ""),
encrypted: msg.encrypted ? true : undefined, encrypted: msg.encrypted ? true : undefined,
sealedKey: msg.sealedKey ? String(msg.sealedKey) : undefined, sealedKey: msg.sealedKey ? String(msg.sealedKey) : undefined,
}); });
} else { } else {
resolver(null); this.resolveFromMap(this.fileUrlResolvers, msgReqId, null);
}
} }
return; return;
} }
if (msg.type === "file_list") { if (msg.type === "file_list") {
const files = (msg.files as Array<{ id: string; name: string; size: number; tags: string[]; uploadedBy: string; uploadedAt: string; persistent: boolean }>) ?? []; const files = (msg.files as Array<{ id: string; name: string; size: number; tags: string[]; uploadedBy: string; uploadedAt: string; persistent: boolean }>) ?? [];
const resolver = this.fileListResolvers.shift(); this.resolveFromMap(this.fileListResolvers, msgReqId, files);
if (resolver) resolver(files);
return; return;
} }
if (msg.type === "file_status_result") { if (msg.type === "file_status_result") {
const accesses = (msg.accesses as Array<{ peerName: string; accessedAt: string }>) ?? []; const accesses = (msg.accesses as Array<{ peerName: string; accessedAt: string }>) ?? [];
const resolver = this.fileStatusResolvers.shift(); this.resolveFromMap(this.fileStatusResolvers, msgReqId, accesses);
if (resolver) resolver(accesses);
return; return;
} }
if (msg.type === "grant_file_access_ok") { if (msg.type === "grant_file_access_ok") {
const resolver = this.grantFileAccessResolvers.shift(); this.resolveFromMap(this.grantFileAccessResolvers, msgReqId, true);
if (resolver) resolver(true);
return; return;
} }
if (msg.type === "vector_stored") { if (msg.type === "vector_stored") {
const resolver = this.vectorStoredResolvers.shift(); this.resolveFromMap(this.vectorStoredResolvers, msgReqId, msg.id ? String(msg.id) : null);
if (resolver) resolver(msg.id ? String(msg.id) : null);
return; return;
} }
if (msg.type === "vector_results") { if (msg.type === "vector_results") {
const results = (msg.results as Array<{ id: string; text: string; score: number; metadata?: Record<string, unknown> }>) ?? []; const results = (msg.results as Array<{ id: string; text: string; score: number; metadata?: Record<string, unknown> }>) ?? [];
const resolver = this.vectorResultsResolvers.shift(); this.resolveFromMap(this.vectorResultsResolvers, msgReqId, results);
if (resolver) resolver(results);
return; return;
} }
if (msg.type === "collection_list") { if (msg.type === "collection_list") {
const collections = (msg.collections as string[]) ?? []; const collections = (msg.collections as string[]) ?? [];
const resolver = this.collectionListResolvers.shift(); this.resolveFromMap(this.collectionListResolvers, msgReqId, collections);
if (resolver) resolver(collections);
return; return;
} }
if (msg.type === "graph_result") { if (msg.type === "graph_result") {
// Broker sends { type: "graph_result", records: [...] } // Broker sends { type: "graph_result", records: [...] }
const rows = (msg.records as Array<Record<string, unknown>>) ?? []; const rows = (msg.records as Array<Record<string, unknown>>) ?? [];
const resolver = this.graphResultResolvers.shift(); this.resolveFromMap(this.graphResultResolvers, msgReqId, rows);
if (resolver) resolver(rows);
return; return;
} }
if (msg.type === "context_list") { if (msg.type === "context_list") {
const contexts = (msg.contexts as Array<{ peerName: string; summary: string; tags: string[]; updatedAt: string }>) ?? []; const contexts = (msg.contexts as Array<{ peerName: string; summary: string; tags: string[]; updatedAt: string }>) ?? [];
const resolver = this.contextListResolvers.shift(); this.resolveFromMap(this.contextListResolvers, msgReqId, contexts);
if (resolver) resolver(contexts);
return; return;
} }
if (msg.type === "context_results") { if (msg.type === "context_results") {
const contexts = (msg.contexts as Array<{ peerName: string; summary: string; filesRead: string[]; keyFindings: string[]; tags: string[]; updatedAt: string }>) ?? []; const contexts = (msg.contexts as Array<{ peerName: string; summary: string; filesRead: string[]; keyFindings: string[]; tags: string[]; updatedAt: string }>) ?? [];
const resolver = this.contextResultsResolvers.shift(); this.resolveFromMap(this.contextResultsResolvers, msgReqId, contexts);
if (resolver) resolver(contexts);
return; return;
} }
if (msg.type === "task_created") { if (msg.type === "task_created") {
const resolver = this.taskCreatedResolvers.shift(); this.resolveFromMap(this.taskCreatedResolvers, msgReqId, msg.id ? String(msg.id) : null);
if (resolver) resolver(msg.id ? String(msg.id) : null);
return; return;
} }
if (msg.type === "task_list") { if (msg.type === "task_list") {
const tasks = (msg.tasks as Array<{ id: string; title: string; assignee: string; status: string; priority: string; createdBy: string }>) ?? []; const tasks = (msg.tasks as Array<{ id: string; title: string; assignee: string; status: string; priority: string; createdBy: string }>) ?? [];
const resolver = this.taskListResolvers.shift(); this.resolveFromMap(this.taskListResolvers, msgReqId, tasks);
if (resolver) resolver(tasks);
return; return;
} }
if (msg.type === "mesh_query_result") { if (msg.type === "mesh_query_result") {
const resolver = this.meshQueryResolvers.shift();
if (resolver) {
if (msg.columns) { if (msg.columns) {
resolver({ this.resolveFromMap(this.meshQueryResolvers, msgReqId, {
columns: (msg.columns as string[]) ?? [], columns: (msg.columns as string[]) ?? [],
rows: (msg.rows as Array<Record<string, unknown>>) ?? [], rows: (msg.rows as Array<Record<string, unknown>>) ?? [],
rowCount: (msg.rowCount as number) ?? 0, rowCount: (msg.rowCount as number) ?? 0,
}); });
} else { } else {
resolver(null); this.resolveFromMap(this.meshQueryResolvers, msgReqId, null);
}
} }
return; return;
} }
if (msg.type === "mesh_schema_result") { if (msg.type === "mesh_schema_result") {
const tables = (msg.tables as Array<{ name: string; columns: Array<{ name: string; type: string; nullable: boolean }> }>) ?? []; const tables = (msg.tables as Array<{ name: string; columns: Array<{ name: string; type: string; nullable: boolean }> }>) ?? [];
const resolver = this.meshSchemaResolvers.shift(); this.resolveFromMap(this.meshSchemaResolvers, msgReqId, tables);
if (resolver) resolver(tables);
return; return;
} }
if (msg.type === "stream_created") { if (msg.type === "stream_created") {
const resolver = this.streamCreatedResolvers.shift(); this.resolveFromMap(this.streamCreatedResolvers, msgReqId, msg.id ? String(msg.id) : null);
if (resolver) resolver(msg.id ? String(msg.id) : null);
return; return;
} }
if (msg.type === "stream_list") { if (msg.type === "stream_list") {
const streams = (msg.streams as Array<{ id: string; name: string; createdBy: string; subscriberCount: number }>) ?? []; const streams = (msg.streams as Array<{ id: string; name: string; createdBy: string; subscriberCount: number }>) ?? [];
const resolver = this.streamListResolvers.shift(); this.resolveFromMap(this.streamListResolvers, msgReqId, streams);
if (resolver) resolver(streams);
return; return;
} }
if (msg.type === "stream_data") { if (msg.type === "stream_data") {
@@ -1094,8 +1046,7 @@ export class BrokerClient {
return; return;
} }
if (msg.type === "mesh_info_result") { if (msg.type === "mesh_info_result") {
const resolver = this.meshInfoResolvers.shift(); this.resolveFromMap(this.meshInfoResolvers, msgReqId, msg as Record<string, unknown>);
if (resolver) resolver(msg as Record<string, unknown>);
return; return;
} }
if (msg.type === "error") { if (msg.type === "error") {
@@ -1116,37 +1067,40 @@ export class BrokerClient {
if (!handledByPendingSend) { if (!handledByPendingSend) {
// Best-effort: unblock the first waiting resolver so callers don't // Best-effort: unblock the first waiting resolver so callers don't
// hang for 5s. We don't know which tool triggered the error, so we // hang for 5s. We don't know which tool triggered the error, so we
// pop the first non-empty resolver queue in priority order. // pop the first non-empty resolver map in priority order.
if (this.stateResolvers.length > 0) { const allMaps: Array<[Map<string, { resolve: (v: any) => void; timer: NodeJS.Timeout }>, unknown]> = [
this.stateResolvers.shift()!(null); [this.stateResolvers, null],
} else if (this.stateListResolvers.length > 0) { [this.stateListResolvers, []],
this.stateListResolvers.shift()!([]); [this.memoryStoreResolvers, null],
} else if (this.memoryStoreResolvers.length > 0) { [this.memoryRecallResolvers, []],
this.memoryStoreResolvers.shift()!(null); [this.fileUrlResolvers, null],
} else if (this.memoryRecallResolvers.length > 0) { [this.fileListResolvers, []],
this.memoryRecallResolvers.shift()!([]); [this.fileStatusResolvers, []],
} else if (this.fileUrlResolvers.length > 0) { [this.graphResultResolvers, []],
this.fileUrlResolvers.shift()!(null); [this.vectorStoredResolvers, null],
} else if (this.fileListResolvers.length > 0) { [this.vectorResultsResolvers, []],
this.fileListResolvers.shift()!([]); [this.taskListResolvers, []],
} else if (this.fileStatusResolvers.length > 0) { [this.meshQueryResolvers, null],
this.fileStatusResolvers.shift()!([]); [this.contextResultsResolvers, []],
} else if (this.graphResultResolvers.length > 0) { [this.contextListResolvers, []],
this.graphResultResolvers.shift()!([]); [this.streamListResolvers, []],
} else if (this.vectorStoredResolvers.length > 0) { [this.messageStatusResolvers, null],
this.vectorStoredResolvers.shift()!(null); [this.grantFileAccessResolvers, false],
} else if (this.vectorResultsResolvers.length > 0) { [this.collectionListResolvers, []],
this.vectorResultsResolvers.shift()!([]); [this.meshSchemaResolvers, []],
} else if (this.taskListResolvers.length > 0) { [this.taskCreatedResolvers, null],
this.taskListResolvers.shift()!([]); [this.streamCreatedResolvers, null],
} else if (this.meshQueryResolvers.length > 0) { [this.listPeersResolvers, []],
this.meshQueryResolvers.shift()!(null); [this.meshInfoResolvers, null],
} else if (this.contextResultsResolvers.length > 0) { ];
this.contextResultsResolvers.shift()!([]); for (const [map, defaultVal] of allMaps) {
} else if (this.contextListResolvers.length > 0) { const first = (map as Map<string, any>).entries().next().value as [string, { resolve: (v: unknown) => void; timer: NodeJS.Timeout }] | undefined;
this.contextListResolvers.shift()!([]); if (first) {
} else if (this.streamListResolvers.length > 0) { (map as Map<string, any>).delete(first[0]);
this.streamListResolvers.shift()!([]); clearTimeout(first[1].timer);
first[1].resolve(defaultVal);
break; // only pop one
}
} }
} }
} }