pig/apps/piggy/test/chat-server.test.ts

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 { PiggyApprovalDecision, PiggyChatEvent, PiggyModelOption } from '@pig/core';
import type { Database } from '@pig/db';
import type { CreatePiggySessionOptions, PiggySession } from '../src/agent/session';
import {
  startPiggyChatServer,
  type PiggyChatServerOptions,
  type PiggySessionFactory,
} from '../src/chat-server';
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,
  },
  {
    id: 'anthropic/claude-opus-5',
    label: 'Claude Opus 5',
    costPerMTokIn: 5,
    costPerMTokOut: 25,
    contextWindow: 200_000,
    reasoning: true,
  },
];

/**
 * The chat server writes exactly two statements per turn — one insert, one
 * update — so a fake that records them is enough to assert the whole ledger.
 */
interface RecordedRun {
  values: Record<string, unknown>;
  closed?: Record<string, unknown>;
}

function fakeDatabase(runs: RecordedRun[]): Database {
  return {
    insert: () => ({
      values: (values: Record<string, unknown>) => ({
        returning: async () => {
          runs.push({ values });
          return [{ id: `run-${runs.length}` }];
        },
      }),
    }),
    update: () => ({
      set: (closed: Record<string, unknown>) => ({
        where: async () => {
          const run = runs.at(-1);
          if (run) run.closed = closed;
        },
      }),
    }),
    // The daily spend, read once before every turn. Zero here, so nothing in
    // this file is refused for cost; `turn-limits.test.ts` owns that ceiling.
    select: () => ({ from: () => ({ where: async () => [{ spent: '0' }] }) }),
  } as unknown as Database;
}

/**
 * A turn, written the way the harness would perform it: the script is handed
 * the tool set the server assembled and an emitter, and drives both. Every
 * event it emits has a real `AgentSessionEvent` shape; the casts are there only
 * because an `AssistantMessage` carries thirty fields the translation never
 * reads, and building all of them would test the fake rather than the server.
 *
 * The third argument is the signal a disposed session aborts. The real harness
 * passes one to every tool `execute` and fires it on `session.abort()`, which is
 * how a tool parked on an approval learns that the reader has gone; a fake that
 * did not would deadlock the moment a turn was abandoned mid-approval.
 */
type TurnScript = (
  tools: readonly ToolDefinition[],
  emit: (event: AgentSessionEvent) => void,
  signal: AbortSignal,
) => Promise<void>;

interface SessionSpy {
  options?: CreatePiggySessionOptions;
  disposed: number;
}

function spy(): SessionSpy {
  return { disposed: 0 };
}

