Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 7 additions & 41 deletions apps/memos-local-plugin/bridge.cts
Original file line number Diff line number Diff line change
Expand Up @@ -197,6 +197,9 @@ async function main(): Promise<void> {
const { startStdioServer, waitForShutdown } = (await importEsm(
runtimeModule("bridge/stdio.ts", "dist/bridge/stdio.js")
)) as typeof import("./bridge/stdio.js");
const { createStdioHostLlmBridge } = (await importEsm(
runtimeModule("bridge/host-llm.ts", "dist/bridge/host-llm.js")
)) as typeof import("./bridge/host-llm.js");
const { memoryBuffer, rootLogger } = (await importEsm(
runtimeModule("core/logger/index.ts", "dist/core/logger/index.js")
)) as typeof import("./core/logger/index.js");
Expand Down Expand Up @@ -237,47 +240,10 @@ async function main(): Promise<void> {
// independent `currentBridge` slots. Registering inside bootstrap
// forces both ends to share the same module instance.
let stdio: import("./bridge/stdio.js").StdioServerHandle | null = null;
const lazyHostLlmBridge: import("./core/llm/host-bridge.js").HostLlmBridge =
{
id: `stdio.host.${args.agent}.v1`,
async complete(input) {
if (!stdio) {
throw new Error(
"host LLM bridge invoked before stdio server was ready",
);
}
const result = (await stdio.serverRequest(
"host.llm.complete",
{
messages: input.messages,
model: input.model,
temperature: input.temperature,
maxTokens: input.maxTokens,
timeoutMs: input.timeoutMs,
},
{ timeoutMs: (input.timeoutMs ?? 60_000) + 5_000 },
)) as {
text?: string;
model?: string;
usage?: {
promptTokens?: number;
completionTokens?: number;
totalTokens?: number;
};
durationMs?: number;
};
return {
text: typeof result?.text === "string" ? result.text : "",
model:
typeof result?.model === "string"
? result.model
: input.model ?? "",
usage: result?.usage,
durationMs:
typeof result?.durationMs === "number" ? result.durationMs : 0,
};
},
};
const lazyHostLlmBridge = createStdioHostLlmBridge(
`stdio.host.${args.agent}.v1`,
() => stdio,
);

const { Telemetry } = (await importEsm(
runtimeModule("core/telemetry/index.ts", "dist/core/telemetry/index.js")
Expand Down
46 changes: 5 additions & 41 deletions apps/memos-local-plugin/bridge.mts
Original file line number Diff line number Diff line change
Expand Up @@ -180,6 +180,7 @@ async function main(): Promise<void> {
// from source via `tsx bridge.mts`.
const { bootstrapMemoryCoreFull } = await import("./core/pipeline/index.js");
const { startStdioServer, waitForShutdown } = await import("./bridge/stdio.js");
const { createStdioHostLlmBridge } = await import("./bridge/host-llm.js");
const { memoryBuffer, rootLogger } = await import("./core/logger/index.js");
const { startHttpServer } = await import("./server/http.js");

Expand All @@ -206,47 +207,10 @@ async function main(): Promise<void> {
// independent `currentBridge` slots. Registering inside bootstrap
// forces both ends to share the same module instance.
let stdio: import("./bridge/stdio.js").StdioServerHandle | null = null;
const lazyHostLlmBridge: import("./core/llm/host-bridge.js").HostLlmBridge =
{
id: `stdio.host.${args.agent}.v1`,
async complete(input) {
if (!stdio) {
throw new Error(
"host LLM bridge invoked before stdio server was ready",
);
}
const result = (await stdio.serverRequest(
"host.llm.complete",
{
messages: input.messages,
model: input.model,
temperature: input.temperature,
maxTokens: input.maxTokens,
timeoutMs: input.timeoutMs,
},
{ timeoutMs: (input.timeoutMs ?? 60_000) + 5_000 },
)) as {
text?: string;
model?: string;
usage?: {
promptTokens?: number;
completionTokens?: number;
totalTokens?: number;
};
durationMs?: number;
};
return {
text: typeof result?.text === "string" ? result.text : "",
model:
typeof result?.model === "string"
? result.model
: input.model ?? "",
usage: result?.usage,
durationMs:
typeof result?.durationMs === "number" ? result.durationMs : 0,
};
},
};
const lazyHostLlmBridge = createStdioHostLlmBridge(
`stdio.host.${args.agent}.v1`,
() => stdio,
);

const { Telemetry } = await import("./core/telemetry/index.js");

Expand Down
54 changes: 54 additions & 0 deletions apps/memos-local-plugin/bridge/host-llm.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,54 @@
import type {
HostLlmBridge,
HostLlmCompletion,
} from "../core/llm/host-bridge.js";
import type { StdioServerHandle } from "./stdio.js";

type ServerRequestTransport = Pick<StdioServerHandle, "serverRequest">;

export function createStdioHostLlmBridge(
id: string,
getTransport: () => ServerRequestTransport | null,
): HostLlmBridge {
return {
id,
async complete(input) {
const transport = getTransport();
if (!transport) {
throw new Error("host LLM bridge invoked before stdio server was ready");
}

const configuredTimeoutMs = input.timeoutMs ?? 60_000;
const remainingDeadlineMs =
input.deadlineAt === undefined
? configuredTimeoutMs
: input.deadlineAt - Date.now();
if (remainingDeadlineMs <= 0) {
throw new Error("host LLM request deadline has passed");
}
const timeoutMs = Math.min(configuredTimeoutMs, remainingDeadlineMs);

const result = await transport.serverRequest<Partial<HostLlmCompletion>>(
"host.llm.complete",
{
messages: input.messages,
model: input.model,
temperature: input.temperature,
maxTokens: input.maxTokens,
timeoutMs,
},
{ timeoutMs, signal: input.signal },
);
return {
text: typeof result?.text === "string" ? result.text : "",
model:
typeof result?.model === "string"
? result.model
: input.model ?? "",
usage: result?.usage,
durationMs:
typeof result?.durationMs === "number" ? result.durationMs : 0,
};
},
};
}
51 changes: 39 additions & 12 deletions apps/memos-local-plugin/bridge/stdio.ts
Original file line number Diff line number Diff line change
Expand Up @@ -64,7 +64,7 @@ export interface StdioServerHandle {
serverRequest<R = unknown>(
method: string,
params?: unknown,
options?: { timeoutMs?: number },
options?: { timeoutMs?: number; signal?: AbortSignal },
): Promise<R>;
}

Expand All @@ -91,9 +91,22 @@ export function startStdioServer(options: StdioServerOptions): StdioServerHandle
resolve: (value: unknown) => void;
reject: (err: unknown) => void;
timer: ReturnType<typeof setTimeout> | null;
signal?: AbortSignal;
abortHandler?: () => void;
}
>();

function takeServerPending(id: string) {
const entry = serverPending.get(id);
if (!entry) return undefined;
serverPending.delete(id);
if (entry.timer) clearTimeout(entry.timer);
if (entry.signal && entry.abortHandler) {
entry.signal.removeEventListener("abort", entry.abortHandler);
}
return entry;
}

const eventsUnsubscribe = options.core.subscribeEvents((e) => {
writeNotification(RPC_METHODS.EVENTS_NOTIFY, e);
});
Expand All @@ -114,10 +127,8 @@ export function startStdioServer(options: StdioServerOptions): StdioServerHandle
} catch {
/* ignore */
}
for (const [id, entry] of serverPending) {
if (entry.timer) clearTimeout(entry.timer);
entry.reject(err ?? new Error("stdio bridge closed"));
serverPending.delete(id);
for (const id of serverPending.keys()) {
takeServerPending(id)?.reject(err ?? new Error("stdio bridge closed"));
}
}

Expand Down Expand Up @@ -179,10 +190,8 @@ export function startStdioServer(options: StdioServerOptions): StdioServerHandle
(raw.result !== undefined || raw.error !== undefined)
) {
const id = raw.id as string;
const pending = serverPending.get(id);
const pending = takeServerPending(id);
if (pending) {
serverPending.delete(id);
if (pending.timer) clearTimeout(pending.timer);
if (raw.error != null) {
pending.reject(raw.error);
} else {
Expand Down Expand Up @@ -260,8 +269,12 @@ export function startStdioServer(options: StdioServerOptions): StdioServerHandle
function serverRequest<R = unknown>(
method: string,
params?: unknown,
options?: { timeoutMs?: number },
options?: { timeoutMs?: number; signal?: AbortSignal },
): Promise<R> {
const signal = options?.signal;
if (signal?.aborted) {
return Promise.reject(signal.reason ?? new Error(`serverRequest ${method} aborted`));
}
const id = `srv-${++serverRequestSeq}`;
const timeoutMs = options?.timeoutMs ?? 60_000;
return new Promise<R>((resolve, reject) => {
Expand All @@ -270,15 +283,29 @@ export function startStdioServer(options: StdioServerOptions): StdioServerHandle
return;
}
const timer = setTimeout(() => {
if (serverPending.delete(id)) {
reject(new Error(`serverRequest ${method} timed out after ${timeoutMs}ms`));
}
takeServerPending(id)?.reject(
new Error(`serverRequest ${method} timed out after ${timeoutMs}ms`),
);
}, timeoutMs);
const abortHandler = signal
? () => {
takeServerPending(id)?.reject(
signal.reason ?? new Error(`serverRequest ${method} aborted`),
);
}
: undefined;
serverPending.set(id, {
resolve: resolve as (value: unknown) => void,
reject,
timer,
signal,
abortHandler,
});
if (signal && abortHandler) {
signal.addEventListener("abort", abortHandler, { once: true });
}
if (signal?.aborted) abortHandler?.();
if (!serverPending.has(id)) return;
writeLine({ jsonrpc: "2.0", id, method, params });
});
}
Expand Down
5 changes: 5 additions & 0 deletions apps/memos-local-plugin/core/llm/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -105,6 +105,11 @@ we call the host once as a best-effort fallback. The resulting completion's
`servedBy` is `"host_fallback"` so downstream dashboards can distinguish
host wins from primary wins.

Fallback is skipped once the caller's abort signal fires or its absolute
deadline passes. When fallback is still allowed, both cancellation and the
remaining deadline propagate to the host bridge so the host request cannot
outlive the caller's budget.

Registering from an adapter:

```ts
Expand Down
16 changes: 12 additions & 4 deletions apps/memos-local-plugin/core/llm/client.ts
Original file line number Diff line number Diff line change
Expand Up @@ -182,11 +182,19 @@ export function createLlmClientWithProvider(
);
}

function canUseHostFallback(): boolean {
function callerBudgetIsGone(opts: LlmCallOptions | undefined): boolean {
return (
opts?.signal?.aborted === true ||
(opts?.deadlineAt !== undefined && Date.now() >= opts.deadlineAt)
);
}

function canUseHostFallback(opts?: LlmCallOptions): boolean {
return (
config.fallbackToHost === true &&
provider.name !== "host" &&
getHostLlmBridge() !== null
getHostLlmBridge() !== null &&
!callerBudgetIsGone(opts)
);
}

Expand Down Expand Up @@ -311,7 +319,7 @@ export function createLlmClientWithProvider(
// Overview can surface that suppression is happening.
if (breakerIsOpen()) {
maybeEmitCircuitOpenStatus(opts, op);
if (canUseHostFallback()) {
if (canUseHostFallback(opts)) {
return callHostFallback(makeBreakerOpenError(), messages, input, opts, op, {
keepBreakerOpen: true,
notifyError: false,
Expand Down Expand Up @@ -347,7 +355,7 @@ export function createLlmClientWithProvider(
});
return { completion };
} catch (err) {
if (shouldFallback(err, config, provider.name)) {
if (!callerBudgetIsGone(opts) && shouldFallback(err, config, provider.name)) {
const primaryTerminal = breakerIsTerminal(err);
if (primaryTerminal) breakerTrip(err);
try {
Expand Down
2 changes: 2 additions & 0 deletions apps/memos-local-plugin/core/llm/host-bridge.ts
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,8 @@ export interface HostLlmCompleteInput {
temperature?: number;
maxTokens?: number;
timeoutMs?: number;
/** Absolute caller deadline; adapters must not outlive it. */
deadlineAt?: number;
signal?: AbortSignal;
}

Expand Down
1 change: 1 addition & 0 deletions apps/memos-local-plugin/core/llm/providers/host.ts
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@ export class HostLlmProvider implements LlmProvider {
temperature: opts.temperature,
maxTokens: opts.maxTokens,
timeoutMs: ctx.config.timeoutMs,
deadlineAt: ctx.deadlineAt,
signal: ctx.signal,
});
return {
Expand Down
Loading
Loading