/**
 * That a Piggy transcript belongs to exactly one person.
 *
 * The failure this suite exists to prevent is not exotic. Every statement in
 * `PiggyConversationService` carries `user_id = $me`; the day one of them does
 * not, the route above it keeps working perfectly for its author and quietly
 * starts answering for everybody else's history too, with no error anywhere.
 * So the assertions are made twice, at two different depths:
 *
 *   - against a recording driver, which runs in the default suite and pins
 *     that the predicate actually reaches SQL on every path, including the
 *     ones a fake row store would happily let through;
 *   - against a real Postgres, which is where a cascade, a unique key and a
 *     CHECK constraint are either true or not. That half needs a database and
 *     therefore names its own:
 *
 *         createdb pig_piggy_test
 *         DATABASE_URL=postgres://…/pig_piggy_test pnpm -F @pig/db run migrate
 *         PIG_TEST_DATABASE_URL=postgres://…/pig_piggy_test \
 *           pnpm -F @pig/api run test
 *
 *     A deliberately separate variable from `DATABASE_URL`: this suite writes
 *     and deletes rows, and it must be impossible to point it at a working
 *     database by inheriting the environment.
 */
import { strict as assert } from 'node:assert';
import { randomUUID } from 'node:crypto';
import { readFileSync } from 'node:fs';
import { join } from 'node:path';
import { after, describe, it } from 'node:test';
import { drizzle } from 'drizzle-orm/pg-proxy';
import { eq, inArray } from 'drizzle-orm';
import { Hono } from 'hono';
import type { Database } from '@pig/db';
import { AuthError } from '../src/lib/auth';
import { apiError, type ApiEnv } from '../src/lib/mutation';
import { createPiggyConversationRoutes } from '../src/routes/piggy-conversations';
import {
  derivePiggyTitle,
  PIGGY_TITLE_MAX,
  PIGGY_UNTITLED,
  PiggyConversationService,
  PiggyTurnRecorder,
} from '../src/services/piggy-conversations';
import { principal as makePrincipal } from './helpers/principal';

const ME = '00000000-0000-4000-8000-0000000000aa';
const SOMEONE_ELSE_CONVERSATION = '00000000-0000-4000-8000-0000000000cc';

// ------------------------------------------------------------------- titles

describe('conversation titles', () => {
  it('names a thread after the first thing said in it', () => {
    assert.equal(derivePiggyTitle('Which suppliers are idle this month?'), 'Which suppliers are idle this month?');
  });

  it('collapses a pasted block so a sidebar row stays one line', () => {
    assert.equal(derivePiggyTitle('  Log a call\n\non Northwind   Robotics '), 'Log a call on Northwind Robotics');
  });

  it('cuts on a word boundary and stays inside the budget', () => {
    const long = `${'word '.repeat(60)}end`;
    const title = derivePiggyTitle(long);
    assert.ok(title.length <= PIGGY_TITLE_MAX, `${title.length} exceeds ${PIGGY_TITLE_MAX}`);
    assert.ok(title.endsWith('…'));
    assert.ok(!title.includes('  '));
  });

  it('falls back rather than storing an empty title', () => {
    // The column has a CHECK on length > 0; an empty first message must not
    // reach it, because a constraint violation here would fail the turn.
    assert.equal(derivePiggyTitle(''), PIGGY_UNTITLED);
    assert.equal(derivePiggyTitle('   '), PIGGY_UNTITLED);
    assert.equal(derivePiggyTitle(undefined), PIGGY_UNTITLED);
  });
});

// -------------------------------------------------- the predicate reaches SQL

interface Statement {
  sql: string;
  params: unknown[];
}

/**
 * A driver that answers nothing and remembers everything.
 *
 * Empty results are the point: to this database every conversation belongs to
 * somebody else, which is exactly the state a caller reaching for another
 * person's thread is in. A method that only appears to be scoped — reading the
 * row and comparing the owner afterwards — would return it anyway; one that
 * puts the owner in the WHERE clause returns nothing, and the statements it
 * issued are here to be read.
 */