function fakeSessions(script: TurnScript, watched: SessionSpy): PiggySessionFactory {
  return async (options) => {
    watched.options = options;
    const listeners = new Set<(event: AgentSessionEvent) => void>();
    const aborted = new AbortController();
    let disposed = false;
    const session = {
      subscribe(listener: (event: AgentSessionEvent) => void) {
        listeners.add(listener);
        return () => listeners.delete(listener);
      },
      async prompt() {
        await script(
          options.tools,
          (event) => {
            // `createPiggySession`'s dispose aborts the session before dropping
            // its listeners, so a disposed session stops generating rather than
            // streaming into a socket nobody is reading.
            if (disposed) throw new Error('The session was aborted.');
            for (const listener of [...listeners]) listener(event);
          },
          aborted.signal,
        );
      },
      async abort() {},
      dispose() {},
    } as unknown as AgentSession;

    return {
      session,
      modelId: options.modelId ?? MODELS[0]!.id,
      systemPrompt: 'You are Piggy.',
      dispose: () => {
        disposed = true;
        aborted.abort();
        watched.disposed += 1;
      },
    } 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',
  errorMessage?: string,
): AgentSessionEvent {
  return {
    type: 'turn_end',
    message: {
      role: 'assistant',
      usage: { input, output },
      stopReason,
      ...(errorMessage ? { errorMessage } : {}),
    },
    toolResults: [],
  } as unknown as AgentSessionEvent;
}

function toolStart(id: string, name: string, args: unknown): AgentSessionEvent {
  return {
    type: 'tool_execution_start',
    toolCallId: id,
    toolName: name,
    args,
  } as unknown as AgentSessionEvent;
}

function toolEnd(
  id: string,
  name: string,
  result: { content: { type: 'text'; text: string }[]; details?: unknown },
  isError = false,
): AgentSessionEvent {
  return {
    type: 'tool_execution_end',
    toolCallId: id,
    toolName: name,
    result,
    isError,
  } as unknown as AgentSessionEvent;
}

/** An event PIG does not render, which must never reach the wire. */
const noiseEvent = { type: 'queue_update', steering: [], followUp: [] } as unknown as AgentSessionEvent;

/** A tool needs only a name to be assembled, allowlisted and handed over. */
function namedTool(name: string): ToolDefinition {
  return { name } as unknown as ToolDefinition;
}

async function startForTest(
  t: { after: (fn: () => void) => void },
  runs: RecordedRun[],
  options: Partial<PiggyChatServerOptions>,
): Promise<string> {
  const server = startPiggyChatServer(fakeDatabase(runs), {
    port: 0,
    internalToken: TOKEN,
    models: MODELS,
    createReadTools: () => [],
    createWriteTools: () => [],
    ...options,
  });
  t.after(() => server.close());
  // Port 0 is only resolved once the socket is bound.
  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'],
};

function chatBody(overrides: Record<string, unknown> = {}): string {
  return JSON.stringify({
    principal: PRINCIPAL,
    message: 'What is idle costing us?',
    context: { type: 'page', route: '/capacity' },
    mode: 'read_only',
    conversationId: 'conv-1',
    ...overrides,
  });
}

const authorised = { authorization: `Bearer ${TOKEN}`, 'content-type': 'application/json' };

function parseFrames(body: string): PiggyChatEvent[] {
  return body
    .trim()
    .split('\n')
    .filter((line) => line.length > 0)
    .map((line) => JSON.parse(line) as PiggyChatEvent);
}

test('health answers without a token, and nothing else does', async (t) => {
  const runs: RecordedRun[] = [];
  const base = await startForTest(t, runs, { createSession: fakeSessions(async () => {}, spy()) });

  const health = await fetch(`${base}/internal/health`);
  assert.equal(health.status, 200);
  assert.deepEqual(await health.json(), {
    ok: true,
    service: 'piggy-chat',
    model: 'nvidia/nemotron-3-nano-30b-a3b',
  });

  assert.equal((await fetch(`${base}/internal/anything`)).status, 404);
  assert.equal((await fetch(`${base}/internal/models`)).status, 401);
  assert.equal(
    (await fetch(`${base}/internal/chat`, { method: 'POST', body: chatBody() })).status,
    401,
  );
  assert.equal(
    (await fetch(`${base}/internal/approve`, { method: 'POST', body: '{}' })).status,
    401,
  );
});

test('the catalogue is served to the relay, so no client hard-codes a model list', async (t) => {
  const runs: RecordedRun[] = [];
  const base = await startForTest(t, runs, { createSession: fakeSessions(async () => {}, spy()) });

  const response = await fetch(`${base}/internal/models`, { headers: authorised });
  assert.equal(response.status, 200);
  assert.deepEqual(await response.json(), MODELS);
});

test('a turn is translated into PIG events, and the harness noise is dropped', async (t) => {
  const runs: RecordedRun[] = [];
  const watched = spy();
  const base = await startForTest(t, runs, {
    createSession: fakeSessions(async (_tools, emit) => {
      emit(noiseEvent);
      emit(toolStart('call_1', 'pig_get_idle_capacity', { since: '2026-01-01' }));
      emit(
        toolEnd('call_1', 'pig_get_idle_capacity', {
          content: [{ type: 'text', text: '{"totalIdleCostCents":1200000}' }],
          details: { tool: 'pig_get_idle_capacity', result: { totalIdleCostCents: 1_200_000 } },
        }),
      );
      emit(textDelta('Idle is '));
      emit(textDelta('$12,000.'));
      emit(turnEnd(1_240, 180));
    }, watched),
  });

  const response = await fetch(`${base}/internal/chat`, {
    method: 'POST',
    headers: authorised,
    body: chatBody(),
  });
  assert.equal(response.status, 200);
  const frames = parseFrames(await response.text());

  assert.deepEqual(frames[0], {
    type: 'meta',
    model: 'nvidia/nemotron-3-nano-30b-a3b',
    mode: 'read_only',
    conversationId: 'conv-1',
  });
  assert.deepEqual(frames[1], {
    type: 'tool_call',
    id: 'call_1',
    name: 'pig_get_idle_capacity',
    arguments: { since: '2026-01-01' },
  });
  // The bridge already structured the answer in `details`; the panel is handed
  // that rather than the JSON string the model was shown.
  assert.deepEqual(frames[2], {
    type: 'tool_result',
    id: 'call_1',
    name: 'pig_get_idle_capacity',
    ok: true,
    result: { totalIdleCostCents: 1_200_000 },
  });
  assert.deepEqual(frames.slice(3, 5), [
    { type: 'content_delta', delta: 'Idle is ' },
    { type: 'content_delta', delta: '$12,000.' },
  ]);
  assert.deepEqual(frames.at(-1), {
    type: 'done',
    inputTokens: 1_240,
    outputTokens: 180,
    // 1240 x 5 + 180 x 20 micro-cents, at $0.05/$0.20 per million tokens.
    costMicroCents: 9_800,
  });
  assert.equal(frames.length, 6);

  const run = runs[0];
  assert.equal(run?.values.model, 'nvidia/nemotron-3-nano-30b-a3b');
  assert.equal(run?.values.principalUserId, PRINCIPAL.userId);
  assert.equal(run?.closed?.status, 'succeeded');
  assert.equal(run?.closed?.summary, 'Idle is $12,000.');
  assert.equal(run?.closed?.inputTokens, 1_240);
  assert.equal(run?.closed?.costMicroCents, 9_800);
  assert.ok(run?.closed?.finishedAt instanceof Date);
  // The session is unwound on the happy path too; a leaked one holds an
  // inference connection open and billing.
  assert.equal(watched.disposed, 1);
});

// ------------------------------------------------- the harness's vocabulary

/**
 * The four harness events PIG renders something for.
 *
 * Everything else is dropped on the server, deliberately: PIG's own event
 * vocabulary is what the browser speaks, so a harness upgrade is a server change
 * and never a client one.
 */
const MAPPED_EVENTS = [
  'message_update',
  'tool_execution_start',
  'tool_execution_end',
  'turn_end',
] as const;

/**
 * Everything the harness can emit that PIG deliberately drops.
 *
 * Written out in full rather than implied by a `default:` arm, because the
 * failure this catches is an upgrade that ADDS an event type — a harness that
 * starts announcing, say, a delegated sub-agent, or a permission request, would
 * otherwise fall into `default` and be dropped in silence for as long as it took
 * somebody to notice the product had lost a feature it never knew it had.
 *
 * A few of these are worth knowing by name. `bash_execution_update` exists
 * because the harness can run a shell; Piggy cannot, and if this event ever
 * arrives on a Piggy session something is very wrong. `compaction_start` and
 * `compaction_end` are the harness rewriting its own transcript, which PIG
 * neither triggers nor persists — conversations are stored as PIG messages in
 * Postgres. `auto_retry_start` is a retry PIG does not surface because the user
 * is watching a spinner either way.
 */
const DROPPED_EVENTS = [
  'agent_start',
  'agent_end',
  'agent_settled',
  'turn_start',
  'message_start',
  'message_end',
  'tool_execution_update',
  'queue_update',
  'compaction_start',
  'compaction_end',
  'entry_appended',
  'session_info_changed',
  'thinking_level_changed',
  'auto_retry_start',
  'auto_retry_end',
  'summarization_retry_scheduled',
  'summarization_retry_attempt_start',
  'summarization_retry_finished',
  'bash_execution_update',
] as const;

/**
 * Both directions, at compile time.
 *
 * `[A] extends [B]` and back again is mutual assignability rather than
 * assignability one way: a harness event missing from the lists fails, and a
 * name in the lists the harness no longer emits fails too. Either way the
 * failure is `tsc`, before a single test runs.
 */
type MutuallyAssignable<A, B> = [A] extends [B] ? ([B] extends [A] ? true : never) : never;
type HarnessEventType = AgentSessionEvent['type'];
type AccountedFor = (typeof MAPPED_EVENTS)[number] | (typeof DROPPED_EVENTS)[number];
const EVENT_VOCABULARY_IS_ACCOUNTED_FOR: MutuallyAssignable<HarnessEventType, AccountedFor> = true;

test('every harness event is either rendered or deliberately dropped', async (t) => {
  // The type above is the real assertion; this keeps it from being deleted as
  // an unused declaration, and states in words what it is for.
  assert.equal(EVENT_VOCABULARY_IS_ACCOUNTED_FOR, true);
  assert.equal(
    new Set([...MAPPED_EVENTS, ...DROPPED_EVENTS]).size,
    MAPPED_EVENTS.length + DROPPED_EVENTS.length,
    'an event cannot be both rendered and dropped',
  );

  const runs: RecordedRun[] = [];
  const base = await startForTest(t, runs, {
    createSession: fakeSessions(async (_tools, emit) => {
      // Every dropped type, in one turn. None of them may reach the wire.
      for (const type of DROPPED_EVENTS) emit({ type } as unknown as AgentSessionEvent);
      emit(turnEnd(10, 5));
    }, spy()),
  });

  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', 'done'],
    'a harness event PIG does not render must not reach the browser at all',
  );
  // And the turn still completed: dropping an event is not the same as being
  // confused by one.
  assert.equal(runs[0]?.closed?.status, 'succeeded');
});

