Compare commits
10 Commits
e75d2ba296
...
fix/mcp-br
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e85250fedf | ||
| cf85e36ede | |||
|
|
34e00af731 | ||
| 2f9e7be762 | |||
|
|
a432e9e19f | ||
| 3b96b0399e | |||
|
|
5225b54901 | ||
|
|
27480181bf | ||
| 86942b1e91 | |||
|
|
2729580974 |
@@ -11,6 +11,14 @@ export interface McpBridgeOptions {
|
|||||||
stderr: NodeJS.WritableStream;
|
stderr: NodeJS.WritableStream;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Per-request socket-inactivity timeout for the bridge. A stalled upstream must
|
||||||
|
* become a JSON-RPC error, not silence: the client cannot distinguish "still
|
||||||
|
* working" from "wedged", and its own idle limit may be half an hour away.
|
||||||
|
* Raise it for a project with genuinely long tool calls.
|
||||||
|
*/
|
||||||
|
export const BRIDGE_TIMEOUT_MS = Number(process.env['MCPCTL_MCP_TIMEOUT_MS']) || 30_000;
|
||||||
|
|
||||||
export function postJsonRpc(
|
export function postJsonRpc(
|
||||||
url: string,
|
url: string,
|
||||||
body: string,
|
body: string,
|
||||||
@@ -37,7 +45,7 @@ export function postJsonRpc(
|
|||||||
path: parsed.pathname,
|
path: parsed.pathname,
|
||||||
method: 'POST',
|
method: 'POST',
|
||||||
headers,
|
headers,
|
||||||
timeout: 30_000,
|
timeout: BRIDGE_TIMEOUT_MS,
|
||||||
},
|
},
|
||||||
(res) => {
|
(res) => {
|
||||||
const chunks: Buffer[] = [];
|
const chunks: Buffer[] = [];
|
||||||
@@ -54,7 +62,7 @@ export function postJsonRpc(
|
|||||||
req.on('error', reject);
|
req.on('error', reject);
|
||||||
req.on('timeout', () => {
|
req.on('timeout', () => {
|
||||||
req.destroy();
|
req.destroy();
|
||||||
reject(new Error('Request timed out'));
|
reject(new Error(`Request timed out after ${BRIDGE_TIMEOUT_MS}ms (set MCPCTL_MCP_TIMEOUT_MS to change)`));
|
||||||
});
|
});
|
||||||
req.write(body);
|
req.write(body);
|
||||||
req.end();
|
req.end();
|
||||||
@@ -128,19 +136,15 @@ export async function runMcpBridge(opts: McpBridgeOptions): Promise<void> {
|
|||||||
|
|
||||||
const rl = createInterface({ input: stdin, crlfDelay: Infinity });
|
const rl = createInterface({ input: stdin, crlfDelay: Infinity });
|
||||||
|
|
||||||
for await (const line of rl) {
|
/**
|
||||||
const trimmed = line.trim();
|
* In-flight requests. Dispatch is CONCURRENT after the session is
|
||||||
if (!trimmed) continue;
|
* established — see the head-of-line note below — so stdin can keep being
|
||||||
|
* read while a slow call is outstanding.
|
||||||
// Parse request ID for error responses
|
*/
|
||||||
let requestId: unknown = null;
|
const inFlight = new Set<Promise<void>>();
|
||||||
try {
|
|
||||||
const parsed = JSON.parse(trimmed) as Record<string, unknown>;
|
|
||||||
requestId = parsed.id ?? null;
|
|
||||||
} catch {
|
|
||||||
// Non-JSON or notification — no id to respond to
|
|
||||||
}
|
|
||||||
|
|
||||||
|
/** POST one JSON-RPC message and write whatever comes back to stdout. */
|
||||||
|
const dispatch = async (trimmed: string, requestId: unknown): Promise<void> => {
|
||||||
try {
|
try {
|
||||||
const result = await postJsonRpc(endpointUrl, trimmed, sessionId, token);
|
const result = await postJsonRpc(endpointUrl, trimmed, sessionId, token);
|
||||||
|
|
||||||
@@ -178,9 +182,48 @@ export async function runMcpBridge(opts: McpBridgeOptions): Promise<void> {
|
|||||||
stdout.write(errorResponse + '\n');
|
stdout.write(errorResponse + '\n');
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
for await (const line of rl) {
|
||||||
|
const trimmed = line.trim();
|
||||||
|
if (!trimmed) continue;
|
||||||
|
|
||||||
|
// Parse request ID for error responses
|
||||||
|
let requestId: unknown = null;
|
||||||
|
try {
|
||||||
|
const parsed = JSON.parse(trimmed) as Record<string, unknown>;
|
||||||
|
requestId = parsed.id ?? null;
|
||||||
|
} catch {
|
||||||
|
// Non-JSON or notification — no id to respond to
|
||||||
}
|
}
|
||||||
|
|
||||||
// stdin closed — cleanup session
|
// HEAD-OF-LINE BLOCKING: this loop used to `await` every request, so the
|
||||||
|
// bridge handled exactly one at a time. A single slow call stalled EVERY
|
||||||
|
// later request — the client saw silence rather than an error, because the
|
||||||
|
// queued requests were never even sent, so nothing could time them out.
|
||||||
|
// Observed 2026-08-05: two gitea calls sat mute until the client aborted
|
||||||
|
// them at its own 1800s idle limit, while the upstream server was healthy
|
||||||
|
// and answering other sessions in milliseconds. JSON-RPC ids exist exactly
|
||||||
|
// so responses can come back out of order; nothing here needs a queue.
|
||||||
|
//
|
||||||
|
// We still serialise until the session id exists: it comes back on the
|
||||||
|
// first response, and firing later requests without it would open a second
|
||||||
|
// upstream session. In practice a client sends `initialize` first and waits
|
||||||
|
// for its reply anyway, so this costs one round trip, not throughput.
|
||||||
|
if (sessionId === undefined) {
|
||||||
|
await dispatch(trimmed, requestId);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
const p = dispatch(trimmed, requestId).finally(() => inFlight.delete(p));
|
||||||
|
inFlight.add(p);
|
||||||
|
}
|
||||||
|
|
||||||
|
// stdin closed — let outstanding work finish before tearing the session down,
|
||||||
|
// otherwise a concurrent call races the DELETE and dies with a 404.
|
||||||
|
if (inFlight.size > 0) {
|
||||||
|
await Promise.allSettled([...inFlight]);
|
||||||
|
}
|
||||||
if (sessionId) {
|
if (sessionId) {
|
||||||
await sendDelete(endpointUrl, sessionId, token);
|
await sendDelete(endpointUrl, sessionId, token);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -483,3 +483,105 @@ describe('createMcpCommand', () => {
|
|||||||
expect(parsed.opts().project).toBe('my-project');
|
expect(parsed.opts().project).toBe('my-project');
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
// Regression: head-of-line blocking (2026-08-05)
|
||||||
|
//
|
||||||
|
// The bridge used to `await` every request inside its stdin loop, so it handled
|
||||||
|
// exactly one at a time. A single slow call stalled every later request, and
|
||||||
|
// because those requests were never even sent, nothing could time them out —
|
||||||
|
// the client just saw silence until its own idle limit fired (30 min, in the
|
||||||
|
// incident that prompted this). These pin both halves of the fix: later
|
||||||
|
// requests must not queue behind a slow one, and a stalled request must produce
|
||||||
|
// a JSON-RPC error rather than nothing.
|
||||||
|
// ---------------------------------------------------------------------------
|
||||||
|
describe('MCP bridge concurrency', () => {
|
||||||
|
let srv: http.Server;
|
||||||
|
let port: number;
|
||||||
|
|
||||||
|
function sse(id: number | string, result: unknown) {
|
||||||
|
return `event: message\ndata: ${JSON.stringify({ jsonrpc: '2.0', id, result })}\n\n`;
|
||||||
|
}
|
||||||
|
|
||||||
|
beforeAll(async () => {
|
||||||
|
srv = http.createServer((req, res) => {
|
||||||
|
let body = '';
|
||||||
|
req.on('data', (c) => (body += c));
|
||||||
|
req.on('end', () => {
|
||||||
|
const msg = JSON.parse(body || '{}') as { id?: number | string; method?: string; params?: any };
|
||||||
|
const headers: Record<string, string> = { 'Content-Type': 'text/event-stream' };
|
||||||
|
if (msg.method === 'initialize') headers['mcp-session-id'] = 'sess-1';
|
||||||
|
// `slow` blocks far longer than `fast`, so a serial bridge would force
|
||||||
|
// fast's response to arrive second.
|
||||||
|
const delay = msg.params?.name === 'slow' ? 400 : 0;
|
||||||
|
setTimeout(() => {
|
||||||
|
res.writeHead(200, headers);
|
||||||
|
res.end(sse(msg.id ?? 0, { ok: msg.params?.name ?? msg.method }));
|
||||||
|
}, delay);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
await new Promise<void>((r) => srv.listen(0, '127.0.0.1', r));
|
||||||
|
port = (srv.address() as any).port;
|
||||||
|
});
|
||||||
|
|
||||||
|
afterAll(async () => {
|
||||||
|
await new Promise<void>((r) => srv.close(() => r()));
|
||||||
|
});
|
||||||
|
|
||||||
|
function bridge(lines: string[], out: string[]) {
|
||||||
|
const stdin = Readable.from(lines.map((l) => l + '\n'));
|
||||||
|
const stdout = new Writable({
|
||||||
|
write(chunk, _enc, cb) {
|
||||||
|
out.push(chunk.toString().trim());
|
||||||
|
cb();
|
||||||
|
},
|
||||||
|
});
|
||||||
|
const stderr = new Writable({ write(_c, _e, cb) { cb(); } });
|
||||||
|
return runMcpBridge({
|
||||||
|
projectName: 'p',
|
||||||
|
mcplocalUrl: `http://127.0.0.1:${port}`,
|
||||||
|
stdin,
|
||||||
|
stdout,
|
||||||
|
stderr,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
it('does not let a slow request block a later fast one', async () => {
|
||||||
|
const out: string[] = [];
|
||||||
|
await bridge(
|
||||||
|
[
|
||||||
|
JSON.stringify({ jsonrpc: '2.0', id: 1, method: 'initialize', params: {} }),
|
||||||
|
JSON.stringify({ jsonrpc: '2.0', id: 2, method: 'tools/call', params: { name: 'slow' } }),
|
||||||
|
JSON.stringify({ jsonrpc: '2.0', id: 3, method: 'tools/call', params: { name: 'fast' } }),
|
||||||
|
],
|
||||||
|
out,
|
||||||
|
);
|
||||||
|
const ids = out.map((l) => (JSON.parse(l) as { id: number }).id);
|
||||||
|
expect(ids).toContain(2);
|
||||||
|
expect(ids).toContain(3);
|
||||||
|
// The whole point: fast (id 3) overtakes slow (id 2). Serially it could not.
|
||||||
|
expect(ids.indexOf(3)).toBeLessThan(ids.indexOf(2));
|
||||||
|
});
|
||||||
|
|
||||||
|
it('answers with a JSON-RPC error instead of silence when a request fails', async () => {
|
||||||
|
const out: string[] = [];
|
||||||
|
const stdin = Readable.from([
|
||||||
|
JSON.stringify({ jsonrpc: '2.0', id: 1, method: 'initialize', params: {} }) + '\n',
|
||||||
|
]);
|
||||||
|
const stdout = new Writable({
|
||||||
|
write(chunk, _enc, cb) { out.push(chunk.toString().trim()); cb(); },
|
||||||
|
});
|
||||||
|
const stderr = new Writable({ write(_c, _e, cb) { cb(); } });
|
||||||
|
// Port with nothing on it: the POST fails fast, and the bridge must still
|
||||||
|
// emit a response carrying the original id.
|
||||||
|
await runMcpBridge({
|
||||||
|
projectName: 'p',
|
||||||
|
mcplocalUrl: 'http://127.0.0.1:1',
|
||||||
|
stdin, stdout, stderr,
|
||||||
|
});
|
||||||
|
expect(out.length).toBeGreaterThan(0);
|
||||||
|
const msg = JSON.parse(out[0]!) as { id: number; error?: { code: number } };
|
||||||
|
expect(msg.id).toBe(1);
|
||||||
|
expect(msg.error?.code).toBe(-32603);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|||||||
@@ -1,12 +1,16 @@
|
|||||||
/**
|
/**
|
||||||
* LLM-based prompt selection for the gating flow.
|
* LLM-based prompt selection for the gating flow.
|
||||||
*
|
*
|
||||||
* Sends tags + prompt index to the heavy LLM, which returns
|
* Sends tags + prompt index to a heavy LLM, which returns a ranked list of
|
||||||
* a ranked list of relevant prompt names.
|
* relevant prompt names. Credential tiering (user rule): prefer mcpd's server
|
||||||
|
* `Llm` (cloud/server keys live at the k8s level) via the inference proxy, and
|
||||||
|
* fall back to the local (personal-token) provider registry only when the
|
||||||
|
* server path is unavailable. Cloud keys never live in mcplocal config.
|
||||||
*/
|
*/
|
||||||
|
|
||||||
import type { ProviderRegistry } from '../providers/registry.js';
|
import type { ProviderRegistry } from '../providers/registry.js';
|
||||||
import type { SystemPromptFetcher } from '../proxymodel/types.js';
|
import type { SystemPromptFetcher } from '../proxymodel/types.js';
|
||||||
|
import type { ChatMessage, CompletionOptions } from '../providers/types.js';
|
||||||
|
|
||||||
export interface PromptIndexForLlm {
|
export interface PromptIndexForLlm {
|
||||||
name: string;
|
name: string;
|
||||||
@@ -20,6 +24,23 @@ export interface LlmSelectionResult {
|
|||||||
reasoning: string;
|
reasoning: string;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Route a selection inference through mcpd's server `Llm` (typically a
|
||||||
|
* `POST /api/v1/llms/:name/infer`). Returns the completion text. Must throw on
|
||||||
|
* transport/HTTP failure so the selector can fall back to the local provider.
|
||||||
|
*/
|
||||||
|
export type ServerInfer = (
|
||||||
|
messages: ChatMessage[],
|
||||||
|
opts: { maxTokens: number; temperature: number; signal?: AbortSignal },
|
||||||
|
) => Promise<string>;
|
||||||
|
|
||||||
|
export interface SelectPromptsOptions {
|
||||||
|
getSystemPromptFn?: SystemPromptFetcher;
|
||||||
|
signal?: AbortSignal;
|
||||||
|
/** Preferred inference path: mcpd server Llm (cloud/server keys at k8s). */
|
||||||
|
serverInfer?: ServerInfer;
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Token budget for the selection call. A reasoning model spends most of its
|
* Token budget for the selection call. A reasoning model spends most of its
|
||||||
* budget inside reasoning_content before emitting the {selectedNames} JSON, so
|
* budget inside reasoning_content before emitting the {selectedNames} JSON, so
|
||||||
@@ -31,18 +52,54 @@ const SELECT_MAX_TOKENS = (() => {
|
|||||||
return Number.isFinite(v) && v > 0 ? v : 4000;
|
return Number.isFinite(v) && v > 0 ? v : 4000;
|
||||||
})();
|
})();
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Pull the assistant text from an OpenAI-shaped completion, falling back to
|
||||||
|
* reasoning_content — thinking models emit their answer there with content null.
|
||||||
|
*/
|
||||||
|
export function pickCompletionText(resp: unknown): string {
|
||||||
|
const m = (resp as {
|
||||||
|
choices?: Array<{
|
||||||
|
message?: {
|
||||||
|
content?: string | null;
|
||||||
|
reasoning_content?: string | null;
|
||||||
|
provider_specific_fields?: { reasoning_content?: string | null };
|
||||||
|
};
|
||||||
|
}>;
|
||||||
|
}).choices?.[0]?.message;
|
||||||
|
const primary = m?.content ?? '';
|
||||||
|
if (primary !== '') return primary;
|
||||||
|
return m?.reasoning_content ?? m?.provider_specific_fields?.reasoning_content ?? '';
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Extract the `{ "selectedNames": [...], "reasoning": "..." }` object, or null. */
|
||||||
|
export function extractSelection(text: string): LlmSelectionResult | null {
|
||||||
|
const jsonMatch = text.match(/\{[\s\S]*"selectedNames"[\s\S]*\}/);
|
||||||
|
if (!jsonMatch) return null;
|
||||||
|
try {
|
||||||
|
const parsed = JSON.parse(jsonMatch[0]) as { selectedNames?: string[]; reasoning?: string };
|
||||||
|
return { selectedNames: parsed.selectedNames ?? [], reasoning: parsed.reasoning ?? '' };
|
||||||
|
} catch {
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
export class LlmPromptSelector {
|
export class LlmPromptSelector {
|
||||||
constructor(
|
constructor(
|
||||||
private readonly providerRegistry: ProviderRegistry,
|
private readonly providerRegistry: ProviderRegistry | null,
|
||||||
) {}
|
) {}
|
||||||
|
|
||||||
async selectPrompts(
|
async selectPrompts(
|
||||||
tags: string[],
|
tags: string[],
|
||||||
promptIndex: PromptIndexForLlm[],
|
promptIndex: PromptIndexForLlm[],
|
||||||
getSystemPromptFn?: SystemPromptFetcher,
|
opts: SelectPromptsOptions = {},
|
||||||
signal?: AbortSignal,
|
|
||||||
): Promise<LlmSelectionResult> {
|
): Promise<LlmSelectionResult> {
|
||||||
const DEFAULT_SYSTEM_PROMPT = `You are a context selection assistant. Given a developer's task keywords and a list of available project prompts, select which prompts are relevant to their work. Return a JSON object with "selectedNames" (array of prompt names) and "reasoning" (brief explanation). Priority 10 prompts must always be included.`;
|
const { getSystemPromptFn, signal, serverInfer } = opts;
|
||||||
|
|
||||||
|
// Keep the answer as SHORT as possible: latency is proportional to output
|
||||||
|
// length (a no-think model runs ~10-18 tok/s). Even an "<=8 words" reasoning
|
||||||
|
// got ignored (models write a full sentence → ~7s). Drop the reasoning field
|
||||||
|
// entirely — the gate only needs the names → just {selectedNames} → ~1-2s.
|
||||||
|
const DEFAULT_SYSTEM_PROMPT = `You are a context selection assistant. Given a developer's task keywords and available project prompts, select the relevant ones. Return ONLY this JSON and nothing else: {"selectedNames":[...]}. No reasoning, no prose, no explanation. Priority 10 prompts must always be included.`;
|
||||||
const systemPrompt = getSystemPromptFn
|
const systemPrompt = getSystemPromptFn
|
||||||
? await getSystemPromptFn('llm-gate-context-selector', DEFAULT_SYSTEM_PROMPT)
|
? await getSystemPromptFn('llm-gate-context-selector', DEFAULT_SYSTEM_PROMPT)
|
||||||
: DEFAULT_SYSTEM_PROMPT;
|
: DEFAULT_SYSTEM_PROMPT;
|
||||||
@@ -52,47 +109,64 @@ export class LlmPromptSelector {
|
|||||||
Available prompts:
|
Available prompts:
|
||||||
${promptIndex.map((p) => `- ${p.name} (priority: ${p.priority}): ${p.summary ?? 'No summary'}${p.chapters?.length ? `\n Chapters: ${p.chapters.join(', ')}` : ''}`).join('\n')}
|
${promptIndex.map((p) => `- ${p.name} (priority: ${p.priority}): ${p.summary ?? 'No summary'}${p.chapters?.length ? `\n Chapters: ${p.chapters.join(', ')}` : ''}`).join('\n')}
|
||||||
|
|
||||||
Select the relevant prompts. Return JSON: { "selectedNames": [...], "reasoning": "..." }`;
|
Select the relevant prompts. Return ONLY: {"selectedNames":[...]}`;
|
||||||
|
|
||||||
const provider = this.providerRegistry.getProvider('heavy');
|
const messages: ChatMessage[] = [
|
||||||
if (!provider) {
|
|
||||||
throw new Error('No heavy LLM provider available');
|
|
||||||
}
|
|
||||||
|
|
||||||
// Deliberately NOT setting `model` here. Prompt-ranking only needs a
|
|
||||||
// working LLM; forcing the project's vLLM model onto the (anthropic) heavy
|
|
||||||
// provider made every selection fail silently. Use the provider's own model.
|
|
||||||
const completionOptions: import('../providers/types.js').CompletionOptions = {
|
|
||||||
messages: [
|
|
||||||
{ role: 'system', content: systemPrompt },
|
{ role: 'system', content: systemPrompt },
|
||||||
{ role: 'user', content: userPrompt },
|
{ role: 'user', content: userPrompt },
|
||||||
],
|
];
|
||||||
|
|
||||||
|
// Sources in priority order: mcpd server Llm first (cloud/server keys at
|
||||||
|
// k8s), then the local personal-token provider as fallback.
|
||||||
|
const sources: Array<{ label: string; run: () => Promise<string> }> = [];
|
||||||
|
if (serverInfer) {
|
||||||
|
sources.push({
|
||||||
|
label: 'mcpd server Llm',
|
||||||
|
run: () => serverInfer(messages, { maxTokens: SELECT_MAX_TOKENS, temperature: 0, ...(signal ? { signal } : {}) }),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
if (this.providerRegistry) {
|
||||||
|
sources.push({
|
||||||
|
label: 'local provider',
|
||||||
|
run: async () => {
|
||||||
|
const provider = this.providerRegistry!.getProvider('heavy');
|
||||||
|
if (!provider) throw new Error('No heavy LLM provider available');
|
||||||
|
const completionOptions: CompletionOptions = {
|
||||||
|
messages,
|
||||||
temperature: 0,
|
temperature: 0,
|
||||||
maxTokens: SELECT_MAX_TOKENS,
|
maxTokens: SELECT_MAX_TOKENS,
|
||||||
...(signal ? { signal } : {}),
|
...(signal ? { signal } : {}),
|
||||||
};
|
};
|
||||||
|
return (await provider.complete(completionOptions)).content;
|
||||||
const result = await provider.complete(completionOptions);
|
},
|
||||||
|
});
|
||||||
const response = result.content;
|
}
|
||||||
|
if (sources.length === 0) {
|
||||||
// Parse JSON from response (may be wrapped in markdown code blocks)
|
throw new Error('No LLM provider available for prompt selection');
|
||||||
const jsonMatch = response.match(/\{[\s\S]*"selectedNames"[\s\S]*\}/);
|
|
||||||
if (!jsonMatch) {
|
|
||||||
throw new Error('LLM response did not contain valid selection JSON');
|
|
||||||
}
|
}
|
||||||
|
|
||||||
const parsed = JSON.parse(jsonMatch[0]) as { selectedNames?: string[]; reasoning?: string };
|
let lastErr: Error | null = null;
|
||||||
const selectedNames = parsed.selectedNames ?? [];
|
for (const src of sources) {
|
||||||
const reasoning = parsed.reasoning ?? '';
|
let text: string;
|
||||||
|
try {
|
||||||
// Always include priority 10 prompts
|
text = await src.run();
|
||||||
|
} catch (err) {
|
||||||
|
lastErr = err instanceof Error ? err : new Error(String(err));
|
||||||
|
continue; // transport/HTTP failure → try the next source
|
||||||
|
}
|
||||||
|
const sel = extractSelection(text);
|
||||||
|
if (!sel) {
|
||||||
|
lastErr = new Error(`${src.label}: LLM response did not contain valid selection JSON`);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
// Always include priority 10 prompts.
|
||||||
for (const p of promptIndex) {
|
for (const p of promptIndex) {
|
||||||
if (p.priority === 10 && !selectedNames.includes(p.name)) {
|
if (p.priority === 10 && !sel.selectedNames.includes(p.name)) {
|
||||||
selectedNames.push(p.name);
|
sel.selectedNames.push(p.name);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
return sel;
|
||||||
return { selectedNames, reasoning };
|
}
|
||||||
|
throw lastErr ?? new Error('LLM prompt-selection failed');
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -147,6 +147,14 @@ export function registerProjectMcpEndpoint(app: FastifyInstance, mcpdClient: Mcp
|
|||||||
providerRegistry: effectiveRegistry,
|
providerRegistry: effectiveRegistry,
|
||||||
};
|
};
|
||||||
if (resolvedModel) pluginConfig.modelOverride = resolvedModel;
|
if (resolvedModel) pluginConfig.modelOverride = resolvedModel;
|
||||||
|
// Route gate prompt-selection through a server Llm (mcpd inference proxy)
|
||||||
|
// so cloud/server keys stay at the k8s level; the local personal-token
|
||||||
|
// provider is the fallback. See credential-tiering. A dedicated fast
|
||||||
|
// (no-think) selection Llm can be pinned globally via
|
||||||
|
// MCPCTL_GATE_SELECTION_LLM — it overrides the project's chat llmProvider so
|
||||||
|
// selection stays fast while chat keeps its (thinking) model.
|
||||||
|
const gateSelectionLlm = process.env['MCPCTL_GATE_SELECTION_LLM'] || mcpdConfig.llmProvider;
|
||||||
|
if (gateSelectionLlm) pluginConfig.llmProvider = gateSelectionLlm;
|
||||||
const basePlugin = createDefaultPlugin(pluginConfig);
|
const basePlugin = createDefaultPlugin(pluginConfig);
|
||||||
// Optional favourite-index presentation: curated favourite/<tool> + full
|
// Optional favourite-index presentation: curated favourite/<tool> + full
|
||||||
// all/<server>/<tool> + a "prefer favourite/" instruction. Composed AFTER
|
// all/<server>/<tool> + a "prefer favourite/" instruction. Composed AFTER
|
||||||
|
|||||||
@@ -13,7 +13,7 @@ import type { ProxyModelPlugin, PluginSessionContext } from '../plugin.js';
|
|||||||
import { SessionGate } from '../../gate/session-gate.js';
|
import { SessionGate } from '../../gate/session-gate.js';
|
||||||
import { TagMatcher, extractKeywordsFromToolCall, tokenizeDescription } from '../../gate/tag-matcher.js';
|
import { TagMatcher, extractKeywordsFromToolCall, tokenizeDescription } from '../../gate/tag-matcher.js';
|
||||||
import type { TagMatchResult } from '../../gate/tag-matcher.js';
|
import type { TagMatchResult } from '../../gate/tag-matcher.js';
|
||||||
import { LlmPromptSelector } from '../../gate/llm-selector.js';
|
import { LlmPromptSelector, pickCompletionText, type ServerInfer } from '../../gate/llm-selector.js';
|
||||||
import type { ProviderRegistry } from '../../providers/registry.js';
|
import type { ProviderRegistry } from '../../providers/registry.js';
|
||||||
import { withTimeout, TimeoutError } from '../../util/with-timeout.js';
|
import { withTimeout, TimeoutError } from '../../util/with-timeout.js';
|
||||||
|
|
||||||
@@ -21,11 +21,38 @@ import { withTimeout, TimeoutError } from '../../util/with-timeout.js';
|
|||||||
* begin_session — on timeout we fall back to deterministic tag matching. */
|
* begin_session — on timeout we fall back to deterministic tag matching. */
|
||||||
const GATE_LLM_TIMEOUT_MS = Number(process.env['MCPCTL_GATE_LLM_TIMEOUT_MS'] ?? '8000');
|
const GATE_LLM_TIMEOUT_MS = Number(process.env['MCPCTL_GATE_LLM_TIMEOUT_MS'] ?? '8000');
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Extra request fields for the gate's server-Llm selection call. Prompt
|
||||||
|
* selection is mechanical classification that gains nothing from chain-of-
|
||||||
|
* thought, so we ask reasoning models to skip thinking (they otherwise burn the
|
||||||
|
* whole budget reasoning and blow the gate timeout). Sent verbatim by mcpd's
|
||||||
|
* passthrough adapter → litellm/vLLM. Harmless for models that ignore them, and
|
||||||
|
* if a backend rejects them the selector falls back to the local provider.
|
||||||
|
* - `chat_template_kwargs.enable_thinking:false` — Qwen3 hard-off.
|
||||||
|
* - `reasoning_effort:'low'` — OpenAI o-series / newer vLLM.
|
||||||
|
* Override with a JSON object in MCPCTL_GATE_SELECT_EXTRA_BODY, or '' to disable.
|
||||||
|
*/
|
||||||
|
const GATE_SELECT_EXTRA_BODY: Record<string, unknown> = (() => {
|
||||||
|
const raw = process.env['MCPCTL_GATE_SELECT_EXTRA_BODY'];
|
||||||
|
if (raw === '') return {};
|
||||||
|
if (raw !== undefined) {
|
||||||
|
try { return JSON.parse(raw) as Record<string, unknown>; } catch { /* use default */ }
|
||||||
|
}
|
||||||
|
return { chat_template_kwargs: { enable_thinking: false }, reasoning_effort: 'low' };
|
||||||
|
})();
|
||||||
|
|
||||||
export interface GatePluginConfig {
|
export interface GatePluginConfig {
|
||||||
gated?: boolean;
|
gated?: boolean;
|
||||||
providerRegistry?: ProviderRegistry | null;
|
providerRegistry?: ProviderRegistry | null;
|
||||||
modelOverride?: string;
|
modelOverride?: string;
|
||||||
byteBudget?: number;
|
byteBudget?: number;
|
||||||
|
/**
|
||||||
|
* Name of the project's server `Llm` (mcpd). When set (and not 'none'), gate
|
||||||
|
* prompt-selection routes through mcpd's inference proxy first — keeping
|
||||||
|
* cloud/server keys at the k8s level — and only falls back to the local
|
||||||
|
* personal-token provider registry. See credential-tiering principle.
|
||||||
|
*/
|
||||||
|
llmProvider?: string;
|
||||||
}
|
}
|
||||||
|
|
||||||
const MAX_RESPONSE_CHARS = 24_000;
|
const MAX_RESPONSE_CHARS = 24_000;
|
||||||
@@ -33,8 +60,12 @@ const MAX_RESPONSE_CHARS = 24_000;
|
|||||||
export function createGatePlugin(config: GatePluginConfig = {}): ProxyModelPlugin {
|
export function createGatePlugin(config: GatePluginConfig = {}): ProxyModelPlugin {
|
||||||
const isGated = config.gated !== false;
|
const isGated = config.gated !== false;
|
||||||
const tagMatcher = new TagMatcher(config.byteBudget);
|
const tagMatcher = new TagMatcher(config.byteBudget);
|
||||||
const llmSelector = config.providerRegistry
|
// Prefer routing selection through mcpd's server Llm (cloud/server keys at
|
||||||
? new LlmPromptSelector(config.providerRegistry)
|
// k8s); the local provider registry (personal tokens) is the fallback. A gate
|
||||||
|
// with a server Llm but no local providers still selects (server-only).
|
||||||
|
const serverLlm = config.llmProvider && config.llmProvider !== 'none' ? config.llmProvider : undefined;
|
||||||
|
const llmSelector = (config.providerRegistry || serverLlm)
|
||||||
|
? new LlmPromptSelector(config.providerRegistry ?? null)
|
||||||
: null;
|
: null;
|
||||||
|
|
||||||
// Per-session state tracking (plugin-scoped, not global SessionGate)
|
// Per-session state tracking (plugin-scoped, not global SessionGate)
|
||||||
@@ -49,7 +80,7 @@ export function createGatePlugin(config: GatePluginConfig = {}): ProxyModelPlugi
|
|||||||
|
|
||||||
// Register begin_session virtual tool
|
// Register begin_session virtual tool
|
||||||
ctx.registerTool(getBeginSessionTool(llmSelector), async (args, callCtx) => {
|
ctx.registerTool(getBeginSessionTool(llmSelector), async (args, callCtx) => {
|
||||||
return handleBeginSession(args, callCtx, sessionGate, tagMatcher, llmSelector);
|
return handleBeginSession(args, callCtx, sessionGate, tagMatcher, llmSelector, serverLlm);
|
||||||
});
|
});
|
||||||
|
|
||||||
// Register read_prompts virtual tool (available even when ungated)
|
// Register read_prompts virtual tool (available even when ungated)
|
||||||
@@ -250,6 +281,7 @@ async function handleBeginSession(
|
|||||||
sessionGate: SessionGate,
|
sessionGate: SessionGate,
|
||||||
tagMatcher: TagMatcher,
|
tagMatcher: TagMatcher,
|
||||||
llmSelector: LlmPromptSelector | null,
|
llmSelector: LlmPromptSelector | null,
|
||||||
|
serverLlm?: string,
|
||||||
): Promise<unknown> {
|
): Promise<unknown> {
|
||||||
const rawTags = args['tags'] as string[] | undefined;
|
const rawTags = args['tags'] as string[] | undefined;
|
||||||
const description = args['description'] as string | undefined;
|
const description = args['description'] as string | undefined;
|
||||||
@@ -287,8 +319,26 @@ async function handleBeginSession(
|
|||||||
chapters: p.chapters,
|
chapters: p.chapters,
|
||||||
}));
|
}));
|
||||||
const getSystemPromptFn = ctx.getSystemPrompt.bind(ctx);
|
const getSystemPromptFn = ctx.getSystemPrompt.bind(ctx);
|
||||||
|
// Prefer mcpd's server Llm (keeps cloud/server keys at k8s); the selector
|
||||||
|
// falls back to the local personal-token provider on failure.
|
||||||
|
const serverInfer: ServerInfer | undefined = serverLlm
|
||||||
|
? async (messages, o) => {
|
||||||
|
const resp = await ctx.postToMcpd(`/api/v1/llms/${encodeURIComponent(serverLlm)}/infer`, {
|
||||||
|
messages,
|
||||||
|
temperature: o.temperature,
|
||||||
|
max_tokens: o.maxTokens,
|
||||||
|
stream: false,
|
||||||
|
...GATE_SELECT_EXTRA_BODY, // ask reasoning models for a fast, no-think answer
|
||||||
|
});
|
||||||
|
return pickCompletionText(resp);
|
||||||
|
}
|
||||||
|
: undefined;
|
||||||
const llmResult = await withTimeout(
|
const llmResult = await withTimeout(
|
||||||
(signal) => llmSelector.selectPrompts(tags, llmIndex, getSystemPromptFn, signal),
|
(signal) => llmSelector.selectPrompts(tags, llmIndex, {
|
||||||
|
getSystemPromptFn,
|
||||||
|
signal,
|
||||||
|
...(serverInfer ? { serverInfer } : {}),
|
||||||
|
}),
|
||||||
GATE_LLM_TIMEOUT_MS,
|
GATE_LLM_TIMEOUT_MS,
|
||||||
'gate LLM prompt-selection',
|
'gate LLM prompt-selection',
|
||||||
);
|
);
|
||||||
|
|||||||
@@ -109,12 +109,49 @@ describe('LlmPromptSelector', () => {
|
|||||||
const selector = new LlmPromptSelector(registry);
|
const selector = new LlmPromptSelector(registry);
|
||||||
const ac = new AbortController();
|
const ac = new AbortController();
|
||||||
|
|
||||||
await selector.selectPrompts(['test'], sampleIndex, undefined, ac.signal);
|
await selector.selectPrompts(['test'], sampleIndex, { signal: ac.signal });
|
||||||
|
|
||||||
const call = (provider.complete as ReturnType<typeof vi.fn>).mock.calls[0]![0] as CompletionOptions;
|
const call = (provider.complete as ReturnType<typeof vi.fn>).mock.calls[0]![0] as CompletionOptions;
|
||||||
expect(call.signal).toBe(ac.signal);
|
expect(call.signal).toBe(ac.signal);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it('prefers the mcpd server Llm (serverInfer) over the local provider', async () => {
|
||||||
|
const provider = makeMockProvider('{ "selectedNames": ["mqtt-config"], "reasoning": "local" }');
|
||||||
|
const registry = makeRegistry(provider);
|
||||||
|
const selector = new LlmPromptSelector(registry);
|
||||||
|
const serverInfer = vi.fn().mockResolvedValue('{ "selectedNames": ["zigbee-pairing"], "reasoning": "server" }');
|
||||||
|
|
||||||
|
const result = await selector.selectPrompts(['x'], sampleIndex, { serverInfer });
|
||||||
|
|
||||||
|
expect(serverInfer).toHaveBeenCalledOnce();
|
||||||
|
expect(provider.complete).not.toHaveBeenCalled(); // server succeeded → no local call
|
||||||
|
expect(result.selectedNames).toContain('zigbee-pairing');
|
||||||
|
expect(result.reasoning).toBe('server');
|
||||||
|
});
|
||||||
|
|
||||||
|
it('falls back to the local provider when serverInfer throws', async () => {
|
||||||
|
const provider = makeMockProvider('{ "selectedNames": ["mqtt-config"], "reasoning": "local" }');
|
||||||
|
const registry = makeRegistry(provider);
|
||||||
|
const selector = new LlmPromptSelector(registry);
|
||||||
|
const serverInfer = vi.fn().mockRejectedValue(new Error('mcpd HTTP 503'));
|
||||||
|
|
||||||
|
const result = await selector.selectPrompts(['x'], sampleIndex, { serverInfer });
|
||||||
|
|
||||||
|
expect(serverInfer).toHaveBeenCalledOnce();
|
||||||
|
expect(provider.complete).toHaveBeenCalledOnce();
|
||||||
|
expect(result.selectedNames).toContain('mqtt-config');
|
||||||
|
expect(result.reasoning).toBe('local');
|
||||||
|
});
|
||||||
|
|
||||||
|
it('works server-only (null registry) when a serverInfer is supplied', async () => {
|
||||||
|
const selector = new LlmPromptSelector(null);
|
||||||
|
const serverInfer = vi.fn().mockResolvedValue('{ "selectedNames": ["mqtt-config"], "reasoning": "s" }');
|
||||||
|
|
||||||
|
const result = await selector.selectPrompts(['x'], sampleIndex, { serverInfer });
|
||||||
|
|
||||||
|
expect(result.selectedNames).toContain('mqtt-config');
|
||||||
|
});
|
||||||
|
|
||||||
it('throws when no heavy provider is available', async () => {
|
it('throws when no heavy provider is available', async () => {
|
||||||
const registry = new ProviderRegistry(); // Empty registry
|
const registry = new ProviderRegistry(); // Empty registry
|
||||||
const selector = new LlmPromptSelector(registry);
|
const selector = new LlmPromptSelector(registry);
|
||||||
|
|||||||
Reference in New Issue
Block a user