PIG/pig
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Make Piggy part of the product rather than a guest in it
CI / verify (push) Successful in 7m10s
Details
CI / publish (push) Has been skipped
Details

Browse Source 
Piggy arrived as a chat panel bolted onto a CRM and then grew a workspace
around it. The layout was already right — the audit found the approval card
to be the best-designed object in the repo, and the account page's empty
panels less finished than anything in the workspace. What was wrong was
vocabulary: nobody had written the small things down, so both halves kept
inventing them.

Piggy was drawn with five different marks — a pig in the dock, a sparkle in
the sidebar and again on the model picker, a speech bubble on the Ask
buttons, and a stock robot glyph on every assistant message, which is the
one people look at most. There is now one mark. The composer, which is the
first control in the product since sign-in lands on /piggy, was the only
un-adapted shadcn field left: 6px radius against a 12px Send button it sat
8px from. A stat tile had been reinvented six times at three numeral scales,
and the same uppercase micro-label existed in five variants, two of them one
tab apart in the same rail. There were 63 hand-written font sizes: not a
scale, sixty-three opinions.

Underneath that, the focus ring was invisible. The global rule used
ring-accent, which Tailwind deliberately aliases onto the hover tint, so the
ring measured 1.01:1 against the light canvas — no visible focus indicator
anywhere in the product, for any accent, in either theme. It is ring-brand
now and measures 17:1. The warning, positive and info tones were darkened
until each clears 4.5:1 on a card, on inset and on its own chip, and the
light canvas moved to 98% so a card lifts without leaning on its shadow.

The mobile work is the part worth reading. A landscape phone gave the
transcript 28% of the viewport and a keyboard-up phone 16%, against a 45%
floor — and the fixed tab bar painted over the composer, covering the safety
sentence and half the Send button, because two source comments asserted the
bar stood down on short viewports and it never had. Both fixed and measured
by hit-testing rather than by screenshot. The composer itself was 64px tall
for a blank second line nobody typed, because the auto-resize effect sizes
to scrollHeight and scrollHeight counts rows — a CSS height could not win
against an inline style, so the attribute was the honest lever.