test('a failed tool is reported as failed rather than as an empty answer', async (t) => {
  const runs: RecordedRun[] = [];
  const base = await startForTest(t, runs, {
    createSession: fakeSessions(async (_tools, emit) => {
      emit(toolStart('call_1', 'pig_get_record', {}));
      emit(
        toolEnd(
          'call_1',
          'pig_get_record',
          { content: [{ type: 'text', text: 'Account 9c1 not found.' }] },
          true,
        ),
      );
      emit(turnEnd(40, 10));
    }, spy()),
  });

  const frames = parseFrames(
    await fetch(`${base}/internal/chat`, { method: 'POST', headers: authorised, body: chatBody() })
      .then((response) => response.text()),
  );
  assert.deepEqual(frames[2], {
    type: 'tool_result',
    id: 'call_1',
    name: 'pig_get_record',
    ok: false,
    error: 'Account 9c1 not found.',
  });
});

test('the chosen model is the one that answers, and is priced as itself', async (t) => {
  const runs: RecordedRun[] = [];
  const watched = spy();
  const base = await startForTest(t, runs, {
    createSession: fakeSessions(async (_tools, emit) => {
      emit(textDelta('Considered.'));
      emit(turnEnd(1_000_000, 1_000_000));
    }, watched),
  });

  const frames = parseFrames(
    await fetch(`${base}/internal/chat`, {
      method: 'POST',
      headers: authorised,
      body: chatBody({ modelId: 'anthropic/claude-opus-5' }),
    }).then((response) => response.text()),
  );

  assert.equal(watched.options?.modelId, 'anthropic/claude-opus-5');
  assert.equal(frames[0]?.type === 'meta' ? frames[0].model : null, 'anthropic/claude-opus-5');
  const done = frames.at(-1);
  // A million tokens each way at $5/$25 per million: 500 + 2500 cents, in
  // micro-cents. Billing that at the nano's price would understate it a
  // hundredfold, which is why the catalogue is the only price list.
  assert.equal(done?.type === 'done' ? done.costMicroCents : null, 3_000_000_000);

  const unknown = await fetch(`${base}/internal/chat`, {
    method: 'POST',
    headers: authorised,
    body: chatBody({ modelId: 'openai/o-something' }),
  });
  assert.equal(unknown.status, 400);
  assert.deepEqual(await unknown.json(), { error: 'Unknown Piggy model.' });
});