function recordingDatabase(): { db: Database; statements: Statement[] } {
  const statements: Statement[] = [];
  const base = drizzle(async (sql: string, params: unknown[]) => {
    statements.push({ sql, params });
    return { rows: [] };
  });

  const db = new Proxy(base, {
    get(target, property) {
      // The proxy driver refuses transactions outright, and `appendMessage`
      // opens one. Running the body inline is sound here because nothing in
      // this half asserts atomicity — the real-database half does.
      if (property === 'transaction') {
        return async (work: (tx: unknown) => Promise<unknown>) => work(db);
      }
      const value = Reflect.get(target, property);
      return typeof value === 'function' ? value.bind(target) : value;
    },
  }) as unknown as Database;

  return { db, statements };
}

function touching(statements: Statement[], table: string): Statement[] {
  return statements.filter((statement) => statement.sql.includes(table));
}

function assertScopedTo(statements: Statement[], userId: string, what: string): void {
  const relevant = touching(statements, 'piggy_conversations');
  assert.ok(relevant.length > 0, `${what} issued no statement against piggy_conversations`);
  for (const statement of relevant) {
    assert.ok(
      statement.sql.includes('"user_id"'),
      `${what} reached piggy_conversations without naming an owner:\n${statement.sql}`,
    );
    assert.ok(
      statement.params.includes(userId),
      `${what} did not bind the caller's own id:\n${statement.sql}\n${JSON.stringify(statement.params)}`,
    );
  }
}

describe('every path is scoped to the caller', () => {
  const me = makePrincipal({ userId: ME });

  it('lists only my conversations', async () => {
    const { db, statements } = recordingDatabase();
    await new PiggyConversationService(db).list(me);
    assertScopedTo(statements, ME, 'list');
  });

  it('reads a transcript only when it is mine', async () => {
    const { db, statements } = recordingDatabase();
    const detail = await new PiggyConversationService(db).detail(me, SOMEONE_ELSE_CONVERSATION);
    assert.equal(detail, null);
    assertScopedTo(statements, ME, 'detail');
    // Nothing was read out of the transcript itself, so an id belonging to
    // someone else cannot leak a message count, let alone a message.
    assert.equal(touching(statements, 'piggy_messages').length, 0);
  });

  it('replays history only from my own thread', async () => {
    const { db, statements } = recordingDatabase();
    assert.deepEqual(
      await new PiggyConversationService(db).promptHistory(me, SOMEONE_ELSE_CONVERSATION),
      [],
    );
    assertScopedTo(statements, ME, 'promptHistory');
    assert.equal(touching(statements, 'piggy_messages').length, 0);
  });

  it('renames with the owner in the UPDATE, not in a check afterwards', async () => {
    const { db, statements } = recordingDatabase();
    const renamed = await new PiggyConversationService(db).rename(
      me,
      SOMEONE_ELSE_CONVERSATION,
      'Mine now',
    );
    assert.equal(renamed, null);
    assertScopedTo(statements, ME, 'rename');
    assert.ok(statements.every((s) => s.sql.trimStart().toLowerCase().startsWith('update')));
  });

  it('deletes with the owner in the DELETE', async () => {
    const { db, statements } = recordingDatabase();
    assert.equal(await new PiggyConversationService(db).remove(me, SOMEONE_ELSE_CONVERSATION), false);
    assertScopedTo(statements, ME, 'remove');
    assert.ok(statements.every((s) => s.sql.trimStart().toLowerCase().startsWith('delete')));
  });

  it('writes nothing into a conversation that is not mine', async () => {
    const { db, statements } = recordingDatabase();
    const appended = await new PiggyConversationService(db).appendMessage(
      me,
      SOMEONE_ELSE_CONVERSATION,
      { role: 'user', content: 'Log a call on Northwind Robotics' },
    );
    assert.equal(appended, null);
    assertScopedTo(statements, ME, 'appendMessage');
    // The whole point: the ownership select fails closed, so no message row
    // and no timestamp bump ever reaches someone else's thread.
    assert.equal(
      statements.filter((s) => s.sql.toLowerCase().startsWith('insert')).length,
      0,
    );
  });

  /**
   * The one statement here that does not touch `piggy_conversations`, and so
   * the one the shared assertion above cannot cover. The conversation id
   * travels through a browser, so without the owner in the WHERE clause this
   * would be a way to re-point a colleague's inference spend at your own thread.
   */
  it('stamps the ledger only for the caller’s own runs', async () => {
    const { db, statements } = recordingDatabase();
    await new PiggyConversationService(db).linkAgentRuns(me, SOMEONE_ELSE_CONVERSATION);
    const relevant = touching(statements, 'agent_runs');
    assert.equal(relevant.length, 1, 'linkAgentRuns issued no statement against agent_runs');
    assert.ok(
      relevant[0]?.sql.includes('"principal_user_id"'),
      `the ledger was stamped without naming an owner:\n${relevant[0]?.sql}`,
    );
    assert.ok(relevant[0]?.params.includes(ME));
    // Idempotent by predicate rather than by a read-then-write: a run that
    // already names a conversation is never re-pointed.
    assert.ok(relevant[0]?.sql.includes('is null'));
  });

  /**
   * Administration is not a key to somebody's chat history. Everywhere else in
   * PIG `isPlatformAdmin` widens what is visible; here it must bind the
   * administrator's own id like anyone else's, because the transcript is a
   * person's half-formed questions and the audit trail lives elsewhere.
   */
  it('gives a platform admin no way past the predicate', async () => {
    const adminId = '00000000-0000-4000-8000-0000000000dd';
    const admin = makePrincipal({ userId: adminId, isPlatformAdmin: true });
    for (const run of [
      (service: PiggyConversationService) => service.detail(admin, SOMEONE_ELSE_CONVERSATION),
      (service: PiggyConversationService) => service.rename(admin, SOMEONE_ELSE_CONVERSATION, 'x'),
      (service: PiggyConversationService) => service.remove(admin, SOMEONE_ELSE_CONVERSATION),
    ]) {
      const { db, statements } = recordingDatabase();
      await run(new PiggyConversationService(db));
      assertScopedTo(statements, adminId, 'platform admin');
      assert.ok(
        statements.every((s) => !s.params.includes(ME)),
        'a platform admin reached a conversation by naming its owner',
      );
    }
  });
});

