diff --git a/apps/api/src/app.ts b/apps/api/src/app.ts index f08a8f6..ffdbe0e 100644 --- a/apps/api/src/app.ts +++ b/apps/api/src/app.ts @@ -49,6 +49,7 @@ import { createConfiguredAuthProvider, type AuthProvider, } from './lib/auth-provider'; +import { activityPayloadColumns, toActivityPayload } from './lib/activity-payload'; import { apiError } from './lib/mutation'; import { createMediaRoutes } from './lib/media'; import { CapacityService } from './services/capacity'; @@ -388,7 +389,7 @@ export function createApp( db.select().from(supplyDeals).where(eq(supplyDeals.accountId, id)), db.select().from(contracts).where(eq(contracts.accountId, id)), db - .select() + .select(activityPayloadColumns) .from(activities) .where(eq(activities.accountId, id)) .orderBy(desc(activities.occurredAt)) @@ -418,7 +419,7 @@ export function createApp( demandDeals: demand, supplyDeals: supply, contracts: paperwork, - activities: recentActivity, + activities: recentActivity.map(toActivityPayload), dealContacts: buyingGroup, }); }); @@ -573,7 +574,7 @@ export function createApp( */ .where(inArray(supplyDeals.stage, [...SUPPLY_OPEN_STAGES])), db - .select({ activity: activities, accountName: accounts.name }) + .select({ ...activityPayloadColumns, accountName: accounts.name }) .from(activities) .leftJoin(accounts, eq(accounts.id, activities.accountId)) .orderBy(desc(activities.occurredAt)) @@ -592,8 +593,13 @@ export function createApp( compliance, // The subject alone reads as an anonymous feed — "Chased the firm quote" // says nothing until you know whose. The name comes from the join rather - // than a second request per row. - recentActivity: recent.map(({ activity, accountName }) => ({ ...activity, accountName })), + // than a second request per row, and the row carries the same attribution + // the account timeline reads, so the same entry cannot be Piggy's on one + // surface and anonymous on the other. + recentActivity: recent.map(({ accountName, ...activity }) => ({ + ...toActivityPayload(activity), + accountName, + })), }); }); diff --git a/apps/api/src/lib/activity-payload.ts b/apps/api/src/lib/activity-payload.ts new file mode 100644 index 0000000..26e549a --- /dev/null +++ b/apps/api/src/lib/activity-payload.ts @@ -0,0 +1,136 @@ +/** + * What an activity looks like on the wire, and who is recorded as having done it. + * + * Two endpoints render a timeline — the account record and the dashboard feed — + * and both selected the whole row, so the payload was whatever the table + * happened to hold that week. That is how `meta` came to ship every stage + * transition blob and Slack permalink to every browser, and it is why nothing + * on the client could safely depend on `external_id` being there at all: it + * arrived by accident, and the next hand-written column list would have removed + * it without anyone noticing. The projection is declared here instead, once, + * and shaped around the two questions a reader of a timeline actually asks: + * what happened, and who did it. + * + * **Who did it is the part that was broken.** Piggy signs a row in one of two + * places, because it makes two shapes of write: + * + * - `pig_log_activity`'s row IS its own audit event, so there is no separate + * audit row to stamp. It carries an `external_id` of `piggy:`, which + * doubles as the idempotency key that stops a retried tool call logging the + * same conversation twice (`write-tools.ts`). + * - Every other write tool leaves `meta.actorAgent = 'piggy'` on the audit row + * the mutation convention inserts beside the change (`attributedToPiggy`). + * + * `activities.actor_agent` itself stays exactly what it has always been — set + * only when the request authenticated as an agent — because a browser session + * authorised by a person really was authorised by that person, and stamping it + * otherwise would be a lie told inside the audit trail. So the derivation lives + * on the read, not on the row: the wire says "an agent produced this entry, and + * it was Piggy", which is precisely what the column means, while the stored row + * goes on recording how the request was authorised. Nothing here writes. + */ +import type { ActivityType } from '@pig/core'; +import { activities } from '@pig/db'; + +/** + * The mark on an `external_id` that Piggy logged the row. + * + * Exported because the client draws the distinction too: a row Piggy logged + * gets attribution in the timeline, and a row synced from Slack or Buzz — which + * also carries an external id — does not. + */ +export const PIGGY_EXTERNAL_ID_PREFIX = 'piggy:'; + +/** What `actorAgent` reads when the row is Piggy's. One spelling, one source. */ +export const PIGGY_AGENT_NAME = 'piggy'; + +/** + * The columns a timeline needs. + * + * `meta`, `source`, `created_at` and `actor_user_id` are deliberately absent. + * `meta` is an internal payload with no reader in the browser, and its one + * client-relevant fact is folded into `actorAgent` below; the other three say + * nothing a timeline shows. Ship them and they become a contract by default. + */ +export const activityPayloadColumns = { + id: activities.id, + type: activities.type, + subject: activities.subject, + body: activities.body, + accountId: activities.accountId, + contactId: activities.contactId, + demandDealId: activities.demandDealId, + supplyDealId: activities.supplyDealId, + actorAgent: activities.actorAgent, + externalId: activities.externalId, + meta: activities.meta, + occurredAt: activities.occurredAt, +}; + +/** The shape `activityPayloadColumns` selects, before serialisation. */ +export interface ActivityRow { + id: string; + type: ActivityType; + subject: string | null; + body: string | null; + accountId: string | null; + contactId: string | null; + demandDealId: string | null; + supplyDealId: string | null; + actorAgent: string | null; + externalId: string | null; + meta: Record | null; + occurredAt: Date; +} + +export interface ActivityPayload { + id: string; + type: ActivityType; + subject: string | null; + body: string | null; + accountId: string | null; + contactId: string | null; + demandDealId: string | null; + supplyDealId: string | null; + /** + * Which agent produced the entry, or null when a person typed it. `'piggy'` + * covers both of the ways Piggy signs a write; see the note at the top. + */ + actorAgent: string | null; + /** + * External identity, for idempotent sync. A `piggy:` prefix — see + * `PIGGY_EXTERNAL_ID_PREFIX` — means Piggy logged this activity itself, as + * opposed to Piggy having changed a record and this being the audit of it. + */ + externalId: string | null; + occurredAt: string; +} + +/** + * Whether Piggy produced this entry, asked of both places it can have said so. + * + * The `meta` arm is read defensively rather than cast: `meta` is free-form JSON + * written by every mutation in the product, and a row where it holds a string + * or an array must answer "no", not throw on the account page. + */ +function producedByPiggy(row: Pick): boolean { + if (row.externalId?.startsWith(PIGGY_EXTERNAL_ID_PREFIX)) return true; + const stamped = row.meta && typeof row.meta === 'object' ? row.meta.actorAgent : null; + return stamped === PIGGY_AGENT_NAME; +} + +export function toActivityPayload(row: ActivityRow): ActivityPayload { + return { + id: row.id, + type: row.type, + subject: row.subject, + body: row.body, + accountId: row.accountId, + contactId: row.contactId, + demandDealId: row.demandDealId, + supplyDealId: row.supplyDealId, + actorAgent: row.actorAgent ?? (producedByPiggy(row) ? PIGGY_AGENT_NAME : null), + externalId: row.externalId, + occurredAt: row.occurredAt.toISOString(), + }; +} diff --git a/apps/api/src/routes/piggy-chat.ts b/apps/api/src/routes/piggy-chat.ts index 15f8c1d..c39202c 100644 --- a/apps/api/src/routes/piggy-chat.ts +++ b/apps/api/src/routes/piggy-chat.ts @@ -539,7 +539,10 @@ export function createPiggyChatRoutes(options: PiggyChatProxyOptions) { { ...apiError( 'piggy_rate_limited', - 'You have reached the hourly limit for Piggy. Try again shortly.', + // No "try again shortly": the client turns `retryAfterSeconds` into + // a wall-clock time and says exactly when Retry comes back, so a + // vaguer version of the same promise here would only contradict it. + "You have used this hour's Piggy questions.", ), retryAfterSeconds: decision.retryAfterSeconds, }, diff --git a/apps/api/src/services/piggy-activity.ts b/apps/api/src/services/piggy-activity.ts index dd2dd98..d940e6a 100644 --- a/apps/api/src/services/piggy-activity.ts +++ b/apps/api/src/services/piggy-activity.ts @@ -29,7 +29,7 @@ * link is resolved only against conversations the caller owns. */ import { and, desc, eq, gte, inArray, isNotNull, isNull, sql } from 'drizzle-orm'; -import type { AgentTaskKind, AgentTaskOutcome } from '@pig/core'; +import { PIGGY_MODES, type AgentTaskKind, type AgentTaskOutcome, type PiggyMode } from '@pig/core'; import type { Database } from '@pig/db'; import { agentRuns, agentTasks, piggyConversations, users } from '@pig/db'; import type { Principal } from '../lib/auth'; @@ -63,6 +63,21 @@ export interface PiggyRunSummary { */ status: string; model: string | null; + /** + * What this turn was allowed to do — the whole safety argument, per row. + * + * PIG's claim is that nothing lands until a person presses Apply, and that + * claim is only auditable if the ledger records which turns were even offered + * write tools. Without it a run that quietly applied five changes under `auto` + * is indistinguishable from one that could not have changed a thing. + * + * Null means the mode was not recorded, which is two real cases and not a + * failure: a queued task run, which has no mode because nobody chose one, and + * a chat turn from before the relay started stamping it. Reported as null + * rather than defaulted to `read_only`, because guessing the safe answer on an + * audit surface is the one direction a wrong guess must never go. + */ + mode: PiggyMode | null; /** The question, for a chat turn; the queued work, for a task run. */ label: string; /** The first line of what Piggy answered. Null on a turn that said nothing. */ @@ -158,6 +173,23 @@ function readString(bag: Record | null, key: string): string | return typeof value === 'string' && value.trim() ? value : null; } +/** + * The mode a chat turn ran in, from the run's `input` blob. + * + * `agent_runs` has no `mode` column; the chat relay writes it into `input` + * alongside the message and the conversation id (`chat-server.ts`, + * `startChatRun`). That is a free-text bag, so the value is checked against the + * ontology rather than cast — a run whose blob says `"mode": "yolo"` must report + * no mode at all, not put an invented one in the ledger. + * + * Exported for the test, which is the only way to exercise a blob the relay + * would never write without standing up a database to hold it. + */ +export function runMode(input: Record | null): PiggyMode | null { + const claimed = readString(input, 'mode'); + return PIGGY_MODES.find((mode) => mode === claimed) ?? null; +} + /** * The conversation a run answered. * @@ -251,6 +283,7 @@ export class PiggyActivityService { agent: row.agent, status: row.status, model: row.model, + mode: runMode(row.input), // A run with neither a question nor a task kind is a row written before // the turn got anywhere; naming it after its status beats an empty cell. label: diff --git a/apps/api/test/activity-attribution.test.ts b/apps/api/test/activity-attribution.test.ts new file mode 100644 index 0000000..4a59c62 --- /dev/null +++ b/apps/api/test/activity-attribution.test.ts @@ -0,0 +1,103 @@ +/** + * That a write Piggy made can be told from one a person typed. + * + * The product's safety argument is that nothing lands until a human presses + * Apply. That argument is only checkable after the fact if the record surfaces + * can say which rows came from the agent — and until this landed they could + * not: an approved write read as hand-typed in the account timeline while the + * seeded row beneath it said "· by piggy". + * + * Piggy signs a row in two places and the payload has to answer for both, so + * both are asserted here, along with the two rows that must NOT be claimed: + * a person's own entry, and a Slack sync that also carries an external id. + */ +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { + PIGGY_EXTERNAL_ID_PREFIX, + toActivityPayload, + type ActivityRow, +} from '../src/lib/activity-payload'; +import { runMode } from '../src/services/piggy-activity'; + +function activity(overrides: Partial = {}): ActivityRow { + return { + id: '60000000-0000-4000-8000-000000000001', + type: 'call', + subject: 'Call with DEMO — Northwind Robotics about extending the H200 block', + body: null, + accountId: '10000000-0000-4000-8000-00000000000a', + contactId: null, + demandDealId: null, + supplyDealId: null, + actorAgent: null, + externalId: null, + meta: null, + occurredAt: new Date('2026-08-13T09:00:00.000Z'), + ...overrides, + }; +} + +test('an activity Piggy logged is attributed to Piggy', () => { + // `pig_log_activity`'s row is its own audit event, so the provenance rides on + // the external id — which is also what stops a retried tool call logging the + // same conversation twice. + const payload = toActivityPayload( + activity({ externalId: `${PIGGY_EXTERNAL_ID_PREFIX}d016db18-a6eb-4857-9cb5-cff3d58c78d0` }), + ); + assert.equal(payload.actorAgent, 'piggy'); + // Still on the wire, because a record surface may want to draw the row Piggy + // logged differently from the audit of a record Piggy changed. + assert.ok(payload.externalId?.startsWith(PIGGY_EXTERNAL_ID_PREFIX)); +}); + +test('the audit of a record Piggy changed is attributed to Piggy', () => { + // Every write tool other than `pig_log_activity` stamps the audit row's meta + // instead, because the mutation convention writes that row, not the tool. + const payload = toActivityPayload(activity({ meta: { actorAgent: 'piggy', piggyTool: 'pig_update_deal_stage' } })); + assert.equal(payload.actorAgent, 'piggy'); +}); + +test('a person’s own entry claims no agent', () => { + assert.equal(toActivityPayload(activity()).actorAgent, null); +}); + +test('a synced entry is not mistaken for Piggy’s', () => { + // Slack and Buzz carry external ids too. Attributing their rows to the agent + // would put words in Piggy's mouth on the surface people audit it from. + const payload = toActivityPayload(activity({ externalId: 'slack:C09QT/1755082800.123' })); + assert.equal(payload.actorAgent, null); +}); + +test('a stored agent stamp still wins', () => { + // An API key really did authenticate as an agent; the derivation must not + // overwrite what the column already recorded. + assert.equal(toActivityPayload(activity({ actorAgent: 'agent' })).actorAgent, 'agent'); +}); + +test('meta that is not an object cannot break the timeline', () => { + // `meta` is free-form JSON written by every mutation in the product. + const hostile = { meta: ['piggy'] as unknown as Record }; + assert.equal(toActivityPayload(activity(hostile)).actorAgent, null); +}); + +test('the payload carries no internal blob', () => { + const payload = toActivityPayload(activity({ meta: { slackPermalink: 'https://…' } })); + assert.equal('meta' in payload, false); + assert.equal(payload.occurredAt, '2026-08-13T09:00:00.000Z'); +}); + +test('a run reports the mode it was allowed to run in', () => { + assert.equal(runMode({ surface: 'chat', mode: 'auto' }), 'auto'); + assert.equal(runMode({ surface: 'chat', mode: 'read_only' }), 'read_only'); +}); + +test('a run that recorded no mode reports none, rather than the safe one', () => { + // A queued task has no mode, and neither do the chat turns written before the + // relay stamped it. Defaulting those to `read_only` would put a claim in the + // ledger that nobody made. + assert.equal(runMode(null), null); + assert.equal(runMode({ surface: 'chat' }), null); + assert.equal(runMode({ mode: 'yolo' }), null); + assert.equal(runMode({ mode: 42 }), null); +}); diff --git a/apps/api/test/piggy-activity.test.ts b/apps/api/test/piggy-activity.test.ts index 660a405..321e17b 100644 --- a/apps/api/test/piggy-activity.test.ts +++ b/apps/api/test/piggy-activity.test.ts @@ -33,6 +33,7 @@ function run(overrides: Partial = {}): PiggyRunSummary { agent: 'piggy', status: 'succeeded', model: 'nvidia/nemotron-3-nano-30b-a3b', + mode: 'confirm', label: 'Are we under water on the Northwind renewal?', summary: 'Yes — the block is 38 per cent idle at the current rate.', error: null, @@ -72,6 +73,9 @@ test('an administrator reads a colleague’s spend and not their question', () = // Everything an audit is actually for, which is the half that belongs to PIG. assert.equal(redacted.status, 'succeeded'); assert.equal(redacted.model, 'nvidia/nemotron-3-nano-30b-a3b'); + // What a turn was allowed to do is the company's record, not the person's + // words: an audit that cannot say which turns could write is not an audit. + assert.equal(redacted.mode, 'confirm'); assert.equal(redacted.costMicroCents, 4_200); assert.equal(redacted.inputTokens, 2_100); assert.equal(redacted.durationMs, 4_000); diff --git a/apps/piggy/src/agent/models.ts b/apps/piggy/src/agent/models.ts index 2100291..b28985a 100644 --- a/apps/piggy/src/agent/models.ts +++ b/apps/piggy/src/agent/models.ts @@ -112,11 +112,27 @@ interface PiggyModelPresentation { const PRESENTATION: Record = { 'nvidia/nemotron-3-nano-30b-a3b': { - hint: 'Fast and cheap. The default: fine for lookups, summaries and logging activity.', - isDefault: true, + hint: 'Cheapest by far, but currently unreliable upstream — see the note on the default below.', }, + /* + * The default is the SUPER, not the nano, and the reason is not quality. + * + * On 2026-08-14 `nvidia/nemotron-3-nano-30b-a3b` stopped answering on Prime + * Inference: the endpoint accepted the connection and never sent response + * headers (UND_ERR_HEADERS_TIMEOUT, three attempts, 45s each), having 429'd + * shortly before. Every other model in this catalogue answered in under two + * seconds on the same key in the same minute, so it was that model's capacity + * rather than our account. The nano had also just fabricated a figure rather + * than admit it had no tool for the question. + * + * Six times the price of the nano is still about $0.0017 a turn, which is + * roughly 117,000 turns on a $200 credit. Availability is worth more than + * that margin for the model everyone lands on. The nano stays in the picker + * for anyone who wants it back. + */ 'nvidia/nemotron-3-super-120b-a12b': { - hint: 'Same family, six times the price. Reach for it when the nano misreads a table.', + hint: 'The default. Same family as the nano, six times the price, and materially steadier.', + isDefault: true, }, 'deepseek/deepseek-v4-pro': { hint: 'Strong arithmetic at open-weight prices. Good for margin and break-even questions.', diff --git a/apps/piggy/src/agent/prompt.ts b/apps/piggy/src/agent/prompt.ts index 08539a8..9dea87f 100644 --- a/apps/piggy/src/agent/prompt.ts +++ b/apps/piggy/src/agent/prompt.ts @@ -51,6 +51,31 @@ const DOMAIN_BRIEFING = `How this business works, so the figures mean what you s - A commitment near expiry at low utilisation is the urgent case, however healthy the book looks in total. - Answer from the tool's own aggregates. If a figure is not in a tool result, say it is not available rather than deriving one.`; +/* + * The grounding rule, stated separately and last so it is the final thing in + * the prompt before the context line. + * + * This is not belt-and-braces. Measured in production: asked how many + * commitments were on the book while the page context offered only + * `pig_get_idle_capacity`, nemotron-nano judged that no tool fitted, called + * nothing, and answered `\(\boxed{4}\)` — a fabricated number, in LaTeX maths + * mode, when the true count was 5. A small model with reasoning disabled will + * reach for prior belief rather than refuse, and it will present the guess with + * the confidence of a calculation. The domain briefing's closing line was + * already telling it not to; it was not enough, because that line reads as + * advice about arithmetic rather than a prohibition on inventing. + * + * So: an explicit ban, the lookup tools named as the way out, and the maths + * formatting forbidden outright — `\boxed{}` is the tell that the model has + * stopped answering about a CRM and started solving a puzzle. + */ +const GROUNDING_RULE = `Grounding, which overrides everything else: +- NEVER state a number, name, date or status about this business unless it appeared in a tool result in THIS conversation. Not from memory, not from what a figure "should" be, not by inference from the page you are on. +- If the tool you were given does not answer the question, do not guess and do not stop: pig_search_records finds a record by name and pig_get_record_by_id opens it. Reach for those before concluding anything. +- A tool result answers only what that tool covers. Never report a filtered count as a total: pig_get_idle_capacity returns the blocks with idle hours, not the book. If the result does not cover the question as asked, say what it does cover and what is missing. +- If no tool can answer it, say exactly that and name what you would need. "I cannot see that from here" is a correct answer. An invented figure is not, and is worse than silence — someone will act on it. +- Never use LaTeX or mathematical notation. No \\boxed{}, no \\(...\\). Write plain prose and plain numbers.`; + /** * The escape hatch from the focus, said out loud. * @@ -199,5 +224,7 @@ ${modeRules(options.mode)} ${toolSection(options.tools ?? [])} +${GROUNDING_RULE} + ${contextLine(options.context)}`; } diff --git a/apps/piggy/src/agent/session.ts b/apps/piggy/src/agent/session.ts index c2010b5..945f299 100644 --- a/apps/piggy/src/agent/session.ts +++ b/apps/piggy/src/agent/session.ts @@ -7,6 +7,7 @@ import { SessionManager, SettingsManager, type AgentSession, + type RetrySettings, type ToolDefinition, } from '@earendil-works/pi-coding-agent'; import type { PiggyChatContext, PiggyMode } from '@pig/core'; @@ -163,6 +164,119 @@ async function piggyAgentRuntime(): Promise { } } +/** + * What Piggy does when Prime Inference says "please retry shortly". + * + * Measured on 2026-08-14, on production, 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"} + * + * and the reader got `{"type":"error","code":"inference_failed"}` and no answer, + * while a `curl` a second later succeeded. The endpoint asked us to retry and we + * did not. `withInferenceRetries` in `apps/piggy/src/provider.ts` still guards + * the queued worker with exactly this policy — bounded attempts, jittered + * backoff, `Retry-After` honoured, 429 and 5xx retried and no other 4xx ever — + * and it was lost for the chat when the harness took over the transport. + * + * The seam is the harness's own provider-request retry rather than a loop of + * ours around `session.prompt()`, and the reason is exactly-once. Read + * `retryProviderRequest` in `@earendil-works/pi-ai/dist/utils/provider-retry.js` + * and then its one caller in `dist/api/openai-completions.js:139`: it wraps the + * creation of the request and nothing else, so every attempt it makes happens + * BEFORE the first byte of the response has been read. A retry there cannot + * duplicate a content delta, cannot re-run `pig_log_activity`, and cannot + * re-apply an approved write, because at that instant none of those has + * happened. The property is structural rather than policed, which is the only + * kind worth having when the failure mode is writing a CRM row twice. It also + * reads `retry-after` and `retry-after-ms`, backs off exponentially with jitter, + * sleeps on the run's own AbortSignal so a caller hanging up wins immediately, + * and retries 408, 409, 429 and 5xx and no other status. + * + * Measured here, with a stubbed fetch, before any of these values were set: + * `retryProviderRequest` defaults `maxRetries` to 0 and `getProviderRetrySettings` + * supplies `undefined`, so the harness made exactly one attempt at every model + * call. That is the whole bug. + * + * `stream` is the second, smaller budget, and it is deliberately not the same + * number. The harness's session-level auto-retry re-drives a turn that failed + * AFTER the response started, by discarding the errored assistant message and + * continuing; that recovers a dropped socket, but it regenerates text the reader + * has already been shown. Measured, on the same stub: a turn that streamed + * "Idle is " and then lost the stream came back as "Idle is Idle is $12,000." in + * the client transcript. So it is kept — a mid-stream drop is the one failure + * the provider-level retry cannot see — but held to a single attempt, and the + * chat server refuses the replay outright once anything has been delivered. + */ +export interface PiggyInferenceRetryPolicy { + /** Attempts at getting a response started, including the first. */ + attempts: number; + /** + * Deadline on one attempt. + * + * A headers deadline, not a turn deadline: the OpenAI client clears its timer + * in a `finally` the moment `fetch` resolves (openai@6.26.0 client.js:387-411), + * so it covers connect and response headers and never the streamed body. That + * is what makes it safe to set this tight — a legitimately long answer is + * measured by the stall watchdog's idle clock instead, which restarts on every + * chunk. 20 seconds is the deadline the hand-rolled chat loop used on the same + * endpoint for the same reason. + */ + headersTimeoutMs: number; + /** + * The longest `Retry-After` worth honouring. + * + * Above this the SDK fails the request immediately and says what was asked + * for, which is the right answer: three attempts each parked on the SDK's own + * 60-second default would leave somebody staring at a docked panel for three + * minutes to be told no. Five seconds twice over is the worst this can add. + */ + maxRetryDelayMs: number; + /** Attempts at a turn that failed after the response started, first included. */ + streamAttempts: number; + /** First backoff for those, doubling per attempt. */ + streamBackoffMs: number; +} + +export const PIGGY_INFERENCE_RETRY: PiggyInferenceRetryPolicy = { + attempts: 4, + headersTimeoutMs: 20_000, + maxRetryDelayMs: 5_000, + streamAttempts: 2, + streamBackoffMs: 1_500, +}; + +/** + * The policy above, in the field names the installed harness actually reads. + * + * Exported because it is the only honest way to test this: the values are read + * by `SettingsManager` and nothing else in PIG, so a test asserts that the + * installed package hands them back rather than asserting that we wrote an + * object. That check matters more than it sounds. The obvious place to put a + * request timeout is the model entry in models.json, and it does nothing there: + * `ModelDefinitionSchema` in the harness (dist/core/model-config.js:133-147) has + * no `timeoutMs`, `Model` in `@earendil-works/pi-ai` has no such field, and the + * only reader is `options.timeoutMs`, which `Agent.createLoopConfig()` never + * populates. A `timeoutMs` written beside `contextWindow` would validate, load, + * freeze, and be ignored, with nothing anywhere to say so. + */ +export function piggyAgentSettings( + policy: PiggyInferenceRetryPolicy = PIGGY_INFERENCE_RETRY, +): NonNullable[0]> { + const retry: RetrySettings = { + enabled: policy.streamAttempts > 1, + maxRetries: Math.max(0, policy.streamAttempts - 1), + baseDelayMs: policy.streamBackoffMs, + provider: { + maxRetries: Math.max(0, policy.attempts - 1), + maxRetryDelayMs: policy.maxRetryDelayMs, + timeoutMs: policy.headersTimeoutMs, + }, + }; + return { retry }; +} + async function buildAgentRuntime(): Promise { const config = loadPiggyConfig(); const agentDir = prepareAgentDir(config.PIGGY_AGENT_DIR); @@ -189,8 +303,10 @@ async function buildAgentRuntime(): Promise { modelRuntime, // In-memory settings, because SettingsManager.create writes the chosen // model and thinking level back to settings.json. With a model picker per - // user, that would make one person's choice the process-wide default. - settingsManager: SettingsManager.inMemory(), + // user, that would make one person's choice the process-wide default. It is + // also the only seam that reaches the harness's HTTP call: the retry budget + // and the request deadline are read off this object once per model call. + settingsManager: SettingsManager.inMemory(piggyAgentSettings()), agentDir, config, }; diff --git a/apps/piggy/src/chat-server.ts b/apps/piggy/src/chat-server.ts index 7d364bd..958097a 100644 --- a/apps/piggy/src/chat-server.ts +++ b/apps/piggy/src/chat-server.ts @@ -27,7 +27,12 @@ import { type PiggyTurnBudget, } from './agent/session'; import { toPrimeTools } from './agent/tool-bridge'; -import { loadPiggyTurnLimits, type PiggyTurnLimits } from './config'; +import { + loadPiggyStallLimits, + loadPiggyTurnLimits, + type PiggyStallLimits, + type PiggyTurnLimits, +} from './config'; import { assertPigToolBoundary, type PiggyChatContext } from './chat'; import { createInteractivePigTools } from './chat-tools'; import { createPigWriteTools, type PigWriteToolDeps } from './write-tools'; @@ -152,6 +157,12 @@ export interface PiggyChatServerOptions { * two that can disagree. */ limits?: PiggyTurnLimits; + /** + * How long a turn may go silent before the endpoint is presumed to have gone + * quiet. Read from the environment when absent, for the same reason as + * `limits`. + */ + stallLimits?: PiggyStallLimits; } export function startPiggyChatServer(db: Database, options: PiggyChatServerOptions): Server { @@ -174,6 +185,7 @@ export function startPiggyChatServer(db: Database, options: PiggyChatServerOptio // Resolved once, at bind time, so a malformed ceiling fails the process // rather than the first user to ask a question. limits: options.limits ?? loadPiggyTurnLimits(), + stall: options.stallLimits ?? loadPiggyStallLimits(), }; const defaultModel = defaultModelOption(resolved.models); @@ -223,6 +235,7 @@ interface ResolvedOptions { createReadTools: (db: Database, context: PiggyChatContext | undefined) => ToolDefinition[]; approvals: ApprovalRegistry; limits: PiggyTurnLimits; + stall: PiggyStallLimits; } /** @@ -240,6 +253,210 @@ function defaultModelOption(models: readonly PiggyModelOption[]): PiggyModelOpti return first; } +// -------------------------------------------------------------- the stall + +/** Which silence ended the turn. */ +type StallPhase = 'first_progress' | 'idle'; + +/** A turn the endpoint stopped answering, and how long it was given first. */ +interface TurnStall { + phase: StallPhase; + /** How long the turn had been silent when the deadline bit. */ + waitedMs: number; + /** The deadline it passed, in its own units. */ + ceilingMs: number; +} + +/** + * The events that mean the model itself is working. + * + * Deliberately narrower than "any event". The harness announces `agent_start` + * and `turn_start` the instant a prompt is submitted, before a byte has left the + * process, so counting those as progress would start the first-progress clock + * and satisfy it in the same tick — which is exactly the hang this guard is for. + * Once a turn has genuinely started, any event at all is accepted as a sign of + * life, because by then the harness is demonstrably running its loop. + */ +const PROGRESS_EVENTS: ReadonlySet = new Set([ + 'message_update', + 'tool_execution_start', + 'turn_end', +]); + +/** + * How long a stalled turn is given to unwind itself before the server stops + * waiting for it. + * + * `session.abort()` should reject the in-flight request and settle `prompt()` + * promptly. Should. The whole reason this guard exists is that the layer holding + * the socket had no deadline of its own, so trusting the same layer to honour an + * abort — and leaving the browser hanging if it does not — would rebuild the bug + * one level up. Two seconds is long enough for a clean unwind and short enough + * that nobody watches it. + */ +const STALL_UNWIND_GRACE_MS = 2_000; + +/** + * The turn-level stall detector. + * + * It watches the session's event stream rather than the HTTP call, because the + * HTTP call belongs to the harness and this must be a guard the harness cannot + * swallow. Two deadlines, and the distinction is the point of the whole class: + * + * first progress — nothing has arrived since `prompt()` was called. The turn + * never started; the request went out and the endpoint did + * not answer. + * idle — the turn started and then went quiet. This clock restarts + * on every event, so a long answer that keeps arriving runs + * as long as it likes. A flat overall deadline would kill + * exactly the legitimate long turns this product wants. + * + * `parked` is the third rule and the one that would otherwise break the write + * flow. A `confirm`-mode turn sits inside `propose()` waiting up to five minutes + * for a human, and by design that produces no events whatsoever. That is not a + * stall, it is the product working, so the clock is pushed forward for as long + * as the rendezvous holds an unanswered card. + */ +class TurnStallWatchdog { + private phaseStartedAt = Date.now(); + private progressed = false; + private timer?: ReturnType; + private stopped = false; + private fired = false; + private abandon: (() => void) | undefined; + + /** + * Resolves only when a stalled turn's harness has not unwound within the + * grace period. Raced against `prompt()` so a hung request cannot hold the + * browser open even if the abort is ignored. + */ + readonly abandoned: Promise; + + constructor( + private readonly limits: PiggyStallLimits, + /** True while a proposed write is waiting on a human. */ + private readonly parked: () => boolean, + private readonly onStall: (stall: TurnStall) => void, + ) { + this.abandoned = new Promise((resolve) => { + this.abandon = resolve; + }); + } + + /** Starts the first-progress clock. Call immediately before `prompt()`. */ + start(): void { + this.phaseStartedAt = Date.now(); + this.arm(); + } + + /** One event off the session stream. */ + observe(type: AgentSessionEvent['type']): void { + const wasProgressing = this.progressed; + if (PROGRESS_EVENTS.has(type)) this.progressed = true; + else if (!this.progressed) return; + this.phaseStartedAt = Date.now(); + if (!wasProgressing) { + /* + * The one reset that must not wait for the sleeping timer. + * + * Every other reset only ever pushes the deadline later, and a timer that + * wakes too early simply re-arms for what is left. The first progress + * event is different: it swaps the generous first-progress window for the + * tighter idle one, and a timer already asleep on the longer of the two + * cannot notice until it has expired. Measured, before this line existed: + * a turn with a 120ms idle deadline that fell silent after its first token + * ran for 30,001ms — the first-progress window — which is the hang this + * whole guard is for, wearing the guard's own clothes. + */ + this.rearm(); + } + } + + /** + * The human rendezvous moved — a card was raised, or answered. + * + * Not a model event, but not silence either, and it closes the narrow window + * where an approval settles a moment before the deadline it was suspending + * would have fired. + */ + touch(): void { + this.phaseStartedAt = Date.now(); + } + + stop(): void { + this.stopped = true; + if (this.timer) clearTimeout(this.timer); + } + + private window(): number { + return this.progressed ? this.limits.idleMs : this.limits.firstProgressMs; + } + + private rearm(): void { + if (this.timer) clearTimeout(this.timer); + this.arm(); + } + + /** + * One self-rescheduling timer rather than a poll: it sleeps exactly as long as + * the current deadline has left, and every reset simply moves the deadline it + * wakes up to compare against. + */ + private arm(): void { + if (this.stopped || this.fired) return; + const waited = Date.now() - this.phaseStartedAt; + const ceilingMs = this.window(); + if (waited < ceilingMs) { + this.timer = setTimeout(() => this.arm(), ceilingMs - waited); + // A deadline must never be a reason for the process to stay alive. + this.timer.unref?.(); + return; + } + if (this.parked()) { + // Waiting on a person, not on the endpoint. The clock restarts from now, + // so the turn gets a full window once the card is answered. + this.phaseStartedAt = Date.now(); + this.arm(); + return; + } + this.fired = true; + this.onStall({ phase: this.progressed ? 'idle' : 'first_progress', waitedMs: waited, ceilingMs }); + this.timer = setTimeout(() => this.abandon?.(), STALL_UNWIND_GRACE_MS); + this.timer.unref?.(); + } +} + +/** + * Tell the user the endpoint went quiet, and tell the ledger which silence it + * was. + * + * A distinct code, because an operator must be able to tell three different + * events apart without reading a log: `inference_stalled` is the endpoint saying + * nothing, `inference_failed` is the model or the endpoint reporting a fault, + * and `turn_limit_exceeded` is PIG's own policy stopping a turn that was working + * perfectly well. They want three different responses — wait, investigate, and + * nothing at all — so they must not share a name. + */ +function reportStall( + stall: TurnStall, + spend: ChatRunOutcome, + emit: (event: PiggyChatEvent) => void, +): void { + spend.stall = stall; + spend.error = + `turn stopped by the ${stall.phase} deadline: no session event for ${stall.waitedMs}ms, ` + + `deadline ${stall.ceilingMs}ms`; + console.warn(`[piggy] ${spend.error}`); + emit({ + type: 'error', + message: + stall.phase === 'first_progress' + ? 'Piggy asked the inference endpoint and it never answered, so nothing was attempted. That is the endpoint rather than your question — try again shortly.' + : 'The inference endpoint went quiet part way through this answer, so it is incomplete. Try again shortly.', + code: 'inference_stalled', + }); +} + // ------------------------------------------------------------------- the turn async function handleChatTurn( @@ -319,8 +536,16 @@ async function handleChatTurn( beginStream(response); + // Declared before `emit` because `emit` feeds it: an approval being raised or + // answered is movement on the turn, and it is the only movement the session's + // own event stream never reports. + let watchdog: TurnStallWatchdog | undefined; + const emit = (event: PiggyChatEvent): void => { recordEvent(spend, event); + if (event.type === 'approval_required' || event.type === 'approval_resolved') { + watchdog?.touch(); + } // An abandoned turn still has approvals to cancel and a session to unwind, // and both emit as they settle. Writing those to a closed socket would // throw inside the unwinding and take the ledger down with it. @@ -344,7 +569,7 @@ async function handleChatTurn( * every turn that went wrong is answering the wrong question, and it errs in * the reassuring direction, which is the worst way for it to be wrong. */ - const state: TurnState = { inputTokens: 0, outputTokens: 0, modelCalls: 0 }; + const state: TurnState = { inputTokens: 0, outputTokens: 0, modelCalls: 0, retries: 0 }; try { const tools = buildToolSet(db, options, { principal, @@ -395,32 +620,94 @@ async function handleChatTurn( void session?.session.abort().catch(() => {}); }; + /* + * The third brake, and the only one that fires when the harness is not + * running at all. + * + * `shouldStopAfterTurn` and the `turn_end` counter above both need the loop + * to be turning; a turn stuck on a request the endpoint never answers turns + * nothing, spends nothing and trips neither. It is stopped the same way a + * runaway is — abort the session — and reported as its own thing, because + * "the endpoint went quiet" and "the turn cost too much" are not the same + * news for either the user or the operator. + */ + watchdog = new TurnStallWatchdog(options.stall, turn.hasPending, (stall) => { + if (state.stopping) return; + state.stopping = true; + state.stalled = stall; + // Fire and forget, for the same reason the budget's abort is: this runs on + // a timer inside the run, and awaiting the run going idle from within it + // would deadlock. + void session?.session.abort().catch(() => {}); + }); + + /* + * The fourth brake, and the only one that stops the turn to protect the + * transcript rather than the bill. + * + * `piggyAgentSettings` buys most of its retries inside the SDK's request + * wrapper, where nothing has been delivered yet and a retry is invisible by + * construction. This is the other kind: the harness's session-level retry + * fires after the response has already started, discards the errored + * assistant message and generates a replacement. Measured against a stubbed + * endpoint, a turn that had streamed "Idle is " before the stream dropped + * came back as "Idle is Idle is $12,000." — the reader is shown a sentence + * twice, in a panel whose whole job is to be trusted about numbers. + * + * So the rule is the simple one rather than the clever one: retry only + * before the first delta. Once anything has been delivered the turn is + * stopped and reported as incomplete, which is what the stall guard already + * says about a half-finished answer and what the reader can act on. Aborting + * during the backoff is enough: `AgentSession.abort` cancels the retry sleep + * before it re-drives (agent-session.js:1168-1172). + */ + const refuseReplay = (): void => { + if (!state.delivered || state.stopping) return; + state.stopping = true; + state.replayRefused = true; + // Fire and forget, for the same reason the other two brakes are: this + // runs inside the run's own event handling, and `abort` awaits the run + // going idle. + void session?.session.abort().catch(() => {}); + }; + const unsubscribe = session.session.subscribe((event) => { + watchdog?.observe(event.type); translateSessionEvent(event, emit, state); if (event.type === 'turn_end') enforceBudget(); + if (event.type === 'auto_retry_start') refuseReplay(); }); try { - await session.session.prompt(body.message); + watchdog.start(); + const prompt = session.session.prompt(body.message); + // Two handlers on one promise. The race is what ends the turn when an + // aborted harness does not unwind; this `catch` is what keeps the prompt's + // eventual rejection from becoming an unhandled rejection once the race + // has already been decided against it. Awaiting `prompt` still throws, so + // the ordinary failure path is untouched. + void prompt.catch(() => {}); + await Promise.race([prompt, watchdog.abandoned]); } finally { unsubscribe(); + watchdog.stop(); abort.signal.removeEventListener('abort', disposeOnAbort); } - if (abort.signal.aborted) { + if (state.stalled) { + // Checked before the abort state, and it costs nothing to do so: a reader + // who had already left would have disposed the session and settled the + // prompt long before this deadline could bite, so a stall recorded here + // came first and is the reason the turn ended. + reportStall(state.stalled, spend, emit); + } else if (abort.signal.aborted) { // The reader left. The harness may have resolved the prompt rather than // rejecting it, and recording that as a completed turn would report an // answer nobody received as delivered. spend.aborted = true; } else if (cutShortByBudget(budget, state)) { reportBudgetBreach(budget, spend, emit); - } else if (state.errorMessage !== undefined) { - // The model stopped on a fault of its own rather than throwing, so the - // turn ends as an error frame and the run is recorded as failed. A `done` - // here would present a truncated answer as a finished one. The ledger - // keeps the real reason; the browser gets the sanitised one. - spend.error = state.errorMessage; - console.error('[piggy] chat turn ended in an inference error:', state.errorMessage); - emit({ type: 'error', message: 'Piggy could not finish this answer.', code: 'inference_failed' }); + } else if (state.replayRefused || state.errorMessage !== undefined) { + reportInferenceFailure(state, spend, emit); } else { // A turn that passed a ceiling on its own last call still records the // breach below, because that is the reading an operator tuning the @@ -446,13 +733,27 @@ async function handleChatTurn( // Server-side, with the real reason. The client gets none of it: the // upstream body is echoed into these messages and is not ours to relay. console.error('[piggy] chat turn failed:', failure); - if (!spend.aborted && budget.breach) { + if (!spend.aborted && state.stalled) { + // The abort this server fires at a silent endpoint usually surfaces here, + // as a rejected prompt rather than a resolved one. What ended the turn is + // still the stall, and saying "Piggy chat failed" would send an operator + // hunting a fault in code that behaved correctly. + reportStall(state.stalled, spend, emit); + if (!response.writableEnded) response.end(); + } else if (!spend.aborted && budget.breach) { // The abort this server fires to stop a runaway can surface here as a // rejected prompt rather than as a resolved one. The ceiling is what // ended the turn; reporting it as "Piggy chat failed" would send an // operator hunting a fault that is really a policy. reportBudgetBreach(budget, spend, emit); if (!response.writableEnded) response.end(); + } else if (!spend.aborted && (state.replayRefused || state.errorMessage !== undefined)) { + // The abort this server fires to refuse a replay surfaces here as a + // rejected prompt. What ended the turn is the upstream fault that + // provoked the retry, and the reader is owed that reason rather than a + // bare "Piggy chat failed" for a decision this server took on purpose. + reportInferenceFailure(state, spend, emit); + if (!response.writableEnded) response.end(); } else { spend.error = failure; if (!response.writableEnded) { @@ -481,6 +782,10 @@ async function handleChatTurn( spend.costMicroCents ??= costMicroCents(state, model); spend.modelCalls = state.modelCalls; spend.overran = budget.overran; + // `??=` because a reported failure has already written the reading it was + // measured against, and a turn that recovered still needs its count here. + spend.attempts ??= state.retries + 1; + spend.retryReason ??= state.retryReason; // In a finally so that every exit closes the row, including the exit // that is not a fault at all: a reader who navigates away aborts the // turn mid-answer. A row left `running` cannot be told from a turn still @@ -530,6 +835,25 @@ interface TurnState { errorMessage?: string; /** The session has already been told to stop; do not tell it twice. */ stopping?: boolean; + /** Set when the stall watchdog, rather than the model, ended the turn. */ + stalled?: TurnStall; + /** + * Turn-level retries the harness announced, which is one per `auto_retry_start`. + * + * It counts the retries that were visible. The provider-level ones — the four + * attempts `piggyAgentSettings` buys inside the SDK's own request wrapper — + * produce no assistant message and no session event, so no honest counter can + * see them from here. An operator reading `agent_runs.result.inference` should + * therefore read `attempts` as "times this turn had to be started again after + * the endpoint had already begun answering", not as a count of HTTP requests. + */ + retries: number; + /** How the endpoint described the fault that caused the last retry. */ + retryReason?: string; + /** Content deltas have been written to the reader's transcript. */ + delivered?: boolean; + /** A retry was refused because it would have replayed a delivered answer. */ + replayRefused?: boolean; } /** @@ -578,6 +902,75 @@ function reportBudgetBreach( }); } +/** + * Is this the endpoint throttling us, or something that will fail again? + * + * Matched on the text because that is all there is: the harness reports a + * failed model call as an `errorMessage` on the assistant message and keeps no + * status code, so by the time the fault reaches PIG the HTTP response is long + * gone. The strings are the ones Prime Inference actually sends — `429: + * {"message":"Rate limit reached. Please retry shortly.","type": + * "rate_limit_exceeded","code":"rate_limited"}` — plus the shapes the SDK + * substitutes when it never got a body, and `ResourceExhausted`, which is what + * a gRPC-backed model behind the same endpoint says instead. + */ +function isRateLimited(errorMessage: string): boolean { + return /\b429\b|rate.?limit|rate_limited|too many requests|resourceexhausted/i.test(errorMessage); +} + +/** + * Tell the user which fault it was, and tell the ledger how hard we tried. + * + * `inference_rate_limited` is its own code because it is the one inference + * fault the person at the keyboard can do something about: waiting ten seconds + * genuinely fixes it, and it is not worth an operator's pager. `inference_failed` + * keeps its old meaning — something broke and somebody should look — so the two + * must not share a name any more than `inference_stalled` and + * `turn_limit_exceeded` do. + * + * The counts go to `agent_runs`, not to the browser. "Piggy tried four times" + * is not a sentence that helps anybody decide what to type next; it is exactly + * the sentence an operator needs when deciding whether today's rate limiting is + * worse than yesterday's. + */ +function reportInferenceFailure( + state: TurnState, + spend: ChatRunOutcome, + emit: (event: PiggyChatEvent) => void, +): void { + const errorMessage = state.errorMessage ?? 'the endpoint failed without saying why'; + const rateLimited = isRateLimited(errorMessage); + const attempts = state.retries + 1; + // Server-side, with the real reason and the counts. The upstream body is + // echoed into this and is not ours to relay to a browser. + spend.error = + `${errorMessage} (${attempts} attempt${attempts === 1 ? '' : 's'}` + + (state.replayRefused ? ', retry refused: part of the answer had already been delivered' : '') + + ')'; + spend.attempts = attempts; + console.error('[piggy] chat turn ended in an inference error:', spend.error); + + if (state.replayRefused) { + // The reader keeps what arrived, and is told plainly that it is not all of + // it. Restarting would have shown them the first half twice. + emit({ + type: 'error', + message: rateLimited + ? 'The inference endpoint started rate limiting us part way through this answer, so it is incomplete. Piggy will not restart it, because that would repeat what you have already been shown. Ask again in a few seconds.' + : 'The inference endpoint failed part way through this answer, so it is incomplete. Piggy will not restart it, because that would repeat what you have already been shown. Ask again shortly.', + code: rateLimited ? 'inference_rate_limited' : 'inference_failed', + }); + return; + } + emit({ + type: 'error', + message: rateLimited + ? `Prime Inference is rate limiting us, so this question was never answered. Piggy asked ${attempts} time${attempts === 1 ? '' : 's'} and was turned away each time. Wait a few seconds and ask again; the refused attempts generated nothing, so none of this was charged to you.` + : 'Piggy could not finish this answer.', + code: rateLimited ? 'inference_rate_limited' : 'inference_failed', + }); +} + /** * The harness's vocabulary, narrowed to PIG's. * @@ -594,11 +987,53 @@ function translateSessionEvent( ): void { switch (event.type) { case 'message_update': { + /* + * Nothing more of this turn may reach the transcript once a replay has + * been refused. + * + * Aborting the session should be enough, and the whole reason that guard + * exists is that a harness which cannot unwind is precisely the fault + * being defended against — the stall watchdog already races `prompt()` + * for the same reason. A harness that carried on regardless would stream + * the replacement answer over the top of the one the reader already has, + * which is the duplication the abort was meant to prevent. Dropping the + * deltas here makes exactly-once a property of PIG's own code rather than + * a favour from somebody else's. + */ + if (state.replayRefused) return; const streamed = event.assistantMessageEvent; - if (streamed.type === 'text_delta') emit({ type: 'content_delta', delta: streamed.delta }); - if (streamed.type === 'thinking_delta') emit({ type: 'reasoning_delta', delta: streamed.delta }); + // Recorded before either is written, because it is what makes a retry + // unsafe: from here on the reader has part of a turn in front of them, + // and starting it again would show them a second copy. Reasoning counts + // as well as answer text — it is a rendered panel, not a private + // scratchpad, and a restarted turn would repeat that too. + if (streamed.type === 'text_delta') { + state.delivered = true; + emit({ type: 'content_delta', delta: streamed.delta }); + } + if (streamed.type === 'thinking_delta') { + state.delivered = true; + emit({ type: 'reasoning_delta', delta: streamed.delta }); + } return; } + /* + * The harness is about to restart a turn that failed. + * + * Announced, backed off and then re-driven by `AgentSession`, which + * discards the errored assistant message and continues. It is counted here + * rather than dropped with the rest of the harness's vocabulary because a + * turn that needed two goes and a turn that needed one look identical in + * the ledger otherwise, and "how often is this happening" is the first + * question an operator asks about a rate limit. + */ + case 'auto_retry_start': + state.retries += 1; + state.retryReason = event.errorMessage; + console.warn( + `[piggy] retrying a chat turn after ${event.errorMessage} (attempt ${event.attempt} of ${event.maxAttempts}, in ${event.delayMs}ms)`, + ); + return; case 'tool_execution_start': emit({ type: 'tool_call', id: event.toolCallId, name: event.toolName, arguments: event.args }); return; @@ -635,7 +1070,21 @@ function translateSessionEvent( state.inputTokens += assistant.inputTokens; state.outputTokens += assistant.outputTokens; state.stopReason = assistant.stopReason ?? state.stopReason; - if (assistant.errorMessage) state.errorMessage = assistant.errorMessage; + // Set on a fault and CLEARED on the next model call that is not one. + // + // It used to latch, and that was the second half of the production + // failure: the harness retries a 429 of its own accord and often + // succeeds, but the errored `turn_end` had already stamped this field, so + // a turn that recovered and streamed a perfectly good answer was still + // closed as `inference_failed` with the 429 in the ledger. The only way a + // model call can follow an errored one at all is that something retried + // it, so the later reading is the one that describes how the turn ended. + // + // Unless the retry was one this server refused. Then the replacement + // answer was never delivered, whatever the harness did with it, and + // letting its success clear the fault would report a turn as finished on + // the strength of an answer nobody was shown. + if (!state.replayRefused) state.errorMessage = assistant.errorMessage; return; } default: @@ -741,7 +1190,20 @@ class ApprovalRegistry { open( conversationId: string, emit: (event: PiggyChatEvent) => void, - ): { propose: PigWriteToolDeps['propose']; cancelAll: () => void } { + ): { + propose: PigWriteToolDeps['propose']; + cancelAll: () => void; + /** + * True while this turn is waiting on a human. + * + * Published because the stall watchdog has to be able to tell a turn parked + * on `propose()` from a turn nobody is answering: the parked one emits + * nothing for up to five minutes by design, and killing it would break the + * write flow entirely. `owned` is the honest source — it holds exactly the + * cards this turn has raised and not yet settled. + */ + hasPending: () => boolean; + } { const owned = new Set(); const propose: PigWriteToolDeps['propose'] = (draft) => @@ -770,7 +1232,7 @@ class ApprovalRegistry { for (const key of [...owned]) this.pending.get(key)?.settle('reject', 'cancelled'); }; - return { propose, cancelAll }; + return { propose, cancelAll, hasPending: () => owned.size > 0 }; } /** False when nothing is pending: an unknown id, or one already settled. */ @@ -871,6 +1333,20 @@ interface ChatRunOutcome { modelCalls?: number; /** The ceiling this turn passed, if it passed one. */ breach?: PiggyTurnBreach; + /** The silence that ended this turn, if one did. */ + stall?: TurnStall; + /** + * Times the turn had to be started again, the first go included. + * + * Written for every turn rather than only the failed ones, because the + * question an operator has is "how often is the endpoint making us retry", + * and a column that only ever appears on failures cannot answer it: a day + * where every turn needed two attempts and succeeded looks, in that ledger, + * exactly like a day where none of them did. + */ + attempts?: number; + /** How the endpoint described the fault behind the last of them. */ + retryReason?: string; /** The loop kept going after the ceiling and had to be aborted. */ overran?: boolean; } @@ -954,6 +1430,14 @@ async function finishChatRun( approvalsRequested: outcome.approvalsRequested, approvalsApplied: outcome.approvalsApplied, modelCalls: outcome.modelCalls ?? 0, + // What it took to get an answer at all. `attempts: 1` is the healthy + // reading and the common one; anything above it is the endpoint + // making the product slower, which is a trend rather than an + // incident and so belongs in a column rather than in a log line. + inference: { + attempts: outcome.attempts ?? 1, + ...(outcome.retryReason ? { retryReason: outcome.retryReason } : {}), + }, ...(outcome.breach ? { limit: { @@ -965,6 +1449,20 @@ async function finishChatRun( }, } : {}), + // A stalled turn closes as `failed`, not as `aborted`: unlike a + // ceiling, which is PIG stopping a turn that was working, this is the + // upstream not answering, and an operator watching the failure rate + // should see it. `stall` says which silence it was, so it can still be + // told from a fault without reading a log. + ...(outcome.stall + ? { + stall: { + phase: outcome.stall.phase, + waitedMs: outcome.stall.waitedMs, + ceilingMs: outcome.stall.ceilingMs, + }, + } + : {}), }, inputTokens: outcome.inputTokens ?? null, outputTokens: outcome.outputTokens ?? null, diff --git a/apps/piggy/src/config.ts b/apps/piggy/src/config.ts index 0881e15..2024ba7 100644 --- a/apps/piggy/src/config.ts +++ b/apps/piggy/src/config.ts @@ -99,6 +99,57 @@ const turnLimitShape = { PIGGY_CHAT_DAILY_LIMIT_CENTS: z.coerce.number().int().nonnegative().default(200), }; +/** + * How long a turn may say nothing at all before the server stops believing in + * it. + * + * This is a guard that existed, was lost, and was then needed on the same day. + * The hand-rolled chat loop had a 20,000ms deadline on an attempt's headers and + * a 30,000ms idle deadline that restarted on every streamed chunk — deliberately + * two deadlines rather than one, because a flat overall deadline kills a + * legitimately long answer, and a long answer that is arriving is exactly the + * turn worth protecting. Moving to the Prime Agent harness handed the HTTP call + * to somebody else, and the guard did not come with it. + * + * Then `POST /chat/completions` began hanging. `GET /models` still answered in + * 0.2s, so the endpoint was up and only the inference path was stalled or + * throttling us; a bare `fetch` from Node ran past 180 seconds without settling. + * The user saw the `meta` frame and then nothing, for ever, with the transcript + * spinning until the browser gave up. The harness cannot help here: its + * OpenAI-completions path passes a request timeout through only when the model + * entry supplies one, and ours does not, so the fetch has no deadline of any + * kind. Hence a deadline at the level the harness cannot swallow — the session's + * own event stream, which the chat server already subscribes to. + * + * The two windows measure different silences and neither substitutes for the + * other: + * + * first progress — from `prompt()` to the first sign that the model is + * working. It has to cover connecting, the endpoint's queue, + * a slow frontier model's first token and any retry the + * harness makes without announcing it. 60 seconds is three + * times the old header deadline, which is the honest premium + * for a harness whose internals we do not time. + * idle — the longest gap between two signs of life once the turn is + * under way. Mid-stream gaps are milliseconds; the widest + * legitimate gap is a tool result followed by the next model + * call's first token, and a retry the harness announces + * resets this clock because an announced retry is an event. + * 45 seconds is half again the old idle deadline and well + * past anything measured, and it resets on every event, so a + * ten-minute answer that keeps arriving is never touched. + * + * Raising these is safe and cheap; the only thing they cost is how long a hung + * socket holds a browser connection. Lowering them below the numbers above is + * how a slow honest answer gets reported as a dead endpoint. + */ +const stallLimitShape = { + /** Milliseconds from `prompt()` to the first sign the model is working. */ + PIGGY_CHAT_FIRST_PROGRESS_TIMEOUT_MS: z.coerce.number().int().positive().default(60_000), + /** Milliseconds of silence allowed between two events once a turn is moving. */ + PIGGY_CHAT_IDLE_TIMEOUT_MS: z.coerce.number().int().positive().default(45_000), +}; + const baseSchema = z.object({ DATABASE_URL: z.string().min(1, 'DATABASE_URL is required.'), /** @@ -163,6 +214,7 @@ const baseSchema = z.object({ .enum(['off', 'minimal', 'low', 'medium', 'high', 'xhigh', 'max']) .default('off'), ...turnLimitShape, + ...stallLimitShape, PIGGY_LEASE_SECONDS: z.coerce.number().int().positive().default(300), PIGGY_POLL_INTERVAL_MS: z.coerce.number().int().positive().default(2_000), PIGGY_MAX_TOKENS: z.coerce.number().int().positive().default(1_024), @@ -236,6 +288,40 @@ export interface PiggyTurnLimits { dailyLimitCents: number; } +/** + * The two silences a turn is allowed, in milliseconds. + * + * Separate from `PiggyTurnLimits` because they answer a different question. + * Those ceilings ask what a turn may spend and are counted in model calls and + * tokens; these ask whether the turn is alive at all and are counted in + * wall-clock. Merging them would invite a future reader to bound a turn's + * duration the way its cost is bounded, which is precisely the flat deadline + * both of these exist to avoid. + */ +export interface PiggyStallLimits { + /** From `prompt()` to the first sign the model is working. */ + firstProgressMs: number; + /** The longest silence allowed between two events once the turn is moving. */ + idleMs: number; +} + +/** + * The stall deadlines alone, parsed without the rest of the environment, for + * the same reason `loadPiggyTurnLimits` exists: the chat server is constructed + * directly by the tests and must not need a DATABASE_URL to hold a deadline. + */ +export function loadPiggyStallLimits(env: NodeJS.ProcessEnv = process.env): PiggyStallLimits { + const parsed = z.object(stallLimitShape).safeParse(env); + if (!parsed.success) { + const issues = parsed.error.issues.map((issue) => ` ${issue.path.join('.')}: ${issue.message}`); + throw new Error(`Invalid Piggy stall deadlines:\n${issues.join('\n')}`); + } + return { + firstProgressMs: parsed.data.PIGGY_CHAT_FIRST_PROGRESS_TIMEOUT_MS, + idleMs: parsed.data.PIGGY_CHAT_IDLE_TIMEOUT_MS, + }; +} + /** * The turn ceilings alone, parsed without the rest of the environment. * diff --git a/apps/piggy/test/agent-models.test.ts b/apps/piggy/test/agent-models.test.ts index 95074e2..bf6d702 100644 --- a/apps/piggy/test/agent-models.test.ts +++ b/apps/piggy/test/agent-models.test.ts @@ -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 }[]; + 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', () => { diff --git a/apps/piggy/test/chat-retry.test.ts b/apps/piggy/test/chat-retry.test.ts new file mode 100644 index 0000000..3c13109 --- /dev/null +++ b/apps/piggy/test/chat-retry.test.ts @@ -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; + closed?: Record; +} + +function fakeDatabase(runs: RecordedRun[]): Database { + return { + insert: () => ({ + values: (values: Record) => ({ + returning: async () => { + runs.push({ values }); + return [{ id: `run-${runs.length}` }]; + }, + }), + }), + update: () => ({ + set: (closed: Record) => ({ + 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; + +interface SessionSpy { + aborted: number; +} + +function sessions(script: TurnScript, watched: SessionSpy) { + return async (options: { tools: readonly ToolDefinition[]; modelId?: string }): Promise => { + 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((resolve) => { + if (signal.aborted) { + resolve(); + return; + } + signal.addEventListener('abort', () => resolve(), { once: true }); + }); + +function stallLimits(overrides: Partial = {}): PiggyStallLimits { + return { firstProgressMs: 5_000, idleMs: 5_000, ...overrides }; +} + +async function startForTest( + t: { after: (fn: () => void) => void }, + runs: RecordedRun[], + options: Partial, +): Promise { + 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 { + 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 { + 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 => frame.type === 'content_delta') + .map((frame) => frame.delta) + .join(''); +} + +function inference(closed: Record | undefined): Record | undefined { + return (closed?.result as { inference?: Record } | 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'); +}); diff --git a/apps/piggy/test/chat-server.test.ts b/apps/piggy/test/chat-server.test.ts index af25232..9805d0f 100644 --- a/apps/piggy/test/chat-server.test.ts +++ b/apps/piggy/test/chat-server.test.ts @@ -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 }, }); }); diff --git a/apps/piggy/test/config.test.ts b/apps/piggy/test/config.test.ts index 4bdfd14..122959d 100644 --- a/apps/piggy/test/config.test.ts +++ b/apps/piggy/test/config.test.ts @@ -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. diff --git a/apps/piggy/test/inference-retry.test.ts b/apps/piggy/test/inference-retry.test.ts new file mode 100644 index 0000000..380c9fa --- /dev/null +++ b/apps/piggy/test/inference-retry.test.ts @@ -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): 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((_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 { + 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[] }>; + }; + 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()`, + ); + } +}); diff --git a/apps/piggy/test/stall-guard.test.ts b/apps/piggy/test/stall-guard.test.ts new file mode 100644 index 0000000..21470c9 --- /dev/null +++ b/apps/piggy/test/stall-guard.test.ts @@ -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; + closed?: Record; +} + +function fakeDatabase(runs: RecordedRun[]): Database { + return { + insert: () => ({ + values: (values: Record) => ({ + returning: async () => { + runs.push({ values }); + return [{ id: `run-${runs.length}` }]; + }, + }), + }), + update: () => ({ + set: (closed: Record) => ({ + 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; + +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 => { + 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 { + return { firstProgressMs: 5_000, idleMs: 5_000, ...overrides }; +} + +async function startForTest( + t: { after: (fn: () => void) => void }, + runs: RecordedRun[], + options: Partial, +): Promise { + 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 { + 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((resolve) => { + if (signal.aborted) { + resolve(); + return; + } + signal.addEventListener('abort', () => resolve(), { once: true }); + }); + +function readStall(closed: Record | undefined): Record | undefined { + return (closed?.result as { stall?: Record } | 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(() => {}), 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'); +}); diff --git a/apps/web/src/App.tsx b/apps/web/src/App.tsx index 5e7ef35..ab0068b 100644 --- a/apps/web/src/App.tsx +++ b/apps/web/src/App.tsx @@ -16,7 +16,8 @@ import { SignIn } from '@/pages/SignIn'; import { CreateProfile } from '@/pages/CreateProfile'; import { Register } from '@/pages/Register'; import { PiggyMark } from '@/components/PiggyMark'; -import { Badge, Card, EmptyState, Skeleton } from '@/components/ui'; +import { Badge, Button, Card, EmptyState, Section, Skeleton, Stat } from '@/components/ui'; +import { PageHeader } from '@/components/ui/page-header'; import { Avatar, AvatarFallback } from '@/components/ui/avatar'; import { Toaster } from '@/components/ui/sonner'; import { usePageTitle } from '@/lib/title'; @@ -323,10 +324,16 @@ function RoutePage({ children }: { children: React.ReactNode }) { * inset's own tab-bar clearance does not apply to it, and without this the * composer would sit underneath the phone tab bar — the exact control a phone * user came here to reach. `lg` matches where the tab bar gives way. + * + * Under 500px tall the reserve is given back. A phone in landscape, or a phone + * with the keyboard up, is spending 72px of a 390px viewport on a bar it can + * reach again by turning the handset back — while the transcript, which is why + * the page exists, is measured at 40px. The tab bar itself stands down at the + * same height (Shell.tsx), so nothing lands underneath it. */ function WorkspaceRoute({ children }: { children: React.ReactNode }) { return ( -
+
{/* `flex-1` on the fallback, or the spinner for a pane this tall sits up against the header while the rest of it stays empty. */}
}> @@ -379,7 +386,7 @@ function Placeholder({ title }: { title: string }) { function Team() { usePageTitle('Team'); - const { data, isLoading, error } = useQuery({ + const { data, isLoading, error, refetch } = useQuery({ queryKey: ['team'], queryFn: () => get< @@ -397,39 +404,55 @@ function Team() { const representedTeams = new Set((data ?? []).flatMap((person) => person.teams.map((team) => team.team))).size; return ( + /* + * The one page that never got a design pass, because it never had an owner: + * it lives inline in App.tsx rather than in `pages/`, so the wave that swept + * all thirteen routes swept past it. Measured against the rest of the + * product it carried a 30px `

` where every other route is 24px, an + * accent-coloured "ACCESS MAP" eyebrow of exactly the kind the direction + * deleted from Growth (identity colour used as decoration, on a page with + * no agent in it), hand-rolled 10px/24px stat tiles instead of `Stat`, and + * a 14px section heading floating on the canvas. It is now the same three + * primitives every other page is built from and nothing else changed. + */
-
-
-

Access map

-

Team

-

- See who can operate each side of the compute business and where ownership is thin. -

-
- - Manage access in Settings - -
+ + Manage access in Settings + + } + /> -
- {[ - ['People', data?.length ?? 0], - ['Teams', representedTeams], - ['Assignments', assignments], - ].map(([label, value]) => ( - -

{label}

-

{value}

- - ))} + {/* `grid-cols-2 gap-3 xl:grid-cols-*`, the same KPI row Overview and + Margin use. This carried `grid-cols-3 sm:max-w-xl`, which made Team + the one page whose headline figures were a different size and whose + row stopped halfway across the page. */} +
+ + +
{error ? ( - + {/* Three routes rendered an honest error and then offered nothing to + do about it. A transient 500 on a page with no Try again is a page + a person has to know to reload. */} + void refetch()}> + Try again + + } + /> ) : null} @@ -446,11 +469,10 @@ function Team() { ) : null} {!isLoading && !error && data?.length ? ( -
-
-

People and permissions

- Roles are enforced server-side -
+
{data.map((person) => { const initials = person.name @@ -486,7 +508,7 @@ function Team() { ); })}
-
+
) : null}
); diff --git a/apps/web/src/components/AccountSwitcher.tsx b/apps/web/src/components/AccountSwitcher.tsx index 023c768..fa6efd4 100644 --- a/apps/web/src/components/AccountSwitcher.tsx +++ b/apps/web/src/components/AccountSwitcher.tsx @@ -1,11 +1,16 @@ /** * The account tile at the top of the sidebar. * - * It carries the Piggy mark in the user's own accent, because that accent is - * the one piece of the interface they chose and the workspace identity is - * where they will look for it. The swatch row in the menu is the same - * `setAccent` the Settings page calls — not a copy of the palette, and not a - * second place a colour could be defined. + * It used to carry the Piggy mark, which made the pig face mean two things at + * once: the agent, and your organisation. Piggy is now one mark with one + * meaning everywhere in the product, so this tile carries a monogram instead — + * the workspace's initials on the trigger, the signed-in person's on the menu + * label above their own email. Both sit in the user's chosen accent, because + * that accent is the one piece of the interface they picked and identity is + * where they will look for it. + * + * The swatch row in the menu is the same `setAccent` the Settings page calls — + * not a copy of the palette, and not a second place a colour could be defined. * * PIG is single-workspace today, so this is a switcher with one entry. It is * still a menu rather than a label: it is where identity, appearance and @@ -18,7 +23,6 @@ import type { ThemeMode } from '@pig/core'; import { getSupabase } from '@/lib/api'; import { useIdentity } from '@/lib/identity'; import { useTheme } from '@/lib/theme'; -import { PiggyMark } from './PiggyMark'; import { SidebarMenu, SidebarMenuButton, SidebarMenuItem, useSidebar } from './ui/sidebar'; import { DropdownMenu, @@ -28,10 +32,48 @@ import { DropdownMenuSeparator, DropdownMenuTrigger, } from './ui/dropdown-menu'; -import { cn } from './ui'; +import { Label, cn } from './ui'; const WORKSPACE_NAME = 'Prime Intellect Growth'; +/** + * Two letters at most. + * + * Three initials in a 32px square is a monogram nobody can read, and a name + * with one word still has to fill the chip rather than sit in the corner of + * it. Falls back to the first character of whatever it was given, because an + * empty chip beside a name reads as a failed avatar load. + */ +function monogram(name: string): string { + const words = name.trim().split(/\s+/).filter(Boolean); + if (words.length === 0) return '?'; + if (words.length === 1) return words[0]!.slice(0, 2).toUpperCase(); + return `${words[0]![0]!}${words[1]![0]!}`.toUpperCase(); +} + +/** + * The chip both monograms sit in, so they cannot drift apart. + * + * Rounded square for an organisation, circle for a person — the convention + * every product this one sits beside already uses, and the fastest way to say + * which of the two rows in this menu is your workspace and which is you. Both + * are decorative: the name they stand for is always printed next to them. + */ +function Monogram({ text, shape }: { text: string; shape: 'workspace' | 'person' }) { + return ( + + {text} + + ); +} + const MODES: { value: ThemeMode; label: string; icon: typeof Sun }[] = [ { value: 'light', label: 'Light', icon: Sun }, { value: 'dark', label: 'Dark', icon: Moon }, @@ -63,9 +105,7 @@ export function AccountSwitcher() { className="data-[state=open]:bg-sidebar-accent" aria-label={`${WORKSPACE_NAME} — account and appearance`} > - - - + {WORKSPACE_NAME} {identity.name} @@ -81,9 +121,7 @@ export function AccountSwitcher() { sideOffset={8} > - - - + {identity.name} {identity.email} @@ -92,8 +130,8 @@ export function AccountSwitcher() { - - Accent + +
{accents.map((option) => ( @@ -124,8 +162,8 @@ export function AccountSwitcher() { - - Appearance + + {MODES.map((option) => ( -
-
-
+
+
-
-
-

Platform control plane

- Admin only -
-

- Configure intelligence, inventory sync, workspace entry, and team authority. -

-
+
Admin only} + />
- - Runtime - Invites - Access - Integrations + + Runtime + Invites + Access + Integrations - + {isLoading || !data ?

Loading runtime settings…

: } - - - + + + ); @@ -165,24 +165,22 @@ function RuntimeForm({ settings }: { settings: AdminRuntimeSettings }) { -
Prime inventory
+
Prime inventory

Compute endpoint: {settings.primeComputeBase}
- {settings.primeApiKey.configured ? 'Credential configured' : 'Credential missing'} + {settings.primeApiKey.configured ? 'Credential configured' : 'Credential missing'} {settings.primeApiKey.source ? {settings.primeApiKey.source} source : null} {settings.primeApiKey.updatedAt ? updated {relativeTime(settings.primeApiKey.updatedAt)} : null}
- + {settings.primeApiKey.source === 'database' ? : null}
- + setIntervalValue(event.target.value)} />
@@ -234,7 +232,7 @@ function PiggyCard({
- + Piggy intelligence
{issuedCode ?

Shown once. Send it through a secure channel.

{issuedCode}
: null} Invite ledger

Only metadata remains visible after issuance.

{/* A failed ledger read must not render as "no invites issued": an admin who believes the workspace is empty issues a second code to someone who already has one. */} - {ledger.isPending ?
Loading invites…{[0, 1].map((row) => )}
: ledger.isError ? } title="Invite ledger unavailable" description={ledger.error.message} action={} /> : ledger.data.length === 0 ?

No invites issued yet.

: ledger.data.map((invite) =>

{invite.email ?? 'Workspace invite'}

{invite.status}

{invite.team ? TEAM_LABELS[invite.team] : 'Team chosen at signup'} · {invite.role} · {invite.usesRemaining} use{invite.usesRemaining === 1 ? '' : 's'} left

{invite.status === 'active' ? : null}
)} + {ledger.isPending ?
Loading invites…{[0, 1].map((row) => )}
: ledger.isError ? } title="Invite ledger unavailable" description={ledger.error.message} action={} /> : ledger.data.length === 0 ? : ledger.data.map((invite) =>

{invite.email ?? 'Workspace invite'}

{invite.status}

{invite.team ? TEAM_LABELS[invite.team] : 'Team chosen at signup'} · {invite.role} · {invite.usesRemaining} use{invite.usesRemaining === 1 ? '' : 's'} left

{invite.status === 'active' ? : null}
)}
; } @@ -500,12 +504,13 @@ function MemberManager() { return (
-
- -
-

Team and role administration

-

Roles are team-scoped. Platform administration is a separate grant.

-
+
+ +
{query.isPending ? (
@@ -555,5 +560,5 @@ function MemberAccess({ member }: { member: Member }) { const [isPlatformAdmin, setIsPlatformAdmin] = useState(member.isPlatformAdmin); const [roles, setRoles] = useState>>(() => Object.fromEntries(member.memberships.map(({ team, role }) => [team, role]))); const save = useMutation({ mutationFn: () => patch(`/api/admin/members/${member.id}/access`, { isPlatformAdmin, memberships: TEAMS.flatMap((team) => roles[team] ? [{ team, role: roles[team] }] : []) }), onSuccess: () => void queryClient.invalidateQueries({ queryKey: ['admin-members'] }) }); - return

{member.name}

{member.isPlatformAdmin ? Platform admin : null}

{member.email}

{TEAMS.map((team) => )}
{member.adminSource === 'environment' ?

Pinned by environment

: null}
{save.error ?

{save.error.message}

: null}; + return

{member.name}

{member.isPlatformAdmin ? Platform admin : null}

{member.email}

{TEAMS.map((team) => )}
{member.adminSource === 'environment' ?

Pinned by environment

: null}
{save.error ?

{save.error.message}

: null}; } diff --git a/apps/web/src/components/AllocationSheet.tsx b/apps/web/src/components/AllocationSheet.tsx index aaf07f3..ad53dac 100644 --- a/apps/web/src/components/AllocationSheet.tsx +++ b/apps/web/src/components/AllocationSheet.tsx @@ -1,4 +1,4 @@ -import { useEffect, useMemo, useState } from 'react'; +import { useEffect, useId, useMemo, useState, type ReactNode } from 'react'; import { zodResolver } from '@hookform/resolvers/zod'; import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query'; import { @@ -9,15 +9,15 @@ import { import { AlertTriangle, Clock3, LoaderCircle, RotateCcw, ShieldCheck } from 'lucide-react'; import { useForm, type Control, type FieldPath, type FieldValues } from 'react-hook-form'; import { z } from 'zod'; -import { Badge, Button, Input } from '@/components/ui'; +import { Badge, Button, EmptyState, Input, Label, Section, Stat } from '@/components/ui'; import { Form, FormControl, FormDescription, FormField, FormItem, - FormLabel, FormMessage, + useFormField, } from '@/components/ui/form'; import { Select, @@ -36,6 +36,7 @@ import { SheetTitle, } from '@/components/ui/sheet'; import { Textarea } from '@/components/ui/textarea'; +import { UtilisationBar } from '@/components/ui/utilisation-bar'; import { ApiError, compactNumber, dateRange, get, percent, post, shortDate, unitPrice } from '@/lib/api'; import { toast } from 'sonner'; @@ -192,6 +193,7 @@ export function AllocationSheet({ onChanged?(): void; }) { const queryClient = useQueryClient(); + const releaseReasonId = useId(); const [releaseReason, setReleaseReason] = useState(''); const [releaseError, setReleaseError] = useState(null); const form = useForm({ @@ -345,20 +347,19 @@ export function AllocationSheet({ return ( - + Reserve capacity Join committed supply to a demand deal. Availability is re-checked by the server when you save. -
save.mutate(values))} > -
+
{(['allocation', 'hold'] as const).map((value) => (