test('read_only withholds the write tools; confirm hands them over', async (t) => {
  const runs: RecordedRun[] = [];
  const watched = spy();
  const seen: PigWriteToolDeps[] = [];
  const base = await startForTest(t, runs, {
    createReadTools: () => [namedTool('pig_get_idle_capacity')],
    createWriteTools: (deps) => {
      seen.push(deps);
      return [namedTool('pig_log_activity')];
    },
    createSession: fakeSessions(async (_tools, emit) => emit(turnEnd(10, 10)), watched),
  });

  await fetch(`${base}/internal/chat`, { method: 'POST', headers: authorised, body: chatBody() })
    .then((response) => response.text());
  // A tool the model is never shown is a tool it cannot be talked into calling.
  assert.deepEqual(watched.options?.tools.map((tool) => tool.name), ['pig_get_idle_capacity']);
  assert.equal(seen.length, 0);

  await fetch(`${base}/internal/chat`, {
    method: 'POST',
    headers: authorised,
    body: chatBody({ mode: 'confirm' }),
  }).then((response) => response.text());
  assert.deepEqual(watched.options?.tools.map((tool) => tool.name), [
    'pig_get_idle_capacity',
    'pig_log_activity',
  ]);
  assert.equal(watched.options?.mode, 'confirm');
  // Built as the caller, never as an elevated or synthetic principal: this is
  // the identity `executeMutation` will check capabilities against.
  assert.deepEqual(seen[0]?.principal, PRINCIPAL);
  assert.equal(seen[0]?.mode, 'confirm');
});

test('a tool outside the PIG boundary never reaches the harness', async (t) => {
  const runs: RecordedRun[] = [];
  const watched = spy();
  const base = await startForTest(t, runs, {
    createReadTools: () => [namedTool('bash')],
    createSession: fakeSessions(async () => {}, watched),
  });

  const frames = parseFrames(
    await fetch(`${base}/internal/chat`, { method: 'POST', headers: authorised, body: chatBody() })
      .then((response) => response.text()),
  );

  assert.deepEqual(frames.at(-1), {
    type: 'error',
    message: 'Piggy chat failed.',
    code: 'agent_failed',
  });
  assert.equal(watched.options, undefined);
  assert.equal(runs[0]?.closed?.status, 'failed');
  assert.match(String(runs[0]?.closed?.error), /outside the PIG tool boundary/);
});

test('a malformed request is the only thing called an invalid request', async (t) => {
  const runs: RecordedRun[] = [];
  const base = await startForTest(t, runs, { createSession: fakeSessions(async () => {}, spy()) });

  for (const body of [
    JSON.stringify({ principal: PRINCIPAL, message: '', mode: 'read_only', conversationId: 'c' }),
    // The old bare `principalUserId`: a relay that has not been updated must
    // fail here rather than write as nobody in particular.
    JSON.stringify({
      principalUserId: PRINCIPAL.userId,
      message: 'Hello',
      mode: 'read_only',
      conversationId: 'c',
    }),
    // A principal the relay shaped differently is a relay that has drifted.
    chatBody({ principal: { ...PRINCIPAL, extra: true } }),
    chatBody({ principal: { ...PRINCIPAL, teams: [{ team: 'legal', role: 'lead' }] } }),
    chatBody({ mode: 'god_mode' }),
    chatBody({ conversationId: '' }),
  ]) {
    const response = await fetch(`${base}/internal/chat`, {
      method: 'POST',
      headers: authorised,
      body,
    });
    assert.equal(response.status, 400);
    assert.deepEqual(await response.json(), { error: 'Invalid Piggy chat request.' });
  }
  // No inference was attempted, so no run should have been opened for any.
  assert.equal(runs.length, 0);
});