// -------------------------------------------------------------------- routes

function conversationApp(principal = makePrincipal({ userId: ME })) {
  const { db, statements } = recordingDatabase();
  const app = new Hono<ApiEnv>();
  app.use('*', async (context, next) => {
    context.set('principal', principal);
    await next();
  });
  app.route('/', createPiggyConversationRoutes(db));
  // The app's own mapping, reproduced so a 403 here means a 403 there.
  app.onError((error, c) =>
    error instanceof AuthError
      ? c.json(apiError(error.code, error.message), error.status)
      : c.json({ error: 'Internal error' }, 500),
  );
  return { app, statements };
}

describe('the routes answer for the caller only', () => {
  for (const [method, path] of [
    ['GET', `/api/piggy/conversations/${SOMEONE_ELSE_CONVERSATION}`],
    ['PATCH', `/api/piggy/conversations/${SOMEONE_ELSE_CONVERSATION}`],
    ['DELETE', `/api/piggy/conversations/${SOMEONE_ELSE_CONVERSATION}`],
  ] as const) {
    it(`answers 404 to ${method} on somebody else's conversation`, async () => {
      const { app } = conversationApp();
      const response = await app.request(path, {
        method,
        ...(method === 'PATCH'
          ? { headers: { 'content-type': 'application/json' }, body: '{"title":"Mine now"}' }
          : {}),
      });
      assert.equal(response.status, 404);
      assert.equal(((await response.json()) as { code: string }).code, 'not_found');
    });
  }

  it('answers a malformed id without asking the database', async () => {
    const { app, statements } = conversationApp();
    const response = await app.request('/api/piggy/conversations/not-a-uuid');
    assert.equal(response.status, 404);
    // Postgres raises on a non-UUID parameter, which would surface as a 500 on
    // any mistyped URL. It never gets that far.
    assert.equal(statements.length, 0);
  });

  it('refuses a read-only credential every write', async () => {
    const readOnly = makePrincipal({ userId: ME, via: 'api_key', scopes: ['read'] });
    for (const [method, path] of [
      ['POST', '/api/piggy/conversations'],
      ['PATCH', `/api/piggy/conversations/${SOMEONE_ELSE_CONVERSATION}`],
      ['DELETE', `/api/piggy/conversations/${SOMEONE_ELSE_CONVERSATION}`],
    ] as const) {
      const { app, statements } = conversationApp(readOnly);
      const response = await app.request(path, {
        method,
        headers: { 'content-type': 'application/json' },
        body: method === 'DELETE' ? undefined : '{}',
      });
      assert.equal(response.status, 403, `${method} ${path}`);
      assert.equal(((await response.json()) as { code: string }).code, 'insufficient_scope');
      assert.equal(statements.length, 0, 'a refused write still reached the database');
    }
  });
});

