Skip to content
Merged
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
62 changes: 62 additions & 0 deletions packages/ai-client/src/providers/localRuntimeFetch.test.ts
Original file line number Diff line number Diff line change
@@ -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<void>((resolve) => server?.close(() => resolve()) ?? resolve());
server = undefined;
});

async function serve(handler: Parameters<typeof createServer>[1]): Promise<string> {
server = createServer(handler);
await new Promise<void>((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/) });
});
});
13 changes: 11 additions & 2 deletions packages/ai-client/src/providers/localRuntimeFetch.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<Response> {
return fetch(url, { ...init, dispatcher: NO_TRANSPORT_LIMIT } as RequestInit);
return undiciFetch(url, {
...init,
dispatcher: NO_TRANSPORT_LIMIT,
} as Parameters<typeof undiciFetch>[1]) as unknown as Promise<Response>;
}
24 changes: 5 additions & 19 deletions packages/ai-client/src/providers/ollama.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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' });
Expand Down
5 changes: 5 additions & 0 deletions packages/ai-client/src/providers/ollamaRuntime.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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();
});
Expand Down
19 changes: 19 additions & 0 deletions scripts/installState.mjs
Original file line number Diff line number Diff line change
@@ -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);
}
32 changes: 32 additions & 0 deletions scripts/installState.test.mjs
Original file line number Diff line number Diff line change
@@ -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);
});
});
13 changes: 13 additions & 0 deletions scripts/start.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -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) => {
Expand Down Expand Up @@ -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
Expand Down
Loading