test('a fault raised mid-turn is not blamed on the user, and closes its run', async (t) => {
  const runs: RecordedRun[] = [];
  const base = await startForTest(t, runs, {
    createSession: fakeSessions(async (_tools, emit) => {
      emit(textDelta('Idle is '));
      throw new Error('inference stream stalled for 30000ms');
    }, spy()),
  });

  const response = await fetch(`${base}/internal/chat`, {
    method: 'POST',
    headers: authorised,
    body: chatBody(),
  });

  // The stream had already begun, so the turn ends as an error frame on a 200.
  assert.equal(response.status, 200);
  const frames = parseFrames(await response.text());
  assert.deepEqual(frames.at(-1), {
    type: 'error',
    message: 'Piggy chat failed.',
    code: 'agent_failed',
  });
  assert.equal(runs[0]?.closed?.status, 'failed');
  assert.equal(runs[0]?.closed?.summary, 'Idle is');
});

test('a model that stops on its own error does not report a finished answer', async (t) => {
  const runs: RecordedRun[] = [];
  const base = await startForTest(t, runs, {
    createSession: fakeSessions(async (_tools, emit) => {
      emit(textDelta('Idle is '));
      emit(turnEnd(120, 4, 'error', 'upstream returned 502'));
    }, spy()),
  });

  const frames = parseFrames(
    await fetch(`${base}/internal/chat`, { method: 'POST', headers: authorised, body: chatBody() })
      .then((response) => response.text()),
  );

  assert.deepEqual(frames.at(-1), {
    type: 'error',
    message: 'Piggy could not finish this answer.',
    code: 'inference_failed',
  });
  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.
  // 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) => {
  // The measurement this pins: a real turn made three tool calls, was billed
  // for every model call behind them, and then the endpoint answered 429. The
  // run closed as `failed` with inputTokens, outputTokens and costMicroCents
  // all null, because the ledger was fed only from the `done` frame — which a
  // failed turn never emits. The spend panel therefore under-reported, and in
  // the reassuring direction, which is the worst way for a money figure to be
  // wrong. Two model calls land before the fault here, so a ledger that reads
  // only the last one would be wrong as well as short.
  const runs: RecordedRun[] = [];
  const base = await startForTest(t, runs, {
    createSession: fakeSessions(async (_tools, emit) => {
      emit(turnEnd(4_000, 100));
      emit(turnEnd(6_000, 300));
      throw new Error('429: rate limit reached');
    }, spy()),
  });

  await fetch(`${base}/internal/chat`, { method: 'POST', headers: authorised, body: chatBody() })
    .then((response) => response.text());

  const closed = runs[0]?.closed;
  assert.equal(closed?.status, 'failed');
  assert.equal(closed?.inputTokens, 10_000);
  assert.equal(closed?.outputTokens, 400);
  // 10,000 x $0.05/Mtok + 400 x $0.20/Mtok, in micro-cents.
  assert.equal(closed?.costMicroCents, 10_000 * 0.05 * 100 + 400 * 0.2 * 100);
});

test('an abandoned turn bills what it generated before the reader left', async (t) => {
  const runs: RecordedRun[] = [];
  const base = await startForTest(t, runs, {
    createSession: fakeSessions(async (_tools, emit) => {
      emit(turnEnd(2_000, 50));
      for (let index = 0; index < 40; index += 1) {
        await new Promise((resolve) => setTimeout(resolve, 20));
        emit(textDelta(`chunk ${index} `));
      }
    }, spy()),
  });

  const abort = new AbortController();
  setTimeout(() => abort.abort(), 120);
  await assert.rejects(
    fetch(`${base}/internal/chat`, {
      method: 'POST',
      headers: authorised,
      body: chatBody(),
      signal: abort.signal,
    }).then((response) => response.text()),
  );

  await waitFor(() => runs[0]?.closed !== undefined);
  assert.equal(runs[0]?.closed?.status, 'aborted');
  // Generated, therefore billed, therefore in the ledger — a closed tab is not
  // a refund.
  assert.equal(runs[0]?.closed?.inputTokens, 2_000);
  assert.equal(runs[0]?.closed?.outputTokens, 50);
  assert.equal(runs[0]?.closed?.costMicroCents, 2_000 * 0.05 * 100 + 50 * 0.2 * 100);
});