// ------------------------------------------------------------------- cascade

/**
 * The cascade is a property of the schema, not of any code path, so it is
 * asserted against the SQL that creates it. Without it, deleting a
 * conversation would leave its messages behind — rows nobody can reach, still
 * holding whatever the transcript said about the book.
 */
describe('the migration', () => {
  const sql = readFileSync(
    join(import.meta.dirname, '..', '..', '..', 'packages', 'db', 'migrations', '0014_piggy_conversations.sql'),
    'utf8',
  );

  it('deletes a transcript with its conversation', () => {
    assert.match(
      sql,
      /ALTER TABLE "piggy_messages" ADD CONSTRAINT "piggy_messages_conversation_id_piggy_conversations_id_fk"[\s\S]*?ON DELETE cascade/,
    );
  });

  it('deletes a conversation with its owner', () => {
    assert.match(
      sql,
      /ALTER TABLE "piggy_conversations" ADD CONSTRAINT "piggy_conversations_user_id_users_id_fk"[\s\S]*?ON DELETE cascade/,
    );
  });

  it('keeps the spend when the conversation goes', () => {
    // Cost accounting outlives the thread: the credit was burned either way.
    assert.match(
      sql,
      /ALTER TABLE "agent_runs" ADD CONSTRAINT "agent_runs_piggy_conversation_id_piggy_conversations_id_fk"[\s\S]*?ON DELETE set null/,
    );
  });
});

// ------------------------------------------------------- against a real database

const testDatabaseUrl = process.env.PIG_TEST_DATABASE_URL;

