diff --git a/packages/ai-client/src/providers/localRuntimeFetch.test.ts b/packages/ai-client/src/providers/localRuntimeFetch.test.ts new file mode 100644 index 00000000..92eadd13 --- /dev/null +++ b/packages/ai-client/src/providers/localRuntimeFetch.test.ts @@ -0,0 +1,62 @@ +import { createServer, type Server } from 'node:http'; +import type { AddressInfo } from 'node:net'; +import { afterEach, describe, expect, it } from 'vitest'; +import { localRuntimeFetch } from './localRuntimeFetch.js'; + +let server: Server | undefined; + +afterEach(async () => { + await new Promise((resolve) => server?.close(() => resolve()) ?? resolve()); + server = undefined; +}); + +async function serve(handler: Parameters[1]): Promise { + server = createServer(handler); + await new Promise((resolve) => server!.listen(0, '127.0.0.1', resolve)); + return `http://127.0.0.1:${(server!.address() as AddressInfo).port}`; +} + +describe('localRuntimeFetch', () => { + it('reaches a real server on this Node, whatever undici Node bundles', async () => { + const base = await serve((req, res) => { + let body = ''; + req.on('data', (chunk: Buffer) => (body += chunk.toString())); + req.on('end', () => { + res.setHeader('content-type', 'application/json'); + res.end(JSON.stringify({ method: req.method, body })); + }); + }); + + const res = await localRuntimeFetch(`${base}/api/chat`, { method: 'POST', body: '{"a":1}' }); + + expect(res.ok).toBe(true); + expect(await res.json()).toEqual({ method: 'POST', body: '{"a":1}' }); + }); + + it('streams a body as it arrives', async () => { + const base = await serve((_req, res) => { + res.write('{"status":"pulling"}\n'); + setTimeout(() => res.end('{"status":"success"}\n'), 20); + }); + + const res = await localRuntimeFetch(`${base}/api/pull`); + const reader = res.body!.getReader(); + const decoder = new TextDecoder(); + let text = ''; + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + text += decoder.decode(value); + } + + expect(text.trim().split('\n')).toHaveLength(2); + }); + + it('ends on the caller’s signal, the only limit it keeps', async () => { + const base = await serve(() => undefined); + + await expect( + localRuntimeFetch(`${base}/api/chat`, { signal: AbortSignal.timeout(50) }), + ).rejects.toMatchObject({ name: expect.stringMatching(/AbortError|TimeoutError/) }); + }); +}); diff --git a/packages/ai-client/src/providers/localRuntimeFetch.ts b/packages/ai-client/src/providers/localRuntimeFetch.ts index 9bf3e0dd..90e6f6e3 100644 --- a/packages/ai-client/src/providers/localRuntimeFetch.ts +++ b/packages/ai-client/src/providers/localRuntimeFetch.ts @@ -9,11 +9,20 @@ * unreachable runtime, and every retry read the prompt again from the start. * The caller's signal is the only limit: the platform's idle window for a * local model on a turn, the lease on a download. + * + * Undici's own fetch, not Node's: Node's is the undici it bundles, and it + * refuses an Agent from another major (`invalid onRequestStart method`) — + * which a hoisted copy of another version supplies — so every request failed + * and read as a runtime not answering. A fetch and an Agent from one package + * agree whatever Node bundles or the install hoists. */ -import { Agent } from 'undici'; +import { Agent, fetch as undiciFetch } from 'undici'; const NO_TRANSPORT_LIMIT = new Agent({ headersTimeout: 0, bodyTimeout: 0 }); export function localRuntimeFetch(url: string, init: RequestInit = {}): Promise { - return fetch(url, { ...init, dispatcher: NO_TRANSPORT_LIMIT } as RequestInit); + return undiciFetch(url, { + ...init, + dispatcher: NO_TRANSPORT_LIMIT, + } as Parameters[1]) as unknown as Promise; } diff --git a/packages/ai-client/src/providers/ollama.test.ts b/packages/ai-client/src/providers/ollama.test.ts index 20d91338..fcbb7a10 100644 --- a/packages/ai-client/src/providers/ollama.test.ts +++ b/packages/ai-client/src/providers/ollama.test.ts @@ -5,6 +5,11 @@ import { createAIClient } from '../client.js'; import { AIClientError } from '../errors.js'; import { createOllamaAdapter } from './ollama.js'; +// The transport is undici's own fetch; these tests answer through the global one. +vi.mock('./localRuntimeFetch.js', () => ({ + localRuntimeFetch: (url: string, init?: RequestInit) => globalThis.fetch(url, init), +})); + const context = { tenantId: 'tenant' as TenantId, runId: 'run' as SessionId, @@ -481,25 +486,6 @@ describe('failures', () => { expect(error.message).toContain('other side closed (ECONNRESET)'); }); - it('waits on the runtime with no transport limit of its own, so only the caller’s signal ends a long prompt', async () => { - const seen: RequestInit[] = []; - vi.stubGlobal( - 'fetch', - vi.fn(async (_url: string, init: RequestInit) => { - seen.push(init); - return done({ content: 'ok' }, { prompt: 5, eval: 1 })(); - }), - ); - - await createOllamaAdapter({}).generateText({ - ...context, - model: 'm', - messages: [{ role: 'user', content: 'x' }], - }); - - expect((seen[0] as { dispatcher?: unknown }).dispatcher).toBeDefined(); - }); - it('does not retry a runtime that stayed silent through the idle window: a retry reads the prompt again', async () => { const controller = new AbortController(); controller.abort({ marker: 'phoenix.executor.timeout', timeoutMs: 900_000, kind: 'idle' }); diff --git a/packages/ai-client/src/providers/ollamaRuntime.test.ts b/packages/ai-client/src/providers/ollamaRuntime.test.ts index 145ec70e..00cded33 100644 --- a/packages/ai-client/src/providers/ollamaRuntime.test.ts +++ b/packages/ai-client/src/providers/ollamaRuntime.test.ts @@ -5,6 +5,11 @@ import { type LoadedLocalModel, } from './ollamaRuntime.js'; +// The transport is undici's own fetch; these tests answer through the global one. +vi.mock('./localRuntimeFetch.js', () => ({ + localRuntimeFetch: (url: string, init?: RequestInit) => globalThis.fetch(url, init), +})); + afterEach(() => { vi.unstubAllGlobals(); }); diff --git a/scripts/installState.mjs b/scripts/installState.mjs new file mode 100644 index 00000000..b7eed88d --- /dev/null +++ b/scripts/installState.mjs @@ -0,0 +1,19 @@ +import { statSync } from 'node:fs'; +import { join } from 'node:path'; + +/** + * Whether the lockfile changed after the last `yarn install` finished, which + * writes its state file last. No state file means nothing was ever installed. + */ +export function installIsStale(repo) { + const mtime = (file) => { + try { + return statSync(join(repo, file)).mtimeMs; + } catch { + return undefined; + } + }; + const installed = mtime('node_modules/.yarn-state.yml'); + const lock = mtime('yarn.lock'); + return installed === undefined || (lock !== undefined && lock > installed); +} diff --git a/scripts/installState.test.mjs b/scripts/installState.test.mjs new file mode 100644 index 00000000..9af221df --- /dev/null +++ b/scripts/installState.test.mjs @@ -0,0 +1,32 @@ +import { mkdirSync, mkdtempSync, utimesSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { describe, expect, it } from 'vitest'; + +import { installIsStale } from './installState.mjs'; + +function checkout({ lockAt, installedAt }) { + const repo = mkdtempSync(join(tmpdir(), 'install-state-')); + writeFileSync(join(repo, 'yarn.lock'), ''); + utimesSync(join(repo, 'yarn.lock'), lockAt, lockAt); + if (installedAt !== undefined) { + mkdirSync(join(repo, 'node_modules')); + writeFileSync(join(repo, 'node_modules', '.yarn-state.yml'), ''); + utimesSync(join(repo, 'node_modules', '.yarn-state.yml'), installedAt, installedAt); + } + return repo; +} + +describe('installIsStale', () => { + it('is stale when the lockfile changed after the last install', () => { + expect(installIsStale(checkout({ lockAt: 2_000, installedAt: 1_000 }))).toBe(true); + }); + + it('is current when the install ran after the lockfile last changed', () => { + expect(installIsStale(checkout({ lockAt: 1_000, installedAt: 2_000 }))).toBe(false); + }); + + it('is stale when nothing was ever installed', () => { + expect(installIsStale(checkout({ lockAt: 1_000, installedAt: undefined }))).toBe(true); + }); +}); diff --git a/scripts/start.mjs b/scripts/start.mjs index be5e1efc..ea02b996 100644 --- a/scripts/start.mjs +++ b/scripts/start.mjs @@ -38,6 +38,7 @@ import { stackEnvPath, } from './stackEnv.mjs'; import { probeRedis } from './stackRedis.mjs'; +import { installIsStale } from './installState.mjs'; const REPO = join(dirname(fileURLToPath(import.meta.url)), '..'); const say = (message) => { @@ -120,6 +121,18 @@ say('credentials:'); for (const line of readiness.lines) say(` ${line}`); if (readiness.failure !== undefined) fail(readiness.failure.message, readiness.failure.remedy); +// ── Installed dependencies ──────────────────────────────────────────────────── +// A merge that adds a dependency changes the lockfile and nothing else, and every +// service then resolves whatever the old install left — another version of the +// package, hoisted from elsewhere, which can fail only at its first call, as +// something unrelated. +if (installIsStale(REPO)) { + say('yarn.lock changed since the last install — installing'); + if (spawnSync('yarn', ['install'], { cwd: REPO, stdio: 'inherit' }).status !== 0) { + fail('yarn install failed.'); + } +} + // ── Build output the web application reads ──────────────────────────────────── // Its bundler resolves the `import` condition, so it reads compiled packages even // in development. Absent, it fails as a module that cannot be found rather than a