Verified across both themes driven through the app's own control: no
horizontal overflow on 15 routes at four viewports, 672 stat values that fit,
297 labels at exactly 11px/500, Escape returning focus to its opener rather
than the body on every overlay, and a rejected write no longer reporting
"Succeeded" with a green check.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
claude
2026-08-14 18:22:15 -07:00
parent f0173440e4
commit 18d5f5bfc0
89 changed files with 8523 additions and 2447 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/piggy/test/agent-models.test.ts
+73 -8
View File
@@ -40,7 +40,14 @@ test('the default is in the catalogue and there is exactly one of it', () => {
assert.equal(defaults.length, 1);
assert.equal(defaults[0]?.id, piggyDefaultModelId());
assert.equal(piggyDefaultModelId(), 'nvidia/nemotron-3-nano-30b-a3b');
// The default is the SUPER, not the nano, and the reason is availability
// rather than quality. On 2026-08-14 `nvidia/nemotron-3-nano-30b-a3b` stopped
// answering on Prime Inference — the connection was accepted and no response
// headers ever arrived, three attempts at 45s each — while every other model
// in this catalogue answered in under two seconds on the same key in the same
// minute. The nano stays in the picker for anyone who wants it back.
assert.equal(piggyDefaultModelId(), 'nvidia/nemotron-3-super-120b-a12b');
assert.equal(isPiggyModelId('nvidia/nemotron-3-super-120b-a12b'), true);
assert.equal(isPiggyModelId('nvidia/nemotron-3-nano-30b-a3b'), true);
assert.equal(isPiggyModelId('nvidia/nemotron-9000'), false);
});
@@ -59,23 +66,81 @@ test('the picker can price and size every choice', () => {
}
});
test('the default is the cheapest thing on offer', () => {
// The panel is docked on every page, so the default is the price of a typo.
// If a costlier model ever becomes the default it should be a deliberate act
// that fails this test first.
/**
* This used to assert that the default was the cheapest thing on offer, and it
* was a good rule until the cheapest thing stopped answering. What actually
* protects the choice is not the ranking but the ceiling: the panel is docked on
* every page, so the default is the price of a typo, and the failure worth
* catching is somebody making a frontier model the default by accident. A
* deliberate move up the price list should pass; a slip to Opus should not.
*/
test('the default is a cheap model, even though it is no longer the cheapest', () => {
const catalogue = piggyModelCatalogue();
const cheapest = [...catalogue].sort((a, b) => a.costPerMTokIn - b.costPerMTokIn)[0];
const chosen = catalogue.find((option) => option.id === piggyDefaultModelId());
assert.ok(chosen && cheapest);
assert.equal(cheapest?.id, piggyDefaultModelId());
assert.notEqual(chosen.id, cheapest.id, 'the cheapest model answers again; revisit the default');
// Six times the price of the nano is still about $0.0017 a turn, or roughly
// 117,000 turns on a $200 credit. A dollar per million input tokens is an
// order of magnitude above that and two below every frontier model here.
assert.ok(chosen.costPerMTokIn <= 1, `${chosen.id} is too dear to be the default`);
const frontier = catalogue.filter((option) => option.costPerMTokIn >= 5);
assert.ok(frontier.length >= 2, 'the picker no longer offers a frontier option to contrast with');
for (const option of frontier) {
assert.notEqual(option.id, chosen.id, 'a frontier model became the default by accident');
}
});
/**
* The half of the reasoning trap that nobody would guess, pinned to whichever
* model is the default rather than to a name.
*
* `@earendil-works/pi-ai@0.84.1` turns a thinking level of `off` into no
* `reasoning_effort` field at all unless the model entry maps it, and the
* endpoint's own default then wins — 6,195 output tokens of reasoning and an
* empty answer. `agent-thinking.test.ts` pins the behaviour end to end; this
* pins the datum it depends on, which is the thing a new default would silently
* arrive without.
*/
test('the default carries a thinking map for the level Piggy is configured to run at', async () => {
process.env.DATABASE_URL ??= 'postgres://pig:pig@localhost:54330/pig';
process.env.PIGGY_INTERNAL_TOKEN ??= 'test-internal-token-for-piggy-000000';
process.env.PRIME_API_KEY ??= 'test-key-not-used-offline';
const { loadPiggyConfig } = await import('../src/config');
const level = loadPiggyConfig().PIGGY_AGENT_THINKING;
const registered = (
modelsJson.providers['prime-inference']?.models ?? []
) as { id: string; thinkingLevelMap?: Record<string, string> }[];
const chosen = registered.find((model) => model.id === piggyDefaultModelId());
assert.ok(chosen, 'the default is not registered with the provider');
assert.ok(
chosen.thinkingLevelMap,
`${chosen.id} is the default and has no thinkingLevelMap, so its reasoning is whatever the endpoint feels like`,
);
assert.equal(
typeof chosen.thinkingLevelMap[level],
'string',
`${chosen.id} does not map the configured thinking level '${level}'`,
);
});
test('the catalogue cannot be reordered by a caller', () => {
// It is serialised to the browser on every session; one sort() at a call
// site would reorder the picker for every other session in the process.
// site would reorder the picker for every other session in the process. The
// order is models.json's, which is no longer the same thing as "the default
// first" — asserting that conflated the two and broke when the default moved.
const registered = (modelsJson.providers['prime-inference']?.models ?? []).map(
(model) => model.id,
);
const first = piggyModelCatalogue();
first.reverse();
assert.equal(piggyModelCatalogue()[0]?.id, piggyDefaultModelId());
assert.deepEqual(
piggyModelCatalogue().map((option) => option.id),
registered,
);
});
test('the provider points at Prime Inference', () => {
apps/piggy/test/chat-retry.test.ts
+462
View File
@@ -0,0 +1,462 @@
/**
* What the chat server tells the user, and the ledger, about a retried turn.
*
* `inference-retry.test.ts` pins the retry itself against the real harness.
* This file pins the half of the same production failure that lived in PIG's
* own code, and it is the half that was doing the visible damage.
*
* Measured on 2026-08-14: the harness retries a rate-limited turn of its own
* accord and often succeeds, but `translateSessionEvent` latched
* `state.errorMessage` on the errored `turn_end` and never cleared it, so a turn
* that recovered and streamed a perfectly good answer was still closed as
* `inference_failed` with the 429 in `agent_runs.error`. The reader was told
* Piggy could not finish an answer they had just been given.
*
* Every session here is a double, for the same reason the stall guard's are: an
* endpoint cannot be asked to rate limit on demand, and the point of these tests
* is the server's reading of the events, not the transport underneath them.
*/
import assert from 'node:assert/strict';
import type { AddressInfo } from 'node:net';
import test from 'node:test';
import type { AgentSession, AgentSessionEvent, ToolDefinition } from '@earendil-works/pi-coding-agent';
import type { PiggyChatEvent, PiggyModelOption } from '@pig/core';
import type { Database } from '@pig/db';
import type { PiggySession } from '../src/agent/session';
import { startPiggyChatServer, type PiggyChatServerOptions } from '../src/chat-server';
import type { PiggyStallLimits } from '../src/config';
const TOKEN = 'test-internal-token-for-piggy-000000';
const MODELS: PiggyModelOption[] = [
{
id: 'nvidia/nemotron-3-super-120b-a12b',
label: 'Nemotron 3 Super',
costPerMTokIn: 0.3,
costPerMTokOut: 0.9,
contextWindow: 131_072,
reasoning: true,
isDefault: true,
},
];
/** The body Prime Inference really sends, verbatim from the production log. */
const RATE_LIMIT_ERROR =
'429: {"message":"Rate limit reached. Please retry shortly.","type":"rate_limit_exceeded","code":"rate_limited"}';
interface RecordedRun {
values: Record<string, unknown>;
closed?: Record<string, unknown>;
}
function fakeDatabase(runs: RecordedRun[]): Database {
return {
insert: () => ({
values: (values: Record<string, unknown>) => ({
returning: async () => {
runs.push({ values });
return [{ id: `run-${runs.length}` }];
},
}),
}),
update: () => ({
set: (closed: Record<string, unknown>) => ({
where: async () => {
const run = runs.at(-1);
if (run) run.closed = closed;
},
}),
}),
select: () => ({ from: () => ({ where: async () => [{ spent: '0' }] }) }),
} as unknown as Database;
}
type TurnScript = (
tools: readonly ToolDefinition[],
emit: (event: AgentSessionEvent) => void,
signal: AbortSignal,
) => Promise<void>;
interface SessionSpy {
aborted: number;
}
function sessions(script: TurnScript, watched: SessionSpy) {
return async (options: { tools: readonly ToolDefinition[]; modelId?: string }): Promise<PiggySession> => {
const listeners = new Set<(event: AgentSessionEvent) => void>();
const aborted = new AbortController();
const session = {
subscribe(listener: (event: AgentSessionEvent) => void) {
listeners.add(listener);
return () => listeners.delete(listener);
},
async prompt() {
await script(
options.tools,
(event) => {
for (const listener of [...listeners]) listener(event);
},
aborted.signal,
);
},
async abort() {
watched.aborted += 1;
aborted.abort();
},
dispose() {},
} as unknown as AgentSession;
return {
session,
modelId: options.modelId ?? MODELS[0]!.id,
systemPrompt: 'You are Piggy.',
dispose: () => aborted.abort(),
} satisfies PiggySession;
};
}
function textDelta(delta: string): AgentSessionEvent {
return {
type: 'message_update',
message: { role: 'assistant' },
assistantMessageEvent: { type: 'text_delta', contentIndex: 0, delta },
} as unknown as AgentSessionEvent;
}
function turnEnd(input: number, output: number, stopReason = 'stop'): AgentSessionEvent {
return {
type: 'turn_end',
message: { role: 'assistant', usage: { input, output }, stopReason },
toolResults: [],
} as unknown as AgentSessionEvent;
}
/** A model call the endpoint refused. This is what a 429 looks like from here. */
function failedTurn(errorMessage: string): AgentSessionEvent {
return {
type: 'turn_end',
message: { role: 'assistant', usage: { input: 0, output: 0 }, stopReason: 'error', errorMessage },
toolResults: [],
} as unknown as AgentSessionEvent;
}
/** The harness announcing that it is about to restart the turn. */
function retryStart(errorMessage: string, attempt = 1): AgentSessionEvent {
return {
type: 'auto_retry_start',
attempt,
maxAttempts: 1,
delayMs: 1_500,
errorMessage,
} as unknown as AgentSessionEvent;
}
/** Silence, until somebody tells the turn to stop. A harness that unwinds. */
const untilAborted: TurnScript = (_tools, _emit, signal) =>
new Promise<void>((resolve) => {
if (signal.aborted) {
resolve();
return;
}
signal.addEventListener('abort', () => resolve(), { once: true });
});
function stallLimits(overrides: Partial<PiggyStallLimits> = {}): PiggyStallLimits {
return { firstProgressMs: 5_000, idleMs: 5_000, ...overrides };
}
async function startForTest(
t: { after: (fn: () => void) => void },
runs: RecordedRun[],
options: Partial<PiggyChatServerOptions>,
): Promise<string> {
const server = startPiggyChatServer(fakeDatabase(runs), {
port: 0,
internalToken: TOKEN,
models: MODELS,
createReadTools: () => [],
createWriteTools: () => [],
limits: { maxModelCalls: 8, maxTurnTokens: 40_000, dailyLimitCents: 0 },
stallLimits: stallLimits(),
...options,
});
t.after(() => server.close());
await new Promise((resolve) => server.once('listening', resolve));
const { port } = server.address() as AddressInfo;
return `http://127.0.0.1:${port}`;
}
const PRINCIPAL = {
userId: '20000000-0000-4000-8000-000000000001',
email: 'ada@primeintellect.example',
name: 'Ada',
isPlatformAdmin: false,
teams: [{ team: 'supply', role: 'lead' }],
via: 'jwt',
scopes: ['read', 'write'],
};
const authorised = { authorization: `Bearer ${TOKEN}`, 'content-type': 'application/json' };
function chatBody(overrides: Record<string, unknown> = {}): string {
return JSON.stringify({
principal: PRINCIPAL,
message: 'What is idle costing us?',
mode: 'read_only',
conversationId: 'conv-retry',
...overrides,
});
}
async function turnFrames(base: string, body = chatBody()): Promise<PiggyChatEvent[]> {
const response = await fetch(`${base}/internal/chat`, { method: 'POST', headers: authorised, body });
return (await response.text())
.trim()
.split('\n')
.filter((line) => line.length > 0)
.map((line) => JSON.parse(line) as PiggyChatEvent);
}
function errorFrame(frames: PiggyChatEvent[]): { message: string; code?: string } | null {
const frame = frames.at(-1);
return frame?.type === 'error'
? { message: frame.message, ...(frame.code ? { code: frame.code } : {}) }
: null;
}
function answerText(frames: PiggyChatEvent[]): string {
return frames
.filter((frame): frame is Extract<PiggyChatEvent, { type: 'content_delta' }> => frame.type === 'content_delta')
.map((frame) => frame.delta)
.join('');
}
function inference(closed: Record<string, unknown> | undefined): Record<string, unknown> | undefined {
return (closed?.result as { inference?: Record<string, unknown> } | undefined)?.inference;
}
// ------------------------------------------------- the turn that recovered anyway
test('a turn the harness retried and finished is reported as finished', async (t) => {
const runs: RecordedRun[] = [];
const watched: SessionSpy = { aborted: 0 };
const base = await startForTest(t, runs, {
createSession: sessions(async (_tools, emit) => {
// The 429 arrives before a byte of the answer, which is the ordinary
// shape of one: the endpoint refuses the request rather than dropping a
// response half way through.
emit(failedTurn(RATE_LIMIT_ERROR));
emit(retryStart(RATE_LIMIT_ERROR));
emit(textDelta('Idle is $12,000.'));
emit(turnEnd(1_240, 180));
}, watched),
});
const frames = await turnFrames(base);
// The whole of the visible bug: this used to end in an error frame with the
// 429 in the ledger, after the reader had already been given the answer.
assert.deepEqual(
frames.map((frame) => frame.type),
['meta', 'content_delta', 'done'],
);
assert.equal(answerText(frames), 'Idle is $12,000.');
assert.equal(watched.aborted, 0, 'a turn that was recovering was stopped');
const closed = runs[0]?.closed;
assert.equal(closed?.status, 'succeeded');
assert.equal(closed?.error, null);
// And an operator can still see that it cost two goes, which is the trend
// they are watching even when every turn eventually answers.
assert.equal(inference(closed)?.attempts, 2);
assert.match(String(inference(closed)?.retryReason), /Rate limit reached/);
});
test('a healthy turn records one attempt rather than none', async (t) => {
const runs: RecordedRun[] = [];
const base = await startForTest(t, runs, {
createSession: sessions(async (_tools, emit) => {
emit(textDelta('Idle is $12,000.'));
emit(turnEnd(1_240, 180));
}, { aborted: 0 }),
});
const frames = await turnFrames(base);
assert.equal(frames.at(-1)?.type, 'done');
// Written on every turn, not only the failed ones: a day where every turn
// needed two attempts and succeeded must not look like a day where none did.
assert.equal(inference(runs[0]?.closed)?.attempts, 1);
assert.equal(inference(runs[0]?.closed)?.retryReason, undefined);
});
// --------------------------------------------------- when the retries run out
test('an exhausted rate limit is its own code, and says what to do about it', async (t) => {
const runs: RecordedRun[] = [];
const base = await startForTest(t, runs, {
createSession: sessions(async (_tools, emit) => {
emit(failedTurn(RATE_LIMIT_ERROR));
emit(retryStart(RATE_LIMIT_ERROR));
emit(failedTurn(RATE_LIMIT_ERROR));
}, { aborted: 0 }),
});
const frames = await turnFrames(base);
// Distinct from `inference_failed`, because it wants a different response:
// waiting ten seconds genuinely fixes it, and it is not worth a pager.
assert.equal(errorFrame(frames)?.code, 'inference_rate_limited');
assert.match(String(errorFrame(frames)?.message), /rate limiting us/);
assert.match(String(errorFrame(frames)?.message), /2 times/);
assert.match(String(errorFrame(frames)?.message), /ask again/i);
assert.equal(answerText(frames), '');
const closed = runs[0]?.closed;
assert.equal(closed?.status, 'failed');
// The ledger keeps the upstream body; the browser is never shown it.
assert.match(String(closed?.error), /rate_limit_exceeded/);
assert.match(String(closed?.error), /2 attempts/);
assert.equal(inference(closed)?.attempts, 2);
});
test('a fault that is not a rate limit keeps the generic code', async (t) => {
const runs: RecordedRun[] = [];
const base = await startForTest(t, runs, {
createSession: sessions(async (_tools, emit) => {
emit(failedTurn('502: {"message":"upstream connect error"}'));
}, { aborted: 0 }),
});
const frames = await turnFrames(base);
// Somebody should look at this one, so it must not wear the name of the fault
// that fixes itself.
assert.equal(errorFrame(frames)?.code, 'inference_failed');
assert.equal(errorFrame(frames)?.message, 'Piggy could not finish this answer.');
assert.equal(inference(runs[0]?.closed)?.attempts, 1);
});
// ------------------------------------------------- what a retry may never replay
test('a retry that would repeat a delivered answer is refused', async (t) => {
const runs: RecordedRun[] = [];
const watched: SessionSpy = { aborted: 0 };
const base = await startForTest(t, runs, {
createSession: sessions(async (tools, emit, signal) => {
// Measured against a stubbed endpoint: the harness's session-level retry
// discards the errored assistant message and generates a replacement, so
// a turn that had streamed "Idle is " came back as
// "Idle is Idle is $12,000." in the transcript.
emit(textDelta('Idle is '));
emit(failedTurn(RATE_LIMIT_ERROR));
emit(retryStart(RATE_LIMIT_ERROR));
// And this script does not stop when it is told to, which is the nastier
// shape of the same fault and the one the stall guard already assumes: a
// harness that ignores the abort would stream the replacement answer over
// the top of the half the reader already has. Neither the abort nor the
// suppression is sufficient on its own.
await untilAborted(tools, emit, signal);
emit(textDelta('Idle is $12,000.'));
emit(turnEnd(1_240, 180));
}, watched),
});
const frames = await turnFrames(base);
assert.equal(answerText(frames), 'Idle is ', 'the reader was shown the answer twice');
assert.equal(watched.aborted, 1, 'the replay was allowed to proceed');
assert.equal(errorFrame(frames)?.code, 'inference_rate_limited');
assert.match(String(errorFrame(frames)?.message), /incomplete/);
assert.match(String(errorFrame(frames)?.message), /already been shown/);
assert.equal(
frames.some((frame) => frame.type === 'done'),
false,
'an incomplete answer must not also report itself finished',
);
const closed = runs[0]?.closed;
assert.equal(closed?.status, 'failed');
assert.equal(closed?.summary, 'Idle is');
assert.match(String(closed?.error), /retry refused/);
assert.equal(inference(closed)?.attempts, 2);
});
test('a retry before anything has been delivered is left alone', async (t) => {
const runs: RecordedRun[] = [];
const watched: SessionSpy = { aborted: 0 };
const base = await startForTest(t, runs, {
createSession: sessions(async (_tools, emit) => {
// A tool ran, so the turn is not untouched — but nothing has reached the
// reader's transcript, so there is nothing to say twice. Stopping here
// would throw away a recoverable turn for no gain.
emit({
type: 'tool_execution_start',
toolCallId: 'call_1',
toolName: 'pig_get_idle_capacity',
args: {},
} as unknown as AgentSessionEvent);
emit(failedTurn(RATE_LIMIT_ERROR));
emit(retryStart(RATE_LIMIT_ERROR));
emit(textDelta('Idle is $12,000.'));
emit(turnEnd(1_240, 180));
}, watched),
});
const frames = await turnFrames(base);
assert.equal(watched.aborted, 0, 'a safe retry was refused');
assert.equal(frames.at(-1)?.type, 'done');
assert.equal(answerText(frames), 'Idle is $12,000.');
assert.equal(runs[0]?.closed?.status, 'succeeded');
});
// -------------------------------------------- the guards that outrank the retry
test('the stall watchdog outranks a pending retry', async (t) => {
const runs: RecordedRun[] = [];
const watched: SessionSpy = { aborted: 0 };
const base = await startForTest(t, runs, {
stallLimits: stallLimits({ firstProgressMs: 30_000, idleMs: 120 }),
createSession: sessions(async (tools, emit, signal) => {
emit(failedTurn(RATE_LIMIT_ERROR));
emit(retryStart(RATE_LIMIT_ERROR));
// The retry was announced and then nothing ever happened, which is the
// shape of a backoff into an endpoint that has stopped answering
// altogether. A retry loop that could outlive the watchdog would hang the
// browser exactly the way the missing deadline used to.
await untilAborted(tools, emit, signal);
}, watched),
});
const frames = await turnFrames(base);
assert.equal(errorFrame(frames)?.code, 'inference_stalled');
assert.equal(watched.aborted, 1);
const closed = runs[0]?.closed;
assert.match(String(closed?.error), /idle deadline/);
// The attempt count is still recorded: the turn really did try twice before
// the silence, and that is what an operator is counting.
assert.equal(inference(closed)?.attempts, 2);
});
test('the turn ceiling outranks a pending retry', async (t) => {
const runs: RecordedRun[] = [];
const watched: SessionSpy = { aborted: 0 };
const base = await startForTest(t, runs, {
limits: { maxModelCalls: 2, maxTurnTokens: 40_000, dailyLimitCents: 0 },
createSession: sessions(async (tools, emit, signal) => {
emit(turnEnd(1_000, 100, 'toolUse'));
emit(failedTurn(RATE_LIMIT_ERROR));
emit(retryStart(RATE_LIMIT_ERROR));
await untilAborted(tools, emit, signal);
}, watched),
});
const frames = await turnFrames(base);
// A retry that resurrected a turn already stopped for cost would spend money
// the ceiling exists to refuse.
assert.equal(errorFrame(frames)?.code, 'turn_limit_exceeded');
assert.equal(runs[0]?.closed?.status, 'aborted');
});
apps/piggy/test/chat-server.test.ts
+6 -1
View File
@@ -657,7 +657,10 @@ test('a model that stops on its own error does not report a finished answer', as
assert.ok(!frames.some((frame) => frame.type === 'done'));
// The upstream body is not ours to relay to the browser, but it belongs in
// the ledger, where an operator can read it.
assert.equal(runs[0]?.closed?.error, 'upstream returned 502');
// The attempt count rides along with it: one go, which is the healthy
// reading and the one an operator needs in order to notice the days when it
// is not one.
assert.equal(runs[0]?.closed?.error, 'upstream returned 502 (1 attempt)');
});
test('a turn that ends badly still bills what it actually spent', async (t) => {
@@ -925,6 +928,7 @@ test('a proposed write waits for the user, then applies once and only once', asy
approvalsRequested: 1,
approvalsApplied: 1,
modelCalls: 1,
inference: { attempts: 1 },
});
});
@@ -998,6 +1002,7 @@ test('a declined write is reported to the model as declined', async (t) => {
approvalsRequested: 1,
approvalsApplied: 0,
modelCalls: 1,
inference: { attempts: 1 },
});
});
apps/piggy/test/config.test.ts
+38 -1
View File
@@ -1,6 +1,6 @@
import assert from 'node:assert/strict';
import test from 'node:test';
import { loadPiggyConfig, loadPiggyTurnLimits } from '../src/config';
import { loadPiggyConfig, loadPiggyStallLimits, loadPiggyTurnLimits } from '../src/config';
const minimum = {
DATABASE_URL: 'postgres://pig:pig@localhost:54330/pig',
@@ -65,6 +65,43 @@ test('the ceilings can be read without the rest of the environment', () => {
);
});
test('a turn has two deadlines for silence, and they are not one flat deadline', () => {
const config = loadPiggyConfig(minimum);
assert.equal(config.PIGGY_CHAT_FIRST_PROGRESS_TIMEOUT_MS, 60_000);
assert.equal(config.PIGGY_CHAT_IDLE_TIMEOUT_MS, 45_000);
// Read on their own too: the chat server is handed a socket and a token.
assert.deepEqual(loadPiggyStallLimits({}), { firstProgressMs: 60_000, idleMs: 45_000 });
assert.deepEqual(
loadPiggyStallLimits({
PIGGY_CHAT_FIRST_PROGRESS_TIMEOUT_MS: '1500',
PIGGY_CHAT_IDLE_TIMEOUT_MS: '900',
}),
{ firstProgressMs: 1_500, idleMs: 900 },
);
// The idle window is the shorter of the two on purpose. Getting started
// covers connecting, the endpoint's queue and a slow model's first token;
// once a turn is under way the gaps are milliseconds, so a long silence
// mid-answer is a dead socket rather than a thoughtful one. Neither bounds
// the turn's total duration, which is the whole design: the idle clock
// restarts on every event.
assert.ok(
loadPiggyStallLimits({}).idleMs < loadPiggyStallLimits({}).firstProgressMs,
'the idle window should not need to be as generous as getting started',
);
// A deadline of zero would stall every turn before it began, so it is a
// configuration error rather than a very impatient deployment.
assert.throws(
() => loadPiggyStallLimits({ PIGGY_CHAT_IDLE_TIMEOUT_MS: '0' }),
/PIGGY_CHAT_IDLE_TIMEOUT_MS/,
);
assert.throws(
() => loadPiggyStallLimits({ PIGGY_CHAT_FIRST_PROGRESS_TIMEOUT_MS: 'patience' }),
/PIGGY_CHAT_FIRST_PROGRESS_TIMEOUT_MS/,
);
});
test('reasoning stays off by default', () => {
// Reasoning tokens are billed like any other and nemotron-nano's are
// verbose. The knob exists for debugging, not for the default deployment.
apps/piggy/test/inference-retry.test.ts
+493
View File
@@ -0,0 +1,493 @@
/**
* What Piggy does when Prime Inference says "please retry shortly".
*
* The failure this file pins was measured on production on 2026-08-14, roughly
* every other turn:
*
* [piggy] chat turn ended in an inference error: 429:
* {"message":"Rate limit reached. Please retry shortly.",
* "type":"rate_limit_exceeded","code":"rate_limited"}
*
* A `curl` a second later succeeded, so these were transient bursts and the
* endpoint was telling us what to do about them. Nothing did.
*
* The endpoint cannot be asked to rate limit on demand, and a test that waited
* for it to happen would be untrustworthy in exactly the conditions it exists
* for, so every upstream here is a stub installed over `globalThis.fetch`. That
* is a real seam and not a convenience: the OpenAI client the harness builds
* resolves its fetch through `getDefaultFetch()` at construction, and it
* constructs one per model call (openai@6.26.0 internal/shims.js:9-14), so a
* stub installed before `prompt()` is the transport the harness genuinely uses.
* Everything below therefore runs the real `createPiggySession`, the real
* harness and the real OpenAI SDK against a fake endpoint — the retry is the
* only thing under test, and none of it is mocked.
*/
import assert from 'node:assert/strict';
import { mkdtempSync, rmSync, writeFileSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import test, { after, before } from 'node:test';
import {
createAgentSession,
defineTool,
ModelRuntime,
SessionManager,
SettingsManager,
type ToolDefinition,
} from '@earendil-works/pi-coding-agent';
import { Type } from 'typebox';
import { piggyDefaultModelId, piggyModelsJsonText, PIGGY_PROVIDER_ID } from '../src/agent/models';
import {
piggyAgentSettings,
PIGGY_INFERENCE_RETRY,
type PiggyInferenceRetryPolicy,
} from '../src/agent/session';
const agentDir = mkdtempSync(join(tmpdir(), 'piggy-retry-test-'));
const realFetch = globalThis.fetch;
before(() => {
process.env.DATABASE_URL = 'postgres://pig:pig@localhost:54330/pig';
process.env.PIGGY_INTERNAL_TOKEN = 'test-internal-token-for-piggy-000000';
// Deliberately fake. Nothing below leaves the process, and a test that needs
// a live key is a test that fails in CI.
process.env.PRIME_API_KEY = 'test-key-not-used-offline';
process.env.PIGGY_AGENT_DIR = agentDir;
});
after(() => {
globalThis.fetch = realFetch;
rmSync(agentDir, { recursive: true, force: true });
});
// ------------------------------------------------------------- the fake endpoint
const MODEL = piggyDefaultModelId();
function chunk(delta: unknown, finish: string | null, usage?: unknown): string {
return JSON.stringify({
id: 'chatcmpl-test',
object: 'chat.completion.chunk',
created: 1,
model: MODEL,
choices: [{ index: 0, delta, finish_reason: finish }],
...(usage ? { usage } : {}),
});
}
function eventStream(chunks: string[], terminated = true): Response {
const body = chunks.map((line) => `data: ${line}\n\n`).join('') + (terminated ? 'data: [DONE]\n\n' : '');
return new Response(body, { status: 200, headers: { 'content-type': 'text/event-stream' } });
}
/** A complete, ordinary answer. */
function answers(text = 'Idle is $12,000.'): Response {
return eventStream([
chunk({ role: 'assistant', content: text }, null),
chunk({}, 'stop', { prompt_tokens: 100, completion_tokens: 8, total_tokens: 108 }),
]);
}
/** One tool call and nothing else, which is how a tool-using turn starts. */
function callsTool(name: string): Response {
return eventStream([
chunk(
{
role: 'assistant',
tool_calls: [
{ index: 0, id: 'call_1', type: 'function', function: { name, arguments: '{}' } },
],
},
null,
),
chunk({}, 'tool_calls', { prompt_tokens: 100, completion_tokens: 8, total_tokens: 108 }),
]);
}
/** The body Prime Inference really sends, verbatim from the production log. */
function rateLimited(retryAfterSeconds?: number): Response {
return new Response(
JSON.stringify({
message: 'Rate limit reached. Please retry shortly.',
type: 'rate_limit_exceeded',
code: 'rate_limited',
}),
{
status: 429,
headers: {
'content-type': 'application/json',
...(retryAfterSeconds === undefined ? {} : { 'retry-after': String(retryAfterSeconds) }),
},
},
);
}
function failsWith(status: number, message: string): Response {
return new Response(JSON.stringify({ message }), {
status,
headers: { 'content-type': 'application/json' },
});
}
interface Upstream {
/** When each request arrived, in milliseconds since the stub was installed. */
readonly at: number[];
readonly count: number;
}
/** Installs a stub over the global fetch and records every request it sees. */
function upstream(reply: (attempt: number) => Response | Promise<Response>): Upstream {
const at: number[] = [];
const started = Date.now();
globalThis.fetch = (async (_input: unknown, init?: RequestInit) => {
at.push(Date.now() - started);
const response = await reply(at.length);
// The caller's signal is honoured so that a stub which never answers can
// still be cancelled by a deadline, which is the whole point of one.
if (init?.signal?.aborted) throw init.signal.reason;
return response;
}) as typeof fetch;
return {
at,
get count() {
return at.length;
},
};
}
/** A stub that never answers, and unblocks only when the request is abandoned. */
function silence(): Upstream {
const at: number[] = [];
const started = Date.now();
globalThis.fetch = ((_input: unknown, init?: RequestInit) => {
at.push(Date.now() - started);
return new Promise<Response>((_resolve, reject) => {
const signal = init?.signal;
if (!signal) return;
if (signal.aborted) {
reject(signal.reason);
return;
}
signal.addEventListener('abort', () => reject(signal.reason), { once: true });
});
}) as typeof fetch;
return {
at,
get count() {
return at.length;
},
};
}
// ------------------------------------------------------------------ the fixtures
function countingTool(name: string, runs: { count: number }): ToolDefinition {
return defineTool({
name,
label: name,
description: `Test double for ${name}.`,
promptSnippet: `${name}: test double.`,
parameters: Type.Object({}),
async execute() {
runs.count += 1;
return { content: [{ type: 'text' as const, text: '{"ok":true}' }], details: { tool: name } };
},
});
}
interface TurnResult {
/** Everything the reader would have been shown, concatenated. */
text: string;
/** How the last model call ended, as the harness reports it. */
errorMessage?: string;
stopReason?: string;
/** Retries the harness announced, which are the ones that replay work. */
announcedRetries: number;
elapsedMs: number;
}
/** One real Piggy turn, driven through the real `createPiggySession`. */
async function drive(tools: ToolDefinition[], message = 'What is idle costing us?'): Promise<TurnResult> {
const { createPiggySession } = await import('../src/agent/session');
const piggy = await createPiggySession({ mode: 'read_only', tools });
const result: TurnResult = { text: '', announcedRetries: 0, elapsedMs: 0 };
const started = Date.now();
const unsubscribe = piggy.session.subscribe((event) => {
if (event.type === 'message_update' && event.assistantMessageEvent.type === 'text_delta') {
result.text += event.assistantMessageEvent.delta;
}
if (event.type === 'auto_retry_start') result.announcedRetries += 1;
if (event.type === 'turn_end') {
const assistant = event.message as { stopReason?: string; errorMessage?: string };
result.stopReason = assistant.stopReason;
result.errorMessage = assistant.errorMessage;
}
});
try {
await piggy.session.prompt(message);
} finally {
unsubscribe();
result.elapsedMs = Date.now() - started;
piggy.dispose();
}
return result;
}
// ------------------------------------------------------- the measured production bug
test('a 429 that clears on the next attempt is answered rather than reported', async () => {
// The bug, in one test. Before the policy existed the harness made exactly
// one attempt per model call — `retryProviderRequest` defaults `maxRetries`
// to 0 and the settings supplied none — so this turn ended as
// `inference_failed` with no answer at all.
const runs = { count: 0 };
const endpoint = upstream((attempt) => (attempt === 1 ? rateLimited() : answers()));
const turn = await drive([countingTool('pig_get_workspace_summary', runs)]);
assert.equal(endpoint.count, 2, 'the refusal was not retried');
assert.equal(turn.errorMessage, undefined);
assert.equal(turn.stopReason, 'stop');
assert.equal(turn.text, 'Idle is $12,000.');
});
test('a retried turn shows the reader one answer, not two', async () => {
// The constraint that makes the seam matter. The retry happens where the
// response has not begun, so there is nothing to replay — no delta is emitted
// twice, and the harness never has to announce a retry at all.
const runs = { count: 0 };
upstream((attempt) => (attempt <= 2 ? rateLimited() : answers('Idle is $12,000.')));
const turn = await drive([countingTool('pig_get_workspace_summary', runs)]);
assert.equal(turn.text, 'Idle is $12,000.');
assert.equal(
turn.text.indexOf('Idle is'),
turn.text.lastIndexOf('Idle is'),
'the answer was streamed to the reader twice',
);
assert.equal(turn.announcedRetries, 0, 'the turn was restarted when it did not need to be');
});
test('a retry never re-runs a tool that has already run', async () => {
// The expensive property. `pig_log_activity` writes a row; a retry that
// re-executed it would write it twice and no diff card would be shown for the
// second one. The tool is called on the first model call, the SECOND model
// call is the one that is rate limited, and the tool must not move.
const runs = { count: 0 };
const endpoint = upstream((attempt) => {
if (attempt === 1) return callsTool('pig_log_activity');
if (attempt === 2) return rateLimited();
return answers('Logged.');
});
const turn = await drive([countingTool('pig_log_activity', runs)], 'Log a call on Northwind.');
assert.equal(endpoint.count, 3);
assert.equal(runs.count, 1, 'the tool ran again on the retry');
assert.equal(turn.text, 'Logged.');
assert.equal(turn.errorMessage, undefined);
});
test('Retry-After is honoured when the endpoint sends one', async () => {
const runs = { count: 0 };
const endpoint = upstream((attempt) => (attempt === 1 ? rateLimited(1) : answers()));
const turn = await drive([countingTool('pig_get_workspace_summary', runs)]);
assert.equal(endpoint.count, 2);
assert.equal(turn.errorMessage, undefined);
// A second is far longer than the jittered backoff this attempt would have
// chosen for itself (500ms, minus up to a quarter), so waiting it out is only
// possible if the header was read.
const waited = endpoint.at[1]! - endpoint.at[0]!;
assert.ok(waited >= 900, `waited ${waited}ms, so Retry-After was ignored`);
assert.ok(waited < 3_000, `waited ${waited}ms, which is longer than was asked for`);
});
test('a refusal with no Retry-After still backs off, and briefly', async () => {
// Jitter matters more than the curve: without it every open chat that hit the
// same limit retries in lockstep and reproduces the limit that caused it.
const runs = { count: 0 };
const endpoint = upstream((attempt) => (attempt === 1 ? rateLimited() : answers()));
await drive([countingTool('pig_get_workspace_summary', runs)]);
const waited = endpoint.at[1]! - endpoint.at[0]!;
assert.ok(waited > 0, 'the retry was fired immediately, which reproduces the limit');
assert.ok(waited < 2_000, `waited ${waited}ms without being asked to`);
});
test('a rate limit that never clears is reported, and inside a bearable wait', async () => {
const runs = { count: 0 };
const endpoint = upstream(() => rateLimited());
const turn = await drive([countingTool('pig_get_workspace_summary', runs)]);
assert.match(String(turn.errorMessage), /429/);
assert.equal(turn.stopReason, 'error');
assert.equal(turn.text, '');
// Every attempt the policy buys was spent: the request-level budget, twice
// over, because the turn-level budget allows one restart of a turn that got
// nothing from the endpoint.
assert.equal(endpoint.count, PIGGY_INFERENCE_RETRY.attempts * PIGGY_INFERENCE_RETRY.streamAttempts);
// Nobody may be left staring at a docked panel for a minute to be told no.
assert.ok(turn.elapsedMs < 30_000, `the failure took ${turn.elapsedMs}ms to arrive`);
});
test('a 500 is retried and a 400 is not', async () => {
const runs = { count: 0 };
const serverError = upstream((attempt) =>
attempt === 1 ? failsWith(500, 'internal error') : answers(),
);
const recovered = await drive([countingTool('pig_get_workspace_summary', runs)]);
assert.equal(serverError.count, 2, 'a 5xx is transient and should have been retried');
assert.equal(recovered.errorMessage, undefined);
// A 4xx that is not 429 will fail identically however often it is retried,
// and each attempt costs a round trip and a place in the queue.
const badRequest = upstream(() => failsWith(400, 'unknown parameter'));
const refused = await drive([countingTool('pig_get_workspace_summary', runs)]);
assert.equal(badRequest.count, 1, 'a 400 was retried, which can only ever fail again');
assert.equal(refused.stopReason, 'error');
assert.match(String(refused.errorMessage), /400/);
});
test('a caller who hangs up wins over the retry', async () => {
// A retry loop that resurrects an abandoned turn is worse than the bug: it
// spends credit generating an answer nobody will read, and it does it while
// the reader has already gone.
const { createPiggySession } = await import('../src/agent/session');
const endpoint = upstream(() => rateLimited());
const runs = { count: 0 };
const piggy = await createPiggySession({
mode: 'read_only',
tools: [countingTool('pig_get_workspace_summary', runs)],
});
try {
const prompt = piggy.session.prompt('What is idle costing us?');
// Long enough for the first attempt to have been refused and the second to
// be sleeping on its backoff, which is where an abort has to be honoured.
await new Promise((resolve) => setTimeout(resolve, 250));
const seenBeforeAbort = endpoint.count;
await piggy.session.abort();
await prompt;
await new Promise((resolve) => setTimeout(resolve, 400));
assert.ok(seenBeforeAbort >= 1, 'the turn had not started, so nothing was proved');
assert.equal(
endpoint.count,
seenBeforeAbort,
'the retry carried on asking after the caller had gone',
);
} finally {
piggy.dispose();
}
});
// ------------------------------------------ the deadline the model entry cannot carry
/**
* A bare harness session, wired the way `createPiggySession` wires one but with
* a policy of the test's choosing.
*
* Built by hand rather than through `createPiggySession` because the shipped
* deadline is twenty seconds and a test may not take twenty seconds to prove
* one. What it proves is a fact about the INSTALLED package rather than about
* PIG's wiring — that `retry.provider.timeoutMs` and `retry.provider.maxRetries`
* are read and acted on — and the wiring itself is proved by every test above,
* all of which go through the real `createPiggySession`.
*/
async function bareSession(policy: PiggyInferenceRetryPolicy, tools: ToolDefinition[]) {
const modelsPath = join(agentDir, 'models-for-timeout-test.json');
writeFileSync(modelsPath, piggyModelsJsonText(), { mode: 0o600 });
const modelRuntime = await ModelRuntime.create({ modelsPath, allowModelNetwork: false });
await modelRuntime.setRuntimeApiKey(PIGGY_PROVIDER_ID, 'test-key-not-used-offline');
const model = modelRuntime.getModel(PIGGY_PROVIDER_ID, MODEL);
assert.ok(model, 'the default model should be registered');
const { session } = await createAgentSession({
agentDir,
cwd: agentDir,
modelRuntime,
model,
settingsManager: SettingsManager.inMemory(piggyAgentSettings(policy)),
thinkingLevel: 'off',
noTools: 'all',
tools: tools.map((tool) => tool.name),
customTools: tools,
sessionManager: SessionManager.inMemory(),
});
return session;
}
test('the request deadline is read from the settings the runtime is built with', async () => {
// The stall watchdog is the outer guard and it stays; this is the deadline
// underneath it, on one HTTP request rather than on the turn. Without it a
// hung fetch has only the harness's own five-minute idle default.
const endpoint = silence();
const runs = { count: 0 };
const session = await bareSession(
{ ...PIGGY_INFERENCE_RETRY, headersTimeoutMs: 150, streamAttempts: 1 },
[countingTool('pig_get_workspace_summary', runs)],
);
let errorMessage: string | undefined;
session.subscribe((event) => {
if (event.type === 'turn_end') {
errorMessage = (event.message as { errorMessage?: string }).errorMessage;
}
});
const started = Date.now();
await session.prompt('What is idle costing us?');
const elapsed = Date.now() - started;
// Every attempt was abandoned at its own deadline and the next one started,
// which is only possible if BOTH fields reached the transport.
assert.equal(endpoint.count, PIGGY_INFERENCE_RETRY.attempts);
assert.ok(elapsed >= 150, `gave up after ${elapsed}ms, before the deadline it was given`);
assert.ok(elapsed < 10_000, `took ${elapsed}ms, so the deadline was not honoured`);
assert.ok(errorMessage, 'a hung request ended as a success');
await session.abort();
session.dispose();
});
test('the settings the harness reads are exactly the policy PIG declares', () => {
// Read back through the installed `SettingsManager` rather than compared to
// the object we wrote, because the field names and their nesting are the
// whole risk: a policy under a key the harness has never heard of parses,
// loads and does nothing, and there is no error anywhere to say so.
const manager = SettingsManager.inMemory(piggyAgentSettings());
const provider = manager.getProviderRetrySettings();
const turn = manager.getRetrySettings();
assert.equal(provider.timeoutMs, PIGGY_INFERENCE_RETRY.headersTimeoutMs);
assert.equal(provider.maxRetries, PIGGY_INFERENCE_RETRY.attempts - 1);
assert.equal(provider.maxRetryDelayMs, PIGGY_INFERENCE_RETRY.maxRetryDelayMs);
assert.equal(turn.enabled, true);
assert.equal(turn.maxRetries, PIGGY_INFERENCE_RETRY.streamAttempts - 1);
assert.equal(turn.baseDelayMs, PIGGY_INFERENCE_RETRY.streamBackoffMs);
// The default this replaces, and the reason the bug existed: the harness
// ships no provider retry budget at all, and `retryProviderRequest` reads a
// missing budget as zero.
assert.equal(SettingsManager.inMemory().getProviderRetrySettings().maxRetries, undefined);
});
test('models.json carries no request timeout, because the harness would ignore one', () => {
// The obvious place to put a request deadline is beside `contextWindow`, and
// it does nothing there. `ModelDefinitionSchema` in the installed harness has
// no `timeoutMs`; neither does `Model` in `@earendil-works/pi-ai`; and the
// only reader is `options.timeoutMs`, which the agent loop never populates.
// A `timeoutMs` written into a model entry validates, loads, freezes and is
// dropped in silence, so this asserts its absence rather than its presence.
const document = JSON.parse(piggyModelsJsonText()) as {
providers: Record<string, { models: Record<string, unknown>[] }>;
};
for (const model of document.providers[PIGGY_PROVIDER_ID]?.models ?? []) {
assert.equal(
'timeoutMs' in model,
false,
`${String(model.id)} declares a timeoutMs that nothing reads; the deadline belongs in piggyAgentSettings()`,
);
}
});
apps/piggy/test/stall-guard.test.ts
+576
View File
@@ -0,0 +1,576 @@
/**
* What the chat server does about a turn the endpoint stops answering.
*
* The failure this file pins was observed in production: `POST
* /chat/completions` began hanging while `GET /models` still answered in 0.2s,
* so the stream emitted its `meta` frame and then nothing at all, for ever, and
* the transcript span until the browser gave up. A direct `fetch` from Node ran
* past 180 seconds without settling. The harness owns the HTTP call now and sets
* no deadline on it, so the guard has to live where PIG can see the turn: the
* session's event stream.
*
* Every session here is a double, and deliberately so — the endpoint that
* caused this cannot be asked to stall on demand, and a test that depended on it
* would be untrustworthy in exactly the conditions it exists for. A double that
* never settles is the same silence, and it is deterministic besides.
*/
import assert from 'node:assert/strict';
import type { AddressInfo } from 'node:net';
import test from 'node:test';
import type { AgentSession, AgentSessionEvent, ToolDefinition } from '@earendil-works/pi-coding-agent';
import type { PiggyChatEvent, PiggyModelOption } from '@pig/core';
import type { Database } from '@pig/db';
import type { PiggySession } from '../src/agent/session';
import { startPiggyChatServer, type PiggyChatServerOptions } from '../src/chat-server';
import type { PiggyStallLimits } from '../src/config';
import type { PigWriteToolDeps } from '../src/write-tools';
const TOKEN = 'test-internal-token-for-piggy-000000';
const MODELS: PiggyModelOption[] = [
{
id: 'nvidia/nemotron-3-nano-30b-a3b',
label: 'Nemotron 3 Nano',
costPerMTokIn: 0.05,
costPerMTokOut: 0.2,
contextWindow: 131_072,
reasoning: true,
isDefault: true,
},
];
interface RecordedRun {
values: Record<string, unknown>;
closed?: Record<string, unknown>;
}
function fakeDatabase(runs: RecordedRun[]): Database {
return {
insert: () => ({
values: (values: Record<string, unknown>) => ({
returning: async () => {
runs.push({ values });
return [{ id: `run-${runs.length}` }];
},
}),
}),
update: () => ({
set: (closed: Record<string, unknown>) => ({
where: async () => {
const run = runs.at(-1);
if (run) run.closed = closed;
},
}),
}),
select: () => ({ from: () => ({ where: async () => [{ spent: '0' }] }) }),
} as unknown as Database;
}
type TurnScript = (
tools: readonly ToolDefinition[],
emit: (event: AgentSessionEvent) => void,
signal: AbortSignal,
) => Promise<void>;
interface SessionSpy {
created: number;
aborted: number;
disposed: number;
}
function spy(): SessionSpy {
return { created: 0, aborted: 0, disposed: 0 };
}
/**
* A session whose `prompt()` does whatever the script does, including nothing.
*
* `abort()` fires the script's signal, which is how the real harness tells a
* turn to stop; a script that ignores it stands in for a harness that cannot
* unwind because the socket underneath it has no deadline either.
*/
function sessions(script: TurnScript, watched: SessionSpy) {
return async (options: {
tools: readonly ToolDefinition[];
modelId?: string;
}): Promise<PiggySession> => {
watched.created += 1;
const listeners = new Set<(event: AgentSessionEvent) => void>();
const aborted = new AbortController();
const session = {
subscribe(listener: (event: AgentSessionEvent) => void) {
listeners.add(listener);
return () => listeners.delete(listener);
},
async prompt() {
await script(
options.tools,
(event) => {
for (const listener of [...listeners]) listener(event);
},
aborted.signal,
);
},
async abort() {
watched.aborted += 1;
aborted.abort();
},
dispose() {},
} as unknown as AgentSession;
return {
session,
modelId: options.modelId ?? MODELS[0]!.id,
systemPrompt: 'You are Piggy.',
dispose: () => {
watched.disposed += 1;
aborted.abort();
},
} satisfies PiggySession;
};
}
function textDelta(delta: string): AgentSessionEvent {
return {
type: 'message_update',
message: { role: 'assistant' },
assistantMessageEvent: { type: 'text_delta', contentIndex: 0, delta },
} as unknown as AgentSessionEvent;
}
function turnEnd(input: number, output: number, stopReason = 'stop'): AgentSessionEvent {
return {
type: 'turn_end',
message: { role: 'assistant', usage: { input, output }, stopReason },
toolResults: [],
} as unknown as AgentSessionEvent;
}
function toolStart(id: string, name: string): AgentSessionEvent {
return {
type: 'tool_execution_start',
toolCallId: id,
toolName: name,
args: {},
} as unknown as AgentSessionEvent;
}
/** The harness's own bookkeeping, which is not the model doing any work. */
function turnStart(): AgentSessionEvent {
return { type: 'turn_start' } as unknown as AgentSessionEvent;
}
function stallLimits(overrides: Partial<PiggyStallLimits> = {}): PiggyStallLimits {
return { firstProgressMs: 5_000, idleMs: 5_000, ...overrides };
}
async function startForTest(
t: { after: (fn: () => void) => void },
runs: RecordedRun[],
options: Partial<PiggyChatServerOptions>,
): Promise<string> {
const server = startPiggyChatServer(fakeDatabase(runs), {
port: 0,
internalToken: TOKEN,
models: MODELS,
createReadTools: () => [],
createWriteTools: () => [],
limits: { maxModelCalls: 8, maxTurnTokens: 40_000, dailyLimitCents: 0 },
stallLimits: stallLimits(),
...options,
});
t.after(() => server.close());
await new Promise((resolve) => server.once('listening', resolve));
const { port } = server.address() as AddressInfo;
return `http://127.0.0.1:${port}`;
}
const PRINCIPAL = {
userId: '20000000-0000-4000-8000-000000000001',
email: 'ada@primeintellect.example',
name: 'Ada',
isPlatformAdmin: false,
teams: [{ team: 'supply', role: 'lead' }],
via: 'jwt',
scopes: ['read', 'write'],
};
const authorised = { authorization: `Bearer ${TOKEN}`, 'content-type': 'application/json' };
function chatBody(overrides: Record<string, unknown> = {}): string {
return JSON.stringify({
principal: PRINCIPAL,
message: 'What is idle costing us?',
mode: 'read_only',
conversationId: 'conv-stall',
...overrides,
});
}
function parseFrames(body: string): PiggyChatEvent[] {
return body
.trim()
.split('\n')
.filter((line) => line.length > 0)
.map((line) => JSON.parse(line) as PiggyChatEvent);
}
function errorFrame(frames: PiggyChatEvent[]): { message: string; code?: string } | null {
const frame = frames.at(-1);
return frame?.type === 'error' ? { message: frame.message, ...(frame.code ? { code: frame.code } : {}) } : null;
}
/** Silence, until somebody tells the turn to stop. A harness that unwinds. */
const untilAborted: TurnScript = (_tools, _emit, signal) =>
new Promise<void>((resolve) => {
if (signal.aborted) {
resolve();
return;
}
signal.addEventListener('abort', () => resolve(), { once: true });
});
function readStall(closed: Record<string, unknown> | undefined): Record<string, unknown> | undefined {
return (closed?.result as { stall?: Record<string, unknown> } | undefined)?.stall;
}
// ------------------------------------------------------- the endpoint goes quiet
test('a turn the endpoint never answers is ended by the first-progress deadline', async (t) => {
const runs: RecordedRun[] = [];
const watched = spy();
const base = await startForTest(t, runs, {
stallLimits: stallLimits({ firstProgressMs: 120 }),
createSession: sessions(untilAborted, watched),
});
const started = Date.now();
const response = await fetch(`${base}/internal/chat`, {
method: 'POST',
headers: authorised,
body: chatBody(),
});
const frames = parseFrames(await response.text());
const elapsed = Date.now() - started;
// The whole bug, in one assertion: this used to hang until the browser gave
// up, and now it settles inside the deadline it was given.
assert.ok(elapsed < 2_000, `the turn took ${elapsed}ms to give up`);
assert.equal(frames[0]?.type, 'meta');
assert.equal(errorFrame(frames)?.code, 'inference_stalled');
assert.match(String(errorFrame(frames)?.message), /never answered/);
assert.equal(
frames.some((frame) => frame.type === 'done'),
false,
'a stalled turn must not also report itself finished',
);
// The session is told to stop rather than left generating into nothing.
assert.equal(watched.aborted, 1);
assert.ok(watched.disposed >= 1);
// And an operator can tell a silent endpoint from a fault without a log: the
// reason names the deadline, and `result.stall` names which of the two it was.
const closed = runs[0]?.closed;
assert.equal(closed?.status, 'failed');
assert.match(String(closed?.error), /first_progress deadline/);
assert.equal(readStall(closed)?.phase, 'first_progress');
assert.equal(readStall(closed)?.ceilingMs, 120);
assert.ok(Number(readStall(closed)?.waitedMs) >= 120);
});
test("the harness's own bookkeeping does not count as the model working", async (t) => {
const runs: RecordedRun[] = [];
const watched = spy();
const base = await startForTest(t, runs, {
stallLimits: stallLimits({ firstProgressMs: 150, idleMs: 30_000 }),
createSession: sessions(async (tools, emit, signal) => {
// `turn_start` is announced the instant a prompt is submitted, before a
// byte has left the process. If it counted as progress the turn would
// fall into the far more generous idle window and the hang would be back.
emit(turnStart());
await untilAborted(tools, emit, signal);
}, watched),
});
const frames = parseFrames(
await fetch(`${base}/internal/chat`, { method: 'POST', headers: authorised, body: chatBody() })
.then((response) => response.text()),
);
assert.equal(errorFrame(frames)?.code, 'inference_stalled');
assert.equal(readStall(runs[0]?.closed)?.phase, 'first_progress');
});
test('a turn that goes quiet part way through is ended by the idle deadline', async (t) => {
const runs: RecordedRun[] = [];
const watched = spy();
const base = await startForTest(t, runs, {
stallLimits: stallLimits({ firstProgressMs: 30_000, idleMs: 120 }),
createSession: sessions(async (tools, emit, signal) => {
emit(toolStart('call_1', 'pig_get_idle_capacity'));
emit(textDelta('Idle is '));
// The socket dies here, mid-sentence, and never says another word.
await untilAborted(tools, emit, signal);
}, watched),
});
const frames = parseFrames(
await fetch(`${base}/internal/chat`, { method: 'POST', headers: authorised, body: chatBody() })
.then((response) => response.text()),
);
// What did arrive is still shown; the reader is told it is not the whole of
// the answer rather than being left with a truncated one that looks finished.
assert.ok(frames.some((frame) => frame.type === 'content_delta'));
assert.equal(errorFrame(frames)?.code, 'inference_stalled');
assert.match(String(errorFrame(frames)?.message), /went quiet/);
assert.equal(watched.aborted, 1);
const closed = runs[0]?.closed;
assert.equal(closed?.status, 'failed');
assert.equal(closed?.summary, 'Idle is');
assert.match(String(closed?.error), /idle deadline/);
assert.equal(readStall(closed)?.phase, 'idle');
assert.equal(readStall(closed)?.ceilingMs, 120);
});
test('a stall is not reported as a fault, and a fault is not reported as a stall', async (t) => {
// Three things can end a turn early and they want three different responses
// from whoever reads the code: wait, investigate, and do nothing. They must
// not share a name.
const runs: RecordedRun[] = [];
const base = await startForTest(t, runs, {
stallLimits: stallLimits({ firstProgressMs: 30_000, idleMs: 30_000 }),
createSession: sessions(async (_tools, emit) => {
emit(textDelta('Idle is '));
emit({
type: 'turn_end',
message: {
role: 'assistant',
usage: { input: 120, output: 4 },
stopReason: 'error',
errorMessage: 'upstream returned 502',
},
toolResults: [],
} as unknown as AgentSessionEvent);
}, spy()),
});
const frames = parseFrames(
await fetch(`${base}/internal/chat`, { method: 'POST', headers: authorised, body: chatBody() })
.then((response) => response.text()),
);
assert.equal(errorFrame(frames)?.code, 'inference_failed');
assert.equal(readStall(runs[0]?.closed), undefined);
});
// ----------------------------------------------------- what must NOT be killed
test('a slow but progressing answer is never cut off, however long it takes', async (t) => {
const runs: RecordedRun[] = [];
const watched = spy();
// Twelve chunks, 40ms apart: 480ms in total, which is four times the idle
// deadline and twice the first-progress one. A flat deadline over the turn —
// the obvious implementation, and the wrong one — would kill this, and it is
// precisely the long answer the product exists to give.
const base = await startForTest(t, runs, {
stallLimits: stallLimits({ firstProgressMs: 250, idleMs: 120 }),
createSession: sessions(async (_tools, emit) => {
for (let index = 0; index < 12; index += 1) {
await new Promise((resolve) => setTimeout(resolve, 40));
emit(textDelta(`part ${index} `));
}
emit(turnEnd(4_000, 400));
}, watched),
});
const started = Date.now();
const frames = parseFrames(
await fetch(`${base}/internal/chat`, { method: 'POST', headers: authorised, body: chatBody() })
.then((response) => response.text()),
);
assert.ok(Date.now() - started >= 400, 'the turn did not actually run long');
assert.equal(frames.at(-1)?.type, 'done');
assert.equal(
frames.some((frame) => frame.type === 'error'),
false,
'a turn that kept arriving was killed for taking a while',
);
assert.equal(watched.aborted, 0);
assert.equal(runs[0]?.closed?.status, 'succeeded');
assert.equal(readStall(runs[0]?.closed), undefined);
});
/** A write tool that parks on a human, the way `confirm` mode really does. */
function proposingWriteTools(applied: string[]): (deps: PigWriteToolDeps) => ToolDefinition[] {
return ({ propose }) => [
{
name: 'pig_log_activity',
async execute() {
const decision = await propose({
tool: 'pig_log_activity',
kind: 'activity',
summary: 'Log a call on Northwind Robotics',
fields: [{ label: 'Subject', value: 'Capacity review' }],
});
if (decision === 'apply') applied.push('applied');
return {
content: [{ type: 'text', text: `The change was ${decision}.` }],
details: { tool: 'pig_log_activity', status: decision === 'apply' ? 'applied' : 'declined' },
};
},
} as unknown as ToolDefinition,
];
}
test('a write parked on a human outlives the idle deadline and still applies', async (t) => {
const runs: RecordedRun[] = [];
const applied: string[] = [];
const watched = spy();
// The card is left on screen for five times the idle deadline. A turn parked
// on `propose()` emits nothing at all by design, so a watchdog that could not
// see the rendezvous would kill every write Piggy ever proposed — and it
// would do it to the one flow where being killed loses real work.
const base = await startForTest(t, runs, {
stallLimits: stallLimits({ firstProgressMs: 500, idleMs: 100 }),
approvalTimeoutMs: 30_000,
createWriteTools: proposingWriteTools(applied),
createSession: sessions(async (tools, emit, signal) => {
const tool = tools.find((candidate) => candidate.name === 'pig_log_activity');
assert.ok(tool, 'the write tool should have been handed over');
emit(toolStart('call_1', 'pig_log_activity'));
const result = await tool.execute('call_1', {}, signal, undefined, undefined as never);
emit({
type: 'tool_execution_end',
toolCallId: 'call_1',
toolName: 'pig_log_activity',
result,
isError: false,
} as unknown as AgentSessionEvent);
emit(textDelta('Logged.'));
emit(turnEnd(200, 20));
}, watched),
});
const response = await fetch(`${base}/internal/chat`, {
method: 'POST',
headers: authorised,
body: chatBody({ mode: 'confirm', message: 'Log a call on Northwind.' }),
});
const body = response.body;
assert.ok(body);
const reader = body.getReader();
const decoder = new TextDecoder();
let buffered = '';
const frames: PiggyChatEvent[] = [];
const drain = (chunk: Uint8Array | undefined): void => {
buffered += decoder.decode(chunk, { stream: true });
const lines = buffered.split('\n');
buffered = lines.pop() ?? '';
for (const line of lines) if (line) frames.push(JSON.parse(line) as PiggyChatEvent);
};
while (!frames.some((frame) => frame.type === 'approval_required')) {
const { done, value } = await reader.read();
if (done) break;
drain(value);
}
const asked = frames.find((frame) => frame.type === 'approval_required');
assert.ok(asked && asked.type === 'approval_required');
const thinking = Date.now();
await new Promise((resolve) => setTimeout(resolve, 500));
const decision = await fetch(`${base}/internal/approve`, {
method: 'POST',
headers: authorised,
body: JSON.stringify({
conversationId: 'conv-stall',
changeId: asked.change.id,
decision: 'apply',
}),
});
assert.equal(decision.status, 202);
assert.ok(Date.now() - thinking >= 500, 'the human did not actually take their time');
while (true) {
const { done, value } = await reader.read();
if (done) break;
drain(value);
}
assert.equal(frames.at(-1)?.type, 'done');
assert.equal(
frames.some((frame) => frame.type === 'error'),
false,
'a turn waiting on a person was reported as a silent endpoint',
);
// And it did not merely survive: the change the human approved was applied.
assert.deepEqual(applied, ['applied']);
const result = frames.find((frame) => frame.type === 'tool_result');
assert.deepEqual(result?.type === 'tool_result' ? result.result : null, {
tool: 'pig_log_activity',
status: 'applied',
});
assert.equal(watched.aborted, 0);
assert.equal(runs[0]?.closed?.status, 'succeeded');
});
test('the happy path is untouched', async (t) => {
const runs: RecordedRun[] = [];
const watched = spy();
const base = await startForTest(t, runs, {
createSession: sessions(async (_tools, emit) => {
emit(toolStart('call_1', 'pig_get_idle_capacity'));
emit(textDelta('Idle is $12,000.'));
emit(turnEnd(1_240, 180));
}, watched),
});
const frames = parseFrames(
await fetch(`${base}/internal/chat`, { method: 'POST', headers: authorised, body: chatBody() })
.then((response) => response.text()),
);
assert.deepEqual(
frames.map((frame) => frame.type),
['meta', 'tool_call', 'content_delta', 'done'],
);
assert.equal(watched.aborted, 0);
const closed = runs[0]?.closed;
assert.equal(closed?.status, 'succeeded');
assert.equal(closed?.error, null);
assert.equal(readStall(closed), undefined);
});
// ---------------------------------------------------- a harness that will not stop
test('a harness that ignores the abort still gives the browser its answer', async (t) => {
const runs: RecordedRun[] = [];
const watched = spy();
// The nastier shape of the same fault: the session is told to stop and the
// request underneath it has no deadline either, so `prompt()` never settles.
// Trusting that promise would rebuild the hang one level up, so the turn is
// raced against the stall and ends anyway.
const base = await startForTest(t, runs, {
stallLimits: stallLimits({ firstProgressMs: 100 }),
createSession: sessions(() => new Promise<void>(() => {}), watched),
});
const started = Date.now();
const frames = parseFrames(
await fetch(`${base}/internal/chat`, { method: 'POST', headers: authorised, body: chatBody() })
.then((response) => response.text()),
);
const elapsed = Date.now() - started;
assert.equal(errorFrame(frames)?.code, 'inference_stalled');
assert.equal(watched.aborted, 1, 'the session was told to stop, even though it did not');
// Long enough to have waited for a clean unwind, short enough to be nothing
// like the three minutes the endpoint spent not answering.
assert.ok(elapsed >= 100, `the turn ended in ${elapsed}ms, before its own deadline`);
assert.ok(elapsed < 10_000, `the turn took ${elapsed}ms to give up`);
assert.equal(runs[0]?.closed?.status, 'failed');
assert.equal(readStall(runs[0]?.closed)?.phase, 'first_progress');
});