describe(
  'against a real database',
  { skip: testDatabaseUrl ? false : 'set PIG_TEST_DATABASE_URL to a scratch database' },
  async () => {
    const { createDatabase, agentRuns, piggyConversations, piggyMessages, users } = await import('@pig/db');
    const db = createDatabase({ url: testDatabaseUrl ?? '', max: 2 });
    const service = new PiggyConversationService(db);

    const owner = { userId: '' };
    const stranger = { userId: '' };

    after(async () => {
      // Users cascade to their conversations, which cascade to their
      // messages; this is also the last assertion the suite makes.
      for (const id of [owner.userId, stranger.userId]) {
        if (id) await db.delete(users).where(eq(users.id, id));
      }
      await db.$client.end();
    });

    it('creates two people to be told apart', async () => {
      const [a] = await db
        .insert(users)
        .values({ email: `piggy-owner-${randomUUID()}@example.test`, name: 'Owner' })
        .returning();
      const [b] = await db
        .insert(users)
        .values({ email: `piggy-stranger-${randomUUID()}@example.test`, name: 'Stranger' })
        .returning();
      assert.ok(a && b);
      owner.userId = a.id;
      stranger.userId = b.id;
    });

    it('names a thread from its first message and keeps the transcript in order', async () => {
      const created = await service.create(owner, { context: { type: 'page', route: '/margin' } });
      assert.equal(created.title, PIGGY_UNTITLED);

      await service.appendMessage(owner, created.id, {
        role: 'user',
        content: 'What is our worst idle block this month?',
      });
      await service.appendMessage(owner, created.id, {
        role: 'tool',
        model: 'nvidia/nemotron-3-nano-30b-a3b',
        mode: 'confirm',
        tool: {
          callId: 'call_1',
          name: 'pig_get_idle_capacity',
          arguments: { thresholdPct: 0.15 },
          result: { worst: 'Northwind H100 block' },
          ok: true,
        },
        readCapability: 'economics:read',
      });
      await service.appendMessage(owner, created.id, {
        role: 'assistant',
        content: 'Northwind Robotics, at 38 per cent idle.',
        model: 'nvidia/nemotron-3-nano-30b-a3b',
        mode: 'confirm',
        inputTokens: 2_100,
        outputTokens: 180,
        costMicroCents: 4_200,
      });

      const detail = await service.detail(makePrincipal({ userId: owner.userId }), created.id);
      assert.ok(detail);
      // The title came from the first user message, not from the placeholder.
      assert.equal(detail.title, 'What is our worst idle block this month?');
      assert.deepEqual(
        detail.messages.map((message) => [message.seq, message.role]),
        [
          [0, 'user'],
          [1, 'tool'],
          [2, 'assistant'],
        ],
      );
      // The evidence survives the reload, which is the whole claim.
      assert.equal(detail.messages[1]?.tool?.name, 'pig_get_idle_capacity');
      assert.deepEqual(detail.messages[1]?.tool?.result, { worst: 'Northwind H100 block' });
      assert.equal(detail.messages[2]?.costMicroCents, 4_200);
      assert.equal(detail.model, 'nvidia/nemotron-3-nano-30b-a3b');

      await service.remove(owner, created.id);
    });

    it('keeps an approval card settled across a reload', async () => {
      const created = await service.create(owner, { firstMessage: 'Log a call on Northwind' });
      const change = {
        id: 'change_1',
        tool: 'pig_log_activity',
        kind: 'activity',
        summary: 'Log a call on Northwind Robotics',
        fields: [{ label: 'Subject', value: 'Chased the firm quote' }],
      };
      await service.appendMessage(owner, created.id, {
        role: 'tool',
        mode: 'confirm',
        tool: { callId: 'call_2', name: 'pig_log_activity', arguments: {}, ok: true },
        approval: { change, decision: 'apply', decidedAt: new Date() },
      });

      const detail = await service.detail(makePrincipal({ userId: owner.userId }), created.id);
      assert.equal(detail?.messages[0]?.approval?.decision, 'apply');
      assert.deepEqual(detail?.messages[0]?.approval?.change, change);

      await service.remove(owner, created.id);
    });

    it('hides a conversation from everyone but its owner', async () => {
      const created = await service.create(owner, { firstMessage: 'Private question' });
      await service.appendMessage(owner, created.id, { role: 'user', content: 'Private question' });

      const asStranger = makePrincipal({ userId: stranger.userId });
      const asAdmin = makePrincipal({ userId: stranger.userId, isPlatformAdmin: true });

      assert.equal(await service.detail(asStranger, created.id), null);
      assert.equal(await service.detail(asAdmin, created.id), null);
      assert.deepEqual(await service.promptHistory(asStranger, created.id), []);
      assert.equal(await service.rename(stranger, created.id, 'Mine now'), null);
      assert.equal(await service.remove(stranger, created.id), false);
      assert.equal(await service.appendMessage(stranger, created.id, { role: 'user', content: 'x' }), null);
      assert.deepEqual(await service.list(stranger), []);

      // Every refusal above left the conversation exactly as it was.
      const detail = await service.detail(makePrincipal({ userId: owner.userId }), created.id);
      assert.equal(detail?.title, 'Private question');
      assert.equal(detail?.messages.length, 1);

      await service.remove(owner, created.id);
    });

    it('refuses the transcript to its own author once they are demoted', async () => {
      const created = await service.create(owner, { firstMessage: 'What is our margin?' });
      await service.appendMessage(owner, created.id, {
        role: 'assistant',
        content: 'Gross margin is 31 per cent.',
        readCapability: 'economics:read',
      });

      const demoted = makePrincipal({
        userId: owner.userId,
        teams: [{ team: 'demand', role: 'viewer' }],
      });
      await assert.rejects(
        () => service.detail(demoted, created.id),
        (error: unknown) => error instanceof AuthError && error.status === 403,
      );
      await assert.rejects(
        () => service.promptHistory(demoted, created.id),
        (error: unknown) => error instanceof AuthError && error.status === 403,
      );
      assert.equal(await service.readCapabilityFor(owner, created.id), 'economics:read');

      await service.remove(owner, created.id);
    });

    it('deletes the messages with the conversation, and keeps the spend', async () => {
      const created = await service.create(owner, { firstMessage: 'Doomed thread' });
      await service.appendMessage(owner, created.id, { role: 'user', content: 'Doomed thread' });
      await service.appendMessage(owner, created.id, { role: 'assistant', content: 'Quite.' });

      const [run] = await db
        .insert(agentRuns)
        .values({
          agent: 'piggy',
          principalUserId: owner.userId,
          piggyConversationId: created.id,
          costMicroCents: 4_200,
        })
        .returning();
      assert.ok(run);

      assert.equal(await service.remove(owner, created.id), true);

      const orphans = await db
        .select()
        .from(piggyMessages)
        .where(eq(piggyMessages.conversationId, created.id));
      assert.equal(orphans.length, 0, 'messages outlived their conversation');

      // The run survives with its cost and loses only the link, because the
      // credit was spent whatever became of the thread.
      const [survivor] = await db.select().from(agentRuns).where(eq(agentRuns.id, run.id));
      assert.equal(survivor?.costMicroCents, 4_200);
      assert.equal(survivor?.piggyConversationId, null);
      await db.delete(agentRuns).where(eq(agentRuns.id, run.id));
    });

    /**
     * The whole of D2, at the layer that has to be true: a turn goes in as the
     * NDJSON the agent streamed, and comes back out as a transcript with its
     * evidence attached. Driven through `PiggyTurnRecorder` against a real
     * Postgres rather than through the relay, because what is in doubt here is
     * the storage — the relay's half is asserted in piggy-chat.test.ts.
     */
    it('reopens a streamed turn complete, with the records behind the answer', async () => {
      const created = await service.create(owner, { id: randomUUID() });
      const change = {
        id: 'change_9',
        tool: 'pig_log_activity',
        kind: 'activity',
        summary: 'Log a call on Northwind Robotics',
        fields: [{ label: 'Subject', value: 'Chased the firm quote' }],
      };
      const recorder = new PiggyTurnRecorder({
        store: service,
        owner,
        conversationId: created.id,
        mode: 'confirm',
        model: 'nvidia/nemotron-3-nano-30b-a3b',
        capability: 'economics:read',
      });
      recorder.question('What is our worst idle block this month?');
      const frames = [
        { type: 'meta', model: 'anthropic/claude-opus-5', mode: 'confirm', conversationId: created.id },
        { type: 'tool_call', id: 'call_9', name: 'pig_get_idle_capacity', arguments: { thresholdPct: 0.15 } },
        { type: 'tool_result', id: 'call_9', name: 'pig_get_idle_capacity', ok: true, result: { worst: 'Northwind H100 block' } },
        { type: 'approval_required', change },
        { type: 'approval_resolved', changeId: 'change_9', decision: 'apply', ok: true },
        { type: 'content_delta', delta: 'Northwind Robotics, at 38 per cent idle.' },
        { type: 'done', inputTokens: 2_100, outputTokens: 180, costMicroCents: 4_200 },
      ];
      const bytes = new TextEncoder().encode(frames.map((f) => `${JSON.stringify(f)}\n`).join(''));
      // Split mid-frame, as a socket would.
      recorder.absorb(bytes.slice(0, 137));
      recorder.absorb(bytes.slice(137));
      await recorder.finish();

      const detail = await service.detail(makePrincipal({ userId: owner.userId }), created.id);
      assert.ok(detail);
      assert.deepEqual(
        detail.messages.map((message) => [message.seq, message.role]),
        [
          [0, 'user'],
          [1, 'tool'],
          [2, 'tool'],
          [3, 'assistant'],
        ],
      );
      assert.equal(detail.messages[0]?.content, 'What is our worst idle block this month?');
      assert.equal(detail.messages[1]?.tool?.name, 'pig_get_idle_capacity');
      assert.deepEqual(detail.messages[1]?.tool?.result, { worst: 'Northwind H100 block' });
      assert.equal(detail.messages[2]?.approval?.decision, 'apply');
      assert.deepEqual(detail.messages[2]?.approval?.change, change);
      assert.equal(detail.messages[3]?.content, 'Northwind Robotics, at 38 per cent idle.');
      assert.equal(detail.messages[3]?.costMicroCents, 4_200);
      // Which model ANSWERED, from `meta` rather than from what was asked for.
      assert.equal(detail.model, 'anthropic/claude-opus-5');
      // The turn read the cost book, so the thread now needs that capability.
      assert.equal(await service.readCapabilityFor(owner, created.id), 'economics:read');
      // And the next turn replays the words without the payloads.
      assert.deepEqual(await service.promptHistory(makePrincipal({ userId: owner.userId }), created.id), [
        { role: 'user', content: 'What is our worst idle block this month?' },
        { role: 'assistant', content: 'Northwind Robotics, at 38 per cent idle.' },
      ]);

      await service.remove(owner, created.id);
    });

    it('opens a conversation under the id the turn is already running with', async () => {
      // The relay settles the id before the store is consulted, because an
      // approval posted mid-turn travels with it.
      const id = randomUUID();
      const created = await service.create(owner, { id, firstMessage: 'Keep my id' });
      assert.equal(created.id, id);
      // And it cannot be used to join a thread that is not the caller's: the
      // primary key refuses, which is what makes this safe to accept.
      await assert.rejects(() => service.create({ userId: stranger.userId }, { id }));
      await service.remove(owner, id);
    });

    it('points this thread’s spend at it, and nobody else’s', async () => {
      const mine = await service.create(owner, { firstMessage: 'What did this cost?' });
      const other = await service.create(owner, { firstMessage: 'A different thread' });

      const rows = await db
        .insert(agentRuns)
        .values([
          // The run this turn opened: stamped.
          { agent: 'piggy', principalUserId: owner.userId, input: { conversationId: mine.id }, costMicroCents: 4_200 },
          // A second turn in the same thread: also stamped, which is what makes
          // per-conversation spend one query rather than a JSON scan.
          { agent: 'piggy', principalUserId: owner.userId, input: { conversationId: mine.id }, costMicroCents: 1_100 },
          // Another thread of mine: untouched.
          { agent: 'piggy', principalUserId: owner.userId, input: { conversationId: other.id } },
          // Somebody else's run naming my conversation — the case the owner
          // predicate exists for, since the id travels through a browser.
          { agent: 'piggy', principalUserId: stranger.userId, input: { conversationId: mine.id } },
          // A queued task run, which carries no conversation at all.
          { agent: 'piggy', principalUserId: owner.userId, input: { surface: 'task' } },
        ])
        .returning({ id: agentRuns.id });
      assert.equal(rows.length, 5);

      await service.linkAgentRuns(owner, mine.id);

      const stamped = await db
        .select({ id: agentRuns.id, conversation: agentRuns.piggyConversationId })
        .from(agentRuns)
        .where(inArray(agentRuns.id, rows.map((row) => row.id)));
      // Keyed by id rather than compared positionally: an UPDATE rewrites the
      // rows it touched, and Postgres is under no obligation to hand them back
      // in insertion order afterwards.
      const byId = new Map(stamped.map((row) => [row.id, row.conversation]));
      assert.deepEqual(
        rows.map((row) => byId.get(row.id)),
        [mine.id, mine.id, null, null, null],
      );

      await db.delete(agentRuns).where(inArray(agentRuns.id, rows.map((row) => row.id)));
      await service.remove(owner, mine.id);
      await service.remove(owner, other.id);
    });

    it('takes every conversation with the person who owned it', async () => {
      const [doomed] = await db
        .insert(users)
        .values({ email: `piggy-doomed-${randomUUID()}@example.test`, name: 'Doomed' })
        .returning();
      assert.ok(doomed);
      const created = await service.create({ userId: doomed.id }, { firstMessage: 'Leaving' });
      await service.appendMessage({ userId: doomed.id }, created.id, {
        role: 'user',
        content: 'Leaving',
      });

      await db.delete(users).where(eq(users.id, doomed.id));

      const conversations = await db
        .select()
        .from(piggyConversations)
        .where(eq(piggyConversations.id, created.id));
      assert.equal(conversations.length, 0);
      const messages = await db
        .select()
        .from(piggyMessages)
        .where(eq(piggyMessages.conversationId, created.id));
      assert.equal(messages.length, 0);
    });
  },
);