test('an answer cut short by the token budget says so', async (t) => {
  const runs: RecordedRun[] = [];
  const base = await startForTest(t, runs, {
    createSession: fakeSessions(async (_tools, emit) => {
      emit(textDelta('The first half of a table'));
      emit(turnEnd(900, 2_048, 'length'));
    }, spy()),
  });

  const frames = parseFrames(
    await fetch(`${base}/internal/chat`, { method: 'POST', headers: authorised, body: chatBody() })
      .then((response) => response.text()),
  );
  const done = frames.at(-1);
  assert.equal(done?.type === 'done' ? done.finishReason : null, 'length');
});

test('a reader who leaves mid-answer closes the run as abandoned, not as running', async (t) => {
  const runs: RecordedRun[] = [];
  const watched = spy();
  const base = await startForTest(t, runs, {
    createSession: fakeSessions(async (_tools, emit) => {
      for (let index = 0; index < 20; index += 1) {
        await new Promise((resolve) => setTimeout(resolve, 20));
        emit(textDelta(`chunk ${index} `));
      }
    }, watched),
  });

  const abort = new AbortController();
  setTimeout(() => abort.abort(), 80);
  await assert.rejects(
    fetch(`${base}/internal/chat`, {
      method: 'POST',
      headers: authorised,
      body: chatBody(),
      signal: abort.signal,
    }).then((response) => response.text()),
  );

  await waitFor(() => runs[0]?.closed !== undefined);
  // Without the finally this row stayed `running` for ever, and no later query
  // could tell it from a turn still in flight.
  assert.equal(runs[0]?.closed?.status, 'aborted');
  // And the session is disposed rather than left generating tokens nobody will
  // ever read.
  assert.ok(watched.disposed >= 1);
});

// ------------------------------------------------------------- the approvals

