feat(cli+docs+smoke): inference-task CLI + GC ticker + smoke + docs (v5 Stage 4)
Some checks failed
CI/CD / lint (pull_request) Successful in 55s
CI/CD / test (pull_request) Successful in 1m12s
CI/CD / typecheck (pull_request) Successful in 2m46s
CI/CD / smoke (pull_request) Failing after 1m44s
CI/CD / build (pull_request) Failing after 7m0s
CI/CD / publish (pull_request) Has been skipped

CLI surface for the durable queue:

- `mcpctl get tasks` — table view (ID, STATUS, POOL, LLM, MODEL,
  STREAM, AGE, WORKER). Aliases `task`, `tasks`, `inference-task`,
  `inference-tasks` all normalize to the canonical plural so URL
  construction works uniformly. RESOURCE_ALIASES + completions
  generator updated.
- `mcpctl chat-llm <name> --async -m <msg>` — enqueue and exit. stdout
  is just the task id (pipeable into `xargs mcpctl get task`); stderr
  carries human-readable status. REPL mode is rejected for --async
  (fire-and-forget doesn't make sense without -m).

GC ticker in mcpd: 5-min interval. Pending tasks past 1 h queue
timeout flip to error with a clear message; terminal tasks past 7 d
retention get deleted. Both queries are index-backed.

Crash fix uncovered by the smoke: when the async route doesn't await
ref.done, a later cancel/error rejected the in-flight Promise as
unhandled and crashed mcpd. The route now attaches a no-op `.catch`
so the legacy `done` semantic still works for sync callers (chat,
direct infer) without taking out the process for async ones. The
EnqueueInferOptions also gained an explicit `ownerId` field so the
async API can stamp the authenticated user on the row instead of
inheriting 'system' from the constructor's resolveOwner — without
this, every GET/DELETE from the original caller would 404 due to
foreign-owner mismatch.

Smoke (tests/smoke/inference-task.smoke.test.ts):

  1. POST /inference-tasks while no worker bound → row=pending.
  2. Bring a registrar online → bindSession drain claims and
     dispatches → worker complete()s → row=completed → GET returns
     the assistant body.
  3. Stop worker, enqueue, DELETE → row=cancelled, persisted.

docs/inference-tasks.md (new): full data model, lifecycle diagram,
async API reference, CLI examples, RBAC table, GC defaults, and the
v5 limitations / v6 roadmap. Cross-linked from virtual-llms.md and
agents.md.

Tests + smoke: mcpd 893/893, mcplocal 723/723, cli 437/437, full
smoke 146/146 (was 144, +2 new task smoke). Live mcpd verified via
manual curl: enqueue → cancel → re-fetch — no crash, owner scoping
returns 404 on foreign ids, GC ticker logs at info when it sweeps.

v5 complete: durable queue (Stage 1) + VirtualLlmService rewire
(Stage 2) + async API & RBAC (Stage 3) + CLI/GC/smoke/docs (Stage 4).
This commit is contained in:
Michal
2026-04-28 15:25:09 +01:00
parent 1dcfdc8b05
commit 7320b50dac
14 changed files with 654 additions and 27 deletions

View File

@@ -0,0 +1,255 @@
/**
* v5 smoke: durable inference task queue end-to-end.
*
* 1. POST /api/v1/inference-tasks while NO worker is bound — task
* should land as status=pending.
* 2. Bring a worker online (in-process VirtualLlmRegistrar) — the
* bindSession drain claims the queued task and pushes it.
* 3. Worker runs the fake provider's complete(), POSTs the result —
* task transitions to status=completed.
* 4. GET /api/v1/inference-tasks/<id> returns the persisted body.
* 5. DELETE /api/v1/inference-tasks/<id> for a still-pending task
* cancels it before any worker picks it up.
*
* Pre-v5 the no-worker case 503'd immediately; the durability is the
* whole point of v5.
*/
import { describe, it, expect, beforeAll, afterAll } from 'vitest';
import http from 'node:http';
import https from 'node:https';
import { mkdtempSync, rmSync, readFileSync, existsSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import {
VirtualLlmRegistrar,
type RegistrarPublishedProvider,
} from '../../src/providers/registrar.js';
import type { LlmProvider, CompletionResult } from '../../src/providers/types.js';
const MCPD_URL = process.env.MCPD_URL ?? 'https://mcpctl.ad.itaz.eu';
const SUFFIX = Date.now().toString(36);
const PROVIDER_NAME = `smoke-task-${SUFFIX}`;
function makeFakeProvider(name: string, content: string): LlmProvider {
return {
name,
async complete(): Promise<CompletionResult> {
return {
content,
toolCalls: [],
usage: { promptTokens: 1, completionTokens: 4, totalTokens: 5 },
finishReason: 'stop',
};
},
async listModels() { return []; },
async isAvailable() { return true; },
};
}
function readToken(): string | null {
try {
const path = join(process.env.HOME ?? '', '.mcpctl', 'credentials');
if (!existsSync(path)) return null;
const parsed = JSON.parse(readFileSync(path, 'utf-8')) as { token?: string };
return parsed.token ?? null;
} catch {
return null;
}
}
function healthz(url: string, timeoutMs = 5000): Promise<boolean> {
return new Promise((resolve) => {
const parsed = new URL(`${url.replace(/\/$/, '')}/healthz`);
const driver = parsed.protocol === 'https:' ? https : http;
const req = driver.get({
hostname: parsed.hostname,
port: parsed.port || (parsed.protocol === 'https:' ? 443 : 80),
path: parsed.pathname,
timeout: timeoutMs,
}, (res) => { resolve((res.statusCode ?? 500) < 500); res.resume(); });
req.on('error', () => resolve(false));
req.on('timeout', () => { req.destroy(); resolve(false); });
});
}
interface HttpResponse { status: number; body: string }
function httpRequest(method: string, urlStr: string, body: unknown): Promise<HttpResponse> {
return new Promise((resolve, reject) => {
const tokenRaw = readToken();
const parsed = new URL(urlStr);
const driver = parsed.protocol === 'https:' ? https : http;
const headers: Record<string, string> = {
Accept: 'application/json',
...(body !== undefined ? { 'Content-Type': 'application/json' } : {}),
...(tokenRaw !== null ? { Authorization: `Bearer ${tokenRaw}` } : {}),
};
const req = driver.request({
hostname: parsed.hostname,
port: parsed.port || (parsed.protocol === 'https:' ? 443 : 80),
path: parsed.pathname + parsed.search,
method,
headers,
timeout: 30_000,
}, (res) => {
const chunks: Buffer[] = [];
res.on('data', (c: Buffer) => chunks.push(c));
res.on('end', () => {
resolve({ status: res.statusCode ?? 0, body: Buffer.concat(chunks).toString('utf-8') });
});
});
req.on('error', reject);
req.on('timeout', () => { req.destroy(); reject(new Error(`httpRequest timeout: ${method} ${urlStr}`)); });
if (body !== undefined) req.write(JSON.stringify(body));
req.end();
});
}
let mcpdUp = false;
let registrar: VirtualLlmRegistrar | null = null;
let tempDir: string;
describe('inference-task smoke (v5)', () => {
beforeAll(async () => {
mcpdUp = await healthz(MCPD_URL);
if (!mcpdUp) {
// eslint-disable-next-line no-console
console.warn(`\n ○ inference-task smoke: skipped — ${MCPD_URL}/healthz unreachable.\n`);
return;
}
if (readToken() === null) {
mcpdUp = false;
// eslint-disable-next-line no-console
console.warn('\n ○ inference-task smoke: skipped — no ~/.mcpctl/credentials.\n');
return;
}
tempDir = mkdtempSync(join(tmpdir(), 'mcpctl-task-smoke-'));
}, 20_000);
afterAll(async () => {
if (registrar !== null) registrar.stop();
if (tempDir !== undefined) rmSync(tempDir, { recursive: true, force: true });
if (!mcpdUp) return;
// Best-effort cleanup of the Llm row (tasks stay until GC retention).
const list = await httpRequest('GET', `${MCPD_URL}/api/v1/llms`, undefined);
if (list.status === 200) {
const rows = JSON.parse(list.body) as Array<{ id: string; name: string }>;
const row = rows.find((r) => r.name === PROVIDER_NAME);
if (row !== undefined) {
await httpRequest('DELETE', `${MCPD_URL}/api/v1/llms/${row.id}`, undefined);
}
}
});
it('end-to-end: enqueue without worker → bind → drain → completed', async () => {
if (!mcpdUp) return;
const token = readToken();
if (token === null) return;
// Step 1: register the virtual Llm WITHOUT binding the SSE session.
// We do this by issuing a register POST directly, then NOT starting
// the registrar's stream. v5 actually re-uses the same path when
// the registrar starts, so the cleanest way is to register-then-stop
// before the inference task lands.
//
// Simpler approach: register normally, immediately stop, leaving
// the row inactive on mcpd's side. Then enqueue (queues), then
// bring up a fresh registrar — its bind drains.
const tempReg = new VirtualLlmRegistrar({
mcpdUrl: MCPD_URL,
token,
publishedProviders: [{
provider: makeFakeProvider(PROVIDER_NAME, 'reply from worker'),
type: 'openai',
model: 'fake-task',
}],
sessionFilePath: join(tempDir, 'session-1'),
log: { info: () => {}, warn: () => {}, error: () => {} },
heartbeatIntervalMs: 60_000,
});
await tempReg.start();
// Stop the registrar so the SSE channel closes; mcpd flips the row
// to inactive immediately on unbindSession. Future bindSession with
// the SAME providerSessionId reactivates the row + drains queued
// tasks (sticky reconnect, v1+ behavior).
tempReg.stop();
// Give mcpd a beat to process the disconnect.
await new Promise((r) => setTimeout(r, 500));
// Step 2: enqueue while no worker is bound. Task should be pending.
const enq = await httpRequest('POST', `${MCPD_URL}/api/v1/inference-tasks`, {
llmName: PROVIDER_NAME,
request: { messages: [{ role: 'user', content: 'hello queue' }] },
streaming: false,
});
expect(enq.status, enq.body).toBe(201);
const created = JSON.parse(enq.body) as { id: string; status: string; poolName: string };
expect(created.status).toBe('pending');
expect(created.id).toMatch(/^c[a-z0-9]+/);
// Step 3: bring a worker online — same providerSessionId so it
// adopts the row, drainPendingForSession picks up the queued task.
registrar = new VirtualLlmRegistrar({
mcpdUrl: MCPD_URL,
token,
publishedProviders: [{
provider: makeFakeProvider(PROVIDER_NAME, 'reply from worker'),
type: 'openai',
model: 'fake-task',
}],
sessionFilePath: join(tempDir, 'session-1'),
log: { info: () => {}, warn: () => {}, error: () => {} },
heartbeatIntervalMs: 60_000,
});
await registrar.start();
// Step 4: poll for completion. The provider's complete() returns
// synchronously so the worker should resolve very quickly — but
// the SSE round-trip takes at least one tick, so we poll up to 5s.
let final: { status: string; responseBody?: { choices?: Array<{ message?: { content?: string } }> } } | null = null;
for (let i = 0; i < 25; i += 1) {
await new Promise((r) => setTimeout(r, 200));
const res = await httpRequest('GET', `${MCPD_URL}/api/v1/inference-tasks/${created.id}`, undefined);
if (res.status !== 200) continue;
const row = JSON.parse(res.body) as typeof final;
if (row !== null && (row.status === 'completed' || row.status === 'error' || row.status === 'cancelled')) {
final = row;
break;
}
}
expect(final?.status, 'task should have reached completed').toBe('completed');
expect(final?.responseBody?.choices?.[0]?.message?.content).toBe('reply from worker');
}, 30_000);
it('cancel: DELETE on a pending task transitions it to cancelled', async () => {
if (!mcpdUp) return;
const token = readToken();
if (token === null) return;
// Stop the worker (if any) so this enqueue stays pending. A short
// sleep gives unbindSession a moment.
if (registrar !== null) {
registrar.stop();
registrar = null;
}
await new Promise((r) => setTimeout(r, 400));
const enq = await httpRequest('POST', `${MCPD_URL}/api/v1/inference-tasks`, {
llmName: PROVIDER_NAME,
request: { messages: [{ role: 'user', content: 'will be cancelled' }] },
streaming: false,
});
expect(enq.status, enq.body).toBe(201);
const created = JSON.parse(enq.body) as { id: string; status: string };
expect(created.status).toBe('pending');
const cancelRes = await httpRequest('DELETE', `${MCPD_URL}/api/v1/inference-tasks/${created.id}`, undefined);
expect(cancelRes.status).toBe(200);
const cancelled = JSON.parse(cancelRes.body) as { status: string };
expect(cancelled.status).toBe('cancelled');
// Re-fetch confirms persistence.
const after = await httpRequest('GET', `${MCPD_URL}/api/v1/inference-tasks/${created.id}`, undefined);
expect(JSON.parse(after.body).status).toBe('cancelled');
}, 20_000);
});