/**
 * A write tool in the shape `createPigWriteTools` builds: it proposes, waits,
 * and reports what really happened as its own tool result — so a declined
 * change cannot be summarised to the reader as a saved one.
 */
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 === 'apply' ? 'applied' : 'declined'}.` }],
          details: {
            tool: 'pig_log_activity',
            kind: 'activity',
            status: decision === 'apply' ? 'applied' : 'declined',
          },
        };
      },
    } as unknown as ToolDefinition,
  ];
}

/** Drives the write tool the way the harness would, around a real approval. */
const writingScript: TurnScript = 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', { subject: 'Capacity review' }));
  // The signal is forwarded because the harness forwards it: it is the only
  // thing that tells a tool parked on an approval that the turn has been
  // abandoned underneath it.
  const result = await tool.execute('call_1', {}, signal, undefined, undefined as never);
  emit(
    toolEnd('call_1', 'pig_log_activity', {
      content: result.content.filter(
        (part): part is { type: 'text'; text: string } => part.type === 'text',
      ),
      details: result.details,
    }),
  );
  emit(textDelta('Logged.'));
  emit(turnEnd(200, 20));
};

/**
 * Reads the stream up to the approval card, then hands back a reader for the
 * rest — because the decision has to be posted while the turn is still open,
 * which is the entire point of the rendezvous.
 */
async function readUntilApproval(
  response: Response,
): Promise<{ frames: PiggyChatEvent[]; rest: () => Promise<PiggyChatEvent[]> }> {
  const body = response.body;
  assert.ok(body, 'the turn should have streamed a body');
  const reader = body.getReader();
  const decoder = new TextDecoder();
  let buffer = '';

  const drain = (chunk: Uint8Array | undefined, into: PiggyChatEvent[]): void => {
    buffer += decoder.decode(chunk, { stream: true });
    const lines = buffer.split('\n');
    buffer = lines.pop() ?? '';
    for (const line of lines) if (line) into.push(JSON.parse(line) as PiggyChatEvent);
  };

  const frames: PiggyChatEvent[] = [];
  while (!frames.some((frame) => frame.type === 'approval_required')) {
    const { done, value } = await reader.read();
    if (done) break;
    drain(value, frames);
  }

  const rest = async (): Promise<PiggyChatEvent[]> => {
    const tail: PiggyChatEvent[] = [];
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      drain(value, tail);
    }
    return tail;
  };
  return { frames, rest };
}

test('a proposed write waits for the user, then applies once and only once', async (t) => {
  const runs: RecordedRun[] = [];
  const applied: string[] = [];
  const base = await startForTest(t, runs, {
    createWriteTools: proposingWriteTools(applied),
    createSession: fakeSessions(writingScript, spy()),
  });

  const response = await fetch(`${base}/internal/chat`, {
    method: 'POST',
    headers: authorised,
    body: chatBody({ mode: 'confirm', message: 'Log a call on Northwind.' }),
  });
  const { frames, rest } = await readUntilApproval(response);

  const asked = frames.find((frame) => frame.type === 'approval_required');
  assert.ok(asked && asked.type === 'approval_required');
  assert.equal(asked.change.summary, 'Log a call on Northwind Robotics');
  assert.ok(asked.change.id.length > 0);
  // The turn is still open, and nothing has been written yet.
  assert.equal(applied.length, 0);

  const decision = await fetch(`${base}/internal/approve`, {
    method: 'POST',
    headers: authorised,
    body: JSON.stringify({ conversationId: 'conv-1', changeId: asked.change.id, decision: 'apply' }),
  });
  assert.equal(decision.status, 202);

  const tail = await rest();
  assert.deepEqual(tail[0], {
    type: 'approval_resolved',
    changeId: asked.change.id,
    decision: 'apply',
    ok: true,
  });
  // The tool result carries the truth, so the model cannot claim a save it did
  // not make.
  const result = tail.find((frame) => frame.type === 'tool_result');
  assert.deepEqual(result?.type === 'tool_result' ? result.result : null, {
    tool: 'pig_log_activity',
    kind: 'activity',
    status: 'applied',
  });
  assert.equal(applied.length, 1);

  // A replayed decision must not apply the change a second time.
  const replay = await fetch(`${base}/internal/approve`, {
    method: 'POST',
    headers: authorised,
    body: JSON.stringify({ conversationId: 'conv-1', changeId: asked.change.id, decision: 'apply' }),
  });
  assert.equal(replay.status, 404);
  assert.equal(applied.length, 1);
  assert.equal(runs[0]?.closed?.status, 'succeeded');
  // One model call, and one that waited five minutes' worth of human time
  // without that counting against anything: the ceilings measure model work.
  assert.deepEqual(runs[0]?.closed?.result, {
    toolCalls: 1,
    approvalsRequested: 1,
    approvalsApplied: 1,
    modelCalls: 1,
    inference: { attempts: 1 },
  });
});

test("a decision on another conversation's id settles nothing", async (t) => {
  const runs: RecordedRun[] = [];
  const applied: string[] = [];
  const base = await startForTest(t, runs, {
    approvalTimeoutMs: 200,
    createWriteTools: proposingWriteTools(applied),
    createSession: fakeSessions(writingScript, spy()),
  });

  const response = await fetch(`${base}/internal/chat`, {
    method: 'POST',
    headers: authorised,
    body: chatBody({ mode: 'confirm' }),
  });
  const { frames, rest } = await readUntilApproval(response);
  const asked = frames.find((frame) => frame.type === 'approval_required');
  assert.ok(asked && asked.type === 'approval_required');

  const wrong = await fetch(`${base}/internal/approve`, {
    method: 'POST',
    headers: authorised,
    body: JSON.stringify({
      conversationId: 'conv-someone-else',
      changeId: asked.change.id,
      decision: 'apply',
    }),
  });
  assert.equal(wrong.status, 404);
  await rest();
  assert.equal(applied.length, 0);
});

test('a declined write is reported to the model as declined', async (t) => {
  const runs: RecordedRun[] = [];
  const applied: string[] = [];
  const base = await startForTest(t, runs, {
    createWriteTools: proposingWriteTools(applied),
    createSession: fakeSessions(writingScript, spy()),
  });

  const response = await fetch(`${base}/internal/chat`, {
    method: 'POST',
    headers: authorised,
    body: chatBody({ mode: 'confirm' }),
  });
  const { frames, rest } = await readUntilApproval(response);
  const asked = frames.find((frame) => frame.type === 'approval_required');
  assert.ok(asked && asked.type === 'approval_required');

  await fetch(`${base}/internal/approve`, {
    method: 'POST',
    headers: authorised,
    body: JSON.stringify({ conversationId: 'conv-1', changeId: asked.change.id, decision: 'reject' }),
  });

  const tail = await rest();
  assert.equal(applied.length, 0);
  const settled = tail.find((frame) => frame.type === 'approval_resolved');
  assert.equal(settled?.type === 'approval_resolved' ? settled.ok : null, true);
  const result = tail.find((frame) => frame.type === 'tool_result');
  assert.deepEqual(result?.type === 'tool_result' ? result.result : null, {
    tool: 'pig_log_activity',
    kind: 'activity',
    status: 'declined',
  });
  assert.deepEqual(runs[0]?.closed?.result, {
    toolCalls: 1,
    approvalsRequested: 1,
    approvalsApplied: 0,
    modelCalls: 1,
    inference: { attempts: 1 },
  });
});

test('an unanswered approval times out as a rejection rather than holding the turn open', async (t) => {
  const runs: RecordedRun[] = [];
  const applied: string[] = [];
  const base = await startForTest(t, runs, {
    approvalTimeoutMs: 60,
    createWriteTools: proposingWriteTools(applied),
    createSession: fakeSessions(writingScript, spy()),
  });

  const frames = parseFrames(
    await fetch(`${base}/internal/chat`, {
      method: 'POST',
      headers: authorised,
      body: chatBody({ mode: 'auto' }),
    }).then((response) => response.text()),
  );

  const settled = frames.find((frame) => frame.type === 'approval_resolved');
  assert.ok(settled && settled.type === 'approval_resolved');
  assert.equal(settled.decision, 'reject');
  assert.equal(settled.ok, false);
  assert.match(String(settled.error), /within five minutes/);
  // The turn finished rather than hanging on a reader who never answered, which
  // is what was holding the inference connection open.
  assert.equal(frames.at(-1)?.type, 'done');
  assert.equal(applied.length, 0);
});

/**
 * A write tool that behaves the way the real ones do when a turn is abandoned.
 *
 * `awaitDecision` in `write-tools.ts` races the pending decision against the
 * abort signal the harness passes to `execute`, precisely so a reader who closes
 * the tab does not leave a tool call — and the billed inference connection
 * behind it — parked for ever. This double mirrors that, and records which
 * decision it actually observed.
 */
function abandonableWriteTools(
  seen: PiggyApprovalDecision[],
): (deps: PigWriteToolDeps) => ToolDefinition[] {
  return ({ propose }) => [
    {
      name: 'pig_log_activity',
      async execute(_id: string, _params: unknown, signal?: AbortSignal) {
        const decision = await Promise.race<PiggyApprovalDecision>([
          propose({
            tool: 'pig_log_activity',
            kind: 'activity',
            summary: 'Log a call on Northwind Robotics',
            fields: [{ label: 'Subject', value: 'Capacity review' }],
          }),
          new Promise<PiggyApprovalDecision>((resolve) => {
            signal?.addEventListener('abort', () => resolve('reject'), { once: true });
          }),
        ]);
        seen.push(decision);
        return {
          content: [{ type: 'text', text: `The change was ${decision}ed.` }],
          details: { tool: 'pig_log_activity', kind: 'activity', status: 'declined' },
        };
      },
    } as unknown as ToolDefinition,
  ];
}

test('an abandoned turn rejects the approval it left open', async (t) => {
  const runs: RecordedRun[] = [];
  const seen: PiggyApprovalDecision[] = [];
  const base = await startForTest(t, runs, {
    // Long enough that the deadline cannot be what settles this: the abandoned
    // turn has to do it, or the assertion below is measuring the timeout.
    approvalTimeoutMs: 60_000,
    createWriteTools: abandonableWriteTools(seen),
    createSession: fakeSessions(writingScript, spy()),
  });

  const abort = new AbortController();
  const response = await fetch(`${base}/internal/chat`, {
    method: 'POST',
    headers: authorised,
    body: chatBody({ mode: 'confirm' }),
    signal: abort.signal,
  });
  const { frames } = await readUntilApproval(response);
  const asked = frames.find((frame) => frame.type === 'approval_required');
  assert.ok(asked && asked.type === 'approval_required');

  // The user closes the tab with the card still on screen.
  abort.abort();
  await waitFor(() => runs[0]?.closed !== undefined);

  assert.equal(runs[0]?.closed?.status, 'aborted');
  assert.deepEqual(seen, ['reject'], 'the tool was told the change was not approved');
  // And the card is gone from the registry rather than sitting there waiting
  // out its five minutes: an answer arriving now settles nothing.
  const late = await fetch(`${base}/internal/approve`, {
    method: 'POST',
    headers: authorised,
    body: JSON.stringify({ conversationId: 'conv-1', changeId: asked.change.id, decision: 'apply' }),
  });
  assert.equal(late.status, 404);
  assert.deepEqual(seen, ['reject'], 'a late approval cannot revive an abandoned change');
});

test('a decision for an unknown change settles nothing, and says so', async (t) => {
  const runs: RecordedRun[] = [];
  const base = await startForTest(t, runs, { createSession: fakeSessions(async () => {}, spy()) });

  const unknown = await fetch(`${base}/internal/approve`, {
    method: 'POST',
    headers: authorised,
    body: JSON.stringify({
      conversationId: 'conv-1',
      changeId: '30000000-0000-4000-8000-000000000009',
      decision: 'apply',
    }),
  });
  assert.equal(unknown.status, 404);

  const malformed = await fetch(`${base}/internal/approve`, {
    method: 'POST',
    headers: authorised,
    body: JSON.stringify({ conversationId: 'conv-1', changeId: 'nope', decision: 'apply' }),
  });
  assert.equal(malformed.status, 400);
  assert.deepEqual(await malformed.json(), { error: 'Invalid Piggy approval decision.' });
});

async function waitFor(condition: () => boolean): Promise<void> {
  for (let attempt = 0; attempt < 100; attempt += 1) {
    if (condition()) return;
    await new Promise((resolve) => setTimeout(resolve, 10));
  }
  assert.fail('the run was never closed');
}