Merge gitea/main into the Motion branch
Motion was written against a base five commits behind main, so the integration is the interesting part of this commit: - The migration is renumbered 0014 -> 0015. Main shipped 0014_piggy_conversations, and two migrations sharing an index is a journal that applies one of them. - The seed-idempotency gate keeps main's all-tables diff rather than the motion_templates counter this branch added; the general check subsumes the specific one. - Nav gains a Motion group alongside main's new Workspace group, and Piggy keeps the mark main gave it. - Stat keeps main's container-scaled figure, which already carries the min-w-0 this branch added for the same reason. - Piggy's page labels keep main's refusal wording for the four pages with no tool of their own, and gain the three Motion routes. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,446 @@
|
||||
/**
|
||||
* The read side of the agent ledger.
|
||||
*
|
||||
* `agent_runs`, `agent_tasks` and `agent_actions` have been written to since
|
||||
* the first wave and read by nothing. This service is what makes them visible:
|
||||
* what Piggy has done, what is still queued, and what the whole thing has cost.
|
||||
* Nothing here writes.
|
||||
*
|
||||
* Three decisions are worth stating, because each of them is a place where an
|
||||
* audit surface can quietly start lying.
|
||||
*
|
||||
* **Cost is carried as an integer all the way to the browser.** The column is
|
||||
* micro-cents — millionths of a cent — because a turn costs a fraction of a
|
||||
* cent and rounding it per turn would drift. Nothing in this file divides; the
|
||||
* conversion to money happens once, in the panel, against a labelled unit. A
|
||||
* factor-of-100 error here would be the worst possible bug on this surface, so
|
||||
* the unit is spelled out in the field name at every hop.
|
||||
*
|
||||
* **Scope is a predicate, not a filter applied afterwards.** A caller sees
|
||||
* their own runs; a platform admin sees the workspace, because the ledger is
|
||||
* the audit surface and an auditor who can only see their own spend is not an
|
||||
* auditor. That is the opposite of `piggy-conversations.ts`, where an admin is
|
||||
* deliberately NOT an exception — and the two are consistent: cost and outcome
|
||||
* are the company's record, the transcript is the person's.
|
||||
*
|
||||
* **A conversation link is never handed across an ownership boundary.** An
|
||||
* admin reading the workspace ledger sees that a run happened, what it cost and
|
||||
* what it answered, but gets no doorway into somebody else's transcript. The
|
||||
* link is resolved only against conversations the caller owns.
|
||||
*/
|
||||
import { and, desc, eq, gte, inArray, isNotNull, isNull, sql } from 'drizzle-orm';
|
||||
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';
|
||||
|
||||
/** How many runs the panel lists. A ledger, not an export. */
|
||||
export const PIGGY_RUN_LIMIT = 25;
|
||||
|
||||
/** Outstanding tasks are all shown; finished ones are the recent tail. */
|
||||
export const PIGGY_TASK_LIMIT = 12;
|
||||
|
||||
/** Long enough to identify a turn in a narrow column, short enough to fit. */
|
||||
const SNIPPET_MAX = 180;
|
||||
|
||||
/**
|
||||
* Where a run came from. A queued background task and a question typed into the
|
||||
* workspace cost the same money and belong in the same ledger, but they are not
|
||||
* the same event and a reader who cannot tell them apart cannot audit either.
|
||||
*/
|
||||
export type PiggyRunKind = 'chat' | 'task';
|
||||
|
||||
export interface PiggyRunSummary {
|
||||
id: string;
|
||||
kind: PiggyRunKind;
|
||||
/** 'piggy', or a user's own connected client. */
|
||||
agent: string;
|
||||
/**
|
||||
* Left as free text rather than narrowed to a union, because the column is
|
||||
* free text: the worker and the chat relay both write it, and a status this
|
||||
* service had never heard of would be silently mislabelled by a mapping. The
|
||||
* panel styles the four known values and shows anything else as it is.
|
||||
*/
|
||||
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. */
|
||||
summary: string | null;
|
||||
error: string | null;
|
||||
inputTokens: number | null;
|
||||
outputTokens: number | null;
|
||||
/** Millionths of a cent. Divide by 100,000,000 for US dollars. */
|
||||
costMicroCents: number | null;
|
||||
startedAt: string;
|
||||
finishedAt: string | null;
|
||||
/** Null while the run is still going — the panel counts up from `startedAt`. */
|
||||
durationMs: number | null;
|
||||
/** The queued work this run drained, when it came from the queue. */
|
||||
taskKind: AgentTaskKind | null;
|
||||
/** Present only when the transcript belongs to the caller. See the header. */
|
||||
conversation: { id: string; title: string } | null;
|
||||
/**
|
||||
* Whose turn it was — populated ONLY when that is somebody other than the
|
||||
* caller, which is the only case where the answer is information. A viewer
|
||||
* scoped to their own runs would otherwise read their own name on every row,
|
||||
* and an admin reading the workspace could not tell at a glance which rows
|
||||
* were theirs.
|
||||
*/
|
||||
principal: { id: string; name: string } | null;
|
||||
}
|
||||
|
||||
/**
|
||||
* What a queued task is doing, as one word.
|
||||
*
|
||||
* Derived rather than stored: the table records timestamps and an outcome, and
|
||||
* "queued" versus "scheduled" versus "running" is a question about now. A
|
||||
* lapsed lease is deliberately reported as queued rather than running — the
|
||||
* worker holding it is gone, and a row that shows as running forever is how a
|
||||
* stuck queue hides.
|
||||
*/
|
||||
export type PiggyTaskState = 'running' | 'queued' | 'scheduled' | AgentTaskOutcome;
|
||||
|
||||
export interface PiggyTaskSummary {
|
||||
id: string;
|
||||
kind: AgentTaskKind;
|
||||
/** The account, contact or commitment id the work is about. */
|
||||
subject: string;
|
||||
/** Why it was queued. Written for a person to read. */
|
||||
reason: string | null;
|
||||
state: PiggyTaskState;
|
||||
attempts: number;
|
||||
maxAttempts: number;
|
||||
priority: number;
|
||||
/** Not eligible before this. In the future means scheduled, not late. */
|
||||
dueAt: string;
|
||||
startedAt: string | null;
|
||||
finishedAt: string | null;
|
||||
error: string | null;
|
||||
}
|
||||
|
||||
/**
|
||||
* The money question, in the unit the column stores.
|
||||
*
|
||||
* `turns` counts the month's runs, so the monthly figure can be read as an
|
||||
* average per turn without a second request. Both windows are calendar
|
||||
* boundaries in the API process's timezone, not rolling 24-hour spans: "today"
|
||||
* that silently means "since this time yesterday" is a number nobody can
|
||||
* reconcile against a provider's invoice.
|
||||
*/
|
||||
export interface PiggySpendSummary {
|
||||
todayMicroCents: number;
|
||||
monthMicroCents: number;
|
||||
turns: number;
|
||||
}
|
||||
|
||||
export interface PiggyActivityOverview {
|
||||
runs: PiggyRunSummary[];
|
||||
tasks: PiggyTaskSummary[];
|
||||
spend: PiggySpendSummary;
|
||||
}
|
||||
|
||||
/** Canonical UUID text. See `conversationIdOf` for the row this saved. */
|
||||
const UUID_PATTERN = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
|
||||
|
||||
function snippet(value: string | null | undefined): string | null {
|
||||
if (!value) return null;
|
||||
// First line only: an answer is often a table or a bulleted list, and pouring
|
||||
// the whole of it into a ledger row turns the list into a wall.
|
||||
const [first = ''] = value.trim().split('\n');
|
||||
const line = first.trim();
|
||||
if (!line) return null;
|
||||
return line.length > SNIPPET_MAX ? `${line.slice(0, SNIPPET_MAX - 1).trimEnd()}…` : line;
|
||||
}
|
||||
|
||||
function readString(bag: Record<string, unknown> | null, key: string): string | null {
|
||||
const value = bag?.[key];
|
||||
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<string, unknown> | null): PiggyMode | null {
|
||||
const claimed = readString(input, 'mode');
|
||||
return PIGGY_MODES.find((mode) => mode === claimed) ?? null;
|
||||
}
|
||||
|
||||
/**
|
||||
* The conversation a run answered.
|
||||
*
|
||||
* `agent_runs.piggy_conversation_id` is the column that means this, and the
|
||||
* chat relay does not yet populate it — it writes the id into the run's `input`
|
||||
* blob instead. Reading both keeps the panel honest today without pretending
|
||||
* the column is redundant; when the relay starts stamping it, this falls back
|
||||
* to the column and the second arm becomes dead weight worth deleting.
|
||||
*
|
||||
* The value in `input` is whatever the client sent, and a real row in this
|
||||
* database has `"conversationId": "drive-write-1"` in it, so it is validated
|
||||
* rather than cast. An unguarded `::uuid` here would take the whole endpoint
|
||||
* down with a Postgres syntax error on that one row.
|
||||
*/
|
||||
function conversationIdOf(row: {
|
||||
piggyConversationId: string | null;
|
||||
input: Record<string, unknown> | null;
|
||||
}): string | null {
|
||||
if (row.piggyConversationId) return row.piggyConversationId;
|
||||
const claimed = readString(row.input, 'conversationId');
|
||||
return claimed && UUID_PATTERN.test(claimed) ? claimed : null;
|
||||
}
|
||||
|
||||
function humaniseKind(kind: string): string {
|
||||
return kind.replaceAll('_', ' ').replace(/^./, (letter) => letter.toUpperCase());
|
||||
}
|
||||
|
||||
function taskState(row: {
|
||||
outcome: AgentTaskOutcome | null;
|
||||
finishedAt: Date | null;
|
||||
startedAt: Date | null;
|
||||
leasedUntil: Date | null;
|
||||
dueAt: Date;
|
||||
}, now: Date): PiggyTaskState {
|
||||
if (row.finishedAt || row.outcome) return row.outcome ?? 'succeeded';
|
||||
if (row.leasedUntil && row.leasedUntil > now) return 'running';
|
||||
return row.dueAt > now ? 'scheduled' : 'queued';
|
||||
}
|
||||
|
||||
export class PiggyActivityService {
|
||||
constructor(private readonly db: Database) {}
|
||||
|
||||
async overview(principal: Principal, now = new Date()): Promise<PiggyActivityOverview> {
|
||||
const [runs, tasks, spend] = await Promise.all([
|
||||
this.runs(principal),
|
||||
this.tasks(principal, now),
|
||||
this.spend(principal, now),
|
||||
]);
|
||||
return { runs, tasks, spend };
|
||||
}
|
||||
|
||||
private async runs(principal: Principal): Promise<PiggyRunSummary[]> {
|
||||
const rows = await this.db
|
||||
.select({
|
||||
id: agentRuns.id,
|
||||
agent: agentRuns.agent,
|
||||
status: agentRuns.status,
|
||||
model: agentRuns.model,
|
||||
summary: agentRuns.summary,
|
||||
error: agentRuns.error,
|
||||
input: agentRuns.input,
|
||||
inputTokens: agentRuns.inputTokens,
|
||||
outputTokens: agentRuns.outputTokens,
|
||||
costMicroCents: agentRuns.costMicroCents,
|
||||
startedAt: agentRuns.startedAt,
|
||||
finishedAt: agentRuns.finishedAt,
|
||||
agentTaskId: agentRuns.agentTaskId,
|
||||
piggyConversationId: agentRuns.piggyConversationId,
|
||||
taskKind: agentTasks.kind,
|
||||
taskSubject: agentTasks.subject,
|
||||
principalId: users.id,
|
||||
principalName: users.name,
|
||||
})
|
||||
.from(agentRuns)
|
||||
.leftJoin(agentTasks, eq(agentTasks.id, agentRuns.agentTaskId))
|
||||
.leftJoin(users, eq(users.id, agentRuns.principalUserId))
|
||||
.where(this.scope(principal))
|
||||
.orderBy(desc(agentRuns.startedAt))
|
||||
.limit(PIGGY_RUN_LIMIT);
|
||||
|
||||
const titles = await this.conversationTitles(principal, rows.map(conversationIdOf));
|
||||
|
||||
return rows.map((row) => {
|
||||
const conversationId = conversationIdOf(row);
|
||||
const title = conversationId ? titles.get(conversationId) : undefined;
|
||||
const kind: PiggyRunKind = row.agentTaskId ? 'task' : 'chat';
|
||||
const ask = snippet(readString(row.input, 'message'));
|
||||
return {
|
||||
id: row.id,
|
||||
kind,
|
||||
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:
|
||||
ask ??
|
||||
(row.taskKind ? humaniseKind(row.taskKind) : null) ??
|
||||
(kind === 'task' ? 'Queued work' : 'Untitled turn'),
|
||||
summary: snippet(row.summary),
|
||||
error: row.error,
|
||||
inputTokens: row.inputTokens,
|
||||
outputTokens: row.outputTokens,
|
||||
costMicroCents: row.costMicroCents,
|
||||
startedAt: row.startedAt.toISOString(),
|
||||
finishedAt: row.finishedAt?.toISOString() ?? null,
|
||||
durationMs: row.finishedAt
|
||||
? row.finishedAt.getTime() - row.startedAt.getTime()
|
||||
: null,
|
||||
taskKind: row.taskKind ?? null,
|
||||
conversation: conversationId && title ? { id: conversationId, title } : null,
|
||||
principal:
|
||||
row.principalId && row.principalId !== principal.userId
|
||||
? { id: row.principalId, name: row.principalName ?? 'Another member' }
|
||||
: null,
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Titles for the runs' conversations, and only for the caller's own.
|
||||
*
|
||||
* One statement for the whole page rather than a join per row, and the
|
||||
* ownership predicate is in the statement — so a run belonging to somebody
|
||||
* else simply resolves to no title, and the panel renders it without a link
|
||||
* rather than with a link that 404s.
|
||||
*/
|
||||
private async conversationTitles(
|
||||
principal: Principal,
|
||||
ids: (string | null)[],
|
||||
): Promise<Map<string, string>> {
|
||||
const wanted = [...new Set(ids.filter((id): id is string => id !== null))];
|
||||
if (wanted.length === 0) return new Map();
|
||||
|
||||
const rows = await this.db
|
||||
.select({ id: piggyConversations.id, title: piggyConversations.title })
|
||||
.from(piggyConversations)
|
||||
.where(
|
||||
and(
|
||||
inArray(piggyConversations.id, wanted),
|
||||
eq(piggyConversations.userId, principal.userId),
|
||||
),
|
||||
);
|
||||
return new Map(rows.map((row) => [row.id, row.title]));
|
||||
}
|
||||
|
||||
/**
|
||||
* Outstanding work first, then the recent tail of finished work.
|
||||
*
|
||||
* Two statements rather than one: what is queued must never be truncated by a
|
||||
* busy week of completions, and a finished-task list that grows without bound
|
||||
* is not a panel. A failed task stays in the tail with its error — hiding a
|
||||
* failure is how a queue looks healthy while nothing drains.
|
||||
*/
|
||||
private async tasks(principal: Principal, now: Date): Promise<PiggyTaskSummary[]> {
|
||||
const columns = {
|
||||
id: agentTasks.id,
|
||||
kind: agentTasks.kind,
|
||||
subject: agentTasks.subject,
|
||||
reason: agentTasks.reason,
|
||||
outcome: agentTasks.outcome,
|
||||
attempts: agentTasks.attempts,
|
||||
maxAttempts: agentTasks.maxAttempts,
|
||||
priority: agentTasks.priority,
|
||||
dueAt: agentTasks.dueAt,
|
||||
leasedUntil: agentTasks.leasedUntil,
|
||||
startedAt: agentTasks.startedAt,
|
||||
finishedAt: agentTasks.finishedAt,
|
||||
error: agentTasks.error,
|
||||
};
|
||||
const mine = principal.isPlatformAdmin
|
||||
? undefined
|
||||
: eq(agentTasks.requestedByUserId, principal.userId);
|
||||
|
||||
const [outstanding, finished] = await Promise.all([
|
||||
this.db
|
||||
.select(columns)
|
||||
.from(agentTasks)
|
||||
.where(and(isNull(agentTasks.finishedAt), mine))
|
||||
.orderBy(agentTasks.dueAt)
|
||||
.limit(PIGGY_TASK_LIMIT),
|
||||
this.db
|
||||
.select(columns)
|
||||
.from(agentTasks)
|
||||
.where(and(isNotNull(agentTasks.finishedAt), mine))
|
||||
.orderBy(desc(agentTasks.finishedAt))
|
||||
.limit(PIGGY_TASK_LIMIT),
|
||||
]);
|
||||
|
||||
return [...outstanding, ...finished].map((row) => ({
|
||||
id: row.id,
|
||||
kind: row.kind,
|
||||
subject: row.subject,
|
||||
reason: row.reason,
|
||||
state: taskState(row, now),
|
||||
attempts: row.attempts,
|
||||
maxAttempts: row.maxAttempts,
|
||||
priority: row.priority,
|
||||
dueAt: row.dueAt.toISOString(),
|
||||
startedAt: row.startedAt?.toISOString() ?? null,
|
||||
finishedAt: row.finishedAt?.toISOString() ?? null,
|
||||
error: row.error,
|
||||
}));
|
||||
}
|
||||
|
||||
/**
|
||||
* Today's and this month's spend, and the month's turn count.
|
||||
*
|
||||
* Summed as `double precision` rather than the column's `int`: a year of
|
||||
* turns overflows int4 long before it troubles a double's 2^53 of integer
|
||||
* precision, and `sum()` over a numeric would come back as a string and get
|
||||
* quietly concatenated somewhere. The result is rounded back to an integer
|
||||
* because the wire unit is micro-cents, which have no fractional part.
|
||||
*/
|
||||
private async spend(principal: Principal, now: Date): Promise<PiggySpendSummary> {
|
||||
const monthStart = new Date(now.getFullYear(), now.getMonth(), 1);
|
||||
const dayStart = new Date(now.getFullYear(), now.getMonth(), now.getDate());
|
||||
|
||||
const [row] = await this.db
|
||||
.select({
|
||||
/*
|
||||
* The boundary is bound as ISO text and cast in SQL. A raw fragment
|
||||
* hands its parameters straight to the driver with none of the column
|
||||
* mapping drizzle applies to `gte()`, and postgres.js answers a Date
|
||||
* there with `ERR_INVALID_ARG_TYPE` — a 500 on the whole panel.
|
||||
*/
|
||||
today: sql<number>`coalesce(sum(${agentRuns.costMicroCents}) filter (
|
||||
where ${agentRuns.startedAt} >= ${dayStart.toISOString()}::timestamptz
|
||||
), 0)::double precision`,
|
||||
month: sql<number>`coalesce(sum(${agentRuns.costMicroCents}), 0)::double precision`,
|
||||
turns: sql<number>`count(*)::int`,
|
||||
})
|
||||
.from(agentRuns)
|
||||
.where(and(gte(agentRuns.startedAt, monthStart), this.scope(principal)));
|
||||
|
||||
return {
|
||||
todayMicroCents: Math.round(row?.today ?? 0),
|
||||
monthMicroCents: Math.round(row?.month ?? 0),
|
||||
turns: row?.turns ?? 0,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Whose ledger this is. Undefined widens to the workspace, which drizzle's
|
||||
* `and()` treats as no predicate at all — deliberate, and the only place the
|
||||
* admin exception is expressed.
|
||||
*/
|
||||
private scope(principal: Principal) {
|
||||
return principal.isPlatformAdmin
|
||||
? undefined
|
||||
: eq(agentRuns.principalUserId, principal.userId);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,996 @@
|
||||
/**
|
||||
* Piggy's conversation store.
|
||||
*
|
||||
* The harness has its own `SessionManager` and PIG deliberately does not use it
|
||||
* for storage — the reasoning is written out on the tables themselves, in
|
||||
* `packages/db/src/schema/agent.ts`, and is worth reading before changing
|
||||
* anything here. In short: a turn gets `SessionManager.inMemory()` and the
|
||||
* history is rehydrated from Postgres, because a file under the agent
|
||||
* directory is neither per-user nor able to survive a second replica.
|
||||
*
|
||||
* Two rules hold everywhere in this file.
|
||||
*
|
||||
* **Ownership is a predicate, never a check after the fact.** Every statement
|
||||
* carries `user_id = $me`, so another person's conversation and a UUID that
|
||||
* does not exist are the same answer: nothing. Reading a row and then
|
||||
* comparing its owner would work equally well until the day someone adds a
|
||||
* path that forgets the comparison, and that path would return the row.
|
||||
*
|
||||
* **A platform admin is not an exception.** Everywhere else in PIG being an
|
||||
* administrator widens what you can see, and here it must not: a transcript is
|
||||
* a person's own half-formed questions about the book, and nobody asked to
|
||||
* have it read. Cost and audit live in `agent_runs` and `activities`, which is
|
||||
* where an administrator looks.
|
||||
*
|
||||
* Writes here do NOT go through `executeMutation`, which is otherwise the
|
||||
* chokepoint for every write in the API. That convention exists to enforce
|
||||
* capabilities and to write an audit activity, and both reasons are absent: a
|
||||
* conversation is scoped to its owner rather than to a team, and an activity
|
||||
* row per message would put "Started a Piggy conversation" into the account
|
||||
* feed and the dashboard's recent activity dozens of times a day, drowning the
|
||||
* log the convention exists to keep readable. The writes Piggy performs ON THE
|
||||
* CRM still go through `executeMutation`, as the calling user — that is a
|
||||
* different code path (`apps/piggy`), and it is the one that must stay honest.
|
||||
*/
|
||||
import { and, desc, eq, inArray, isNull, ne, sql } from 'drizzle-orm';
|
||||
import type { ReadCapability } from '@pig/core';
|
||||
import type {
|
||||
PiggyChatContext,
|
||||
PiggyChatEventType,
|
||||
PiggyConversationSummary,
|
||||
PiggyMode,
|
||||
PiggyProposedChange,
|
||||
} from '@pig/core';
|
||||
import { PIGGY_MODES } from '@pig/core';
|
||||
import type { Database, PiggyMessage, PiggyMessageRole } from '@pig/db';
|
||||
import { agentRuns, piggyConversations, piggyMessages } from '@pig/db';
|
||||
import { requireReadCapability, type Principal } from '../lib/auth';
|
||||
|
||||
/**
|
||||
* How many conversations the sidebar lists. History older than this is not
|
||||
* deleted — it simply is not a list any more, and a "load more" is cheaper to
|
||||
* add later than an unbounded query is to discover in production.
|
||||
*/
|
||||
export const PIGGY_CONVERSATION_LIST_LIMIT = 100;
|
||||
|
||||
/** How much of a thread is replayed into the next prompt. */
|
||||
export const PIGGY_PROMPT_HISTORY_LIMIT = 20;
|
||||
|
||||
/** Long enough to be a sentence, short enough for a sidebar row. */
|
||||
export const PIGGY_TITLE_MAX = 120;
|
||||
|
||||
/** What a conversation is called before anyone has said anything in it. */
|
||||
export const PIGGY_UNTITLED = 'New conversation';
|
||||
|
||||
/** The caller a statement is scoped to. A `Principal` satisfies it as it is. */
|
||||
export interface PiggyConversationOwner {
|
||||
userId: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* Only one read capability outranks the floor, and it is the one worth
|
||||
* protecting. `team:read` and `book:read` are both held by every member; a
|
||||
* transcript that touched supplier cost is the case this ranking exists for.
|
||||
*/
|
||||
const READ_CAPABILITY_RANK: Readonly<Record<ReadCapability, number>> = {
|
||||
'book:read': 0,
|
||||
'team:read': 0,
|
||||
'economics:read': 1,
|
||||
};
|
||||
|
||||
export interface PiggyToolRecord {
|
||||
callId: string;
|
||||
name: string;
|
||||
arguments?: Record<string, unknown> | null;
|
||||
result?: Record<string, unknown> | null;
|
||||
ok?: boolean | null;
|
||||
}
|
||||
|
||||
export interface PiggyApprovalRecord {
|
||||
change: PiggyProposedChange;
|
||||
/** Null while unanswered — a turn that timed out or was abandoned. */
|
||||
decision?: 'apply' | 'reject' | null;
|
||||
decidedAt?: Date | null;
|
||||
}
|
||||
|
||||
/** One transcript entry to be appended. Shape mirrors `PiggyChatEvent`. */
|
||||
export interface PiggyMessageInput {
|
||||
role: PiggyMessageRole;
|
||||
content?: string;
|
||||
reasoning?: string | null;
|
||||
model?: string | null;
|
||||
mode?: PiggyMode | null;
|
||||
inputTokens?: number | null;
|
||||
outputTokens?: number | null;
|
||||
costMicroCents?: number | null;
|
||||
finishReason?: string | null;
|
||||
tool?: PiggyToolRecord;
|
||||
approval?: PiggyApprovalRecord;
|
||||
error?: string | null;
|
||||
/**
|
||||
* Raised on the conversation when this turn read something stronger than the
|
||||
* floor. See `readCapability` on the table: without it a demotion leaves the
|
||||
* old answers readable.
|
||||
*/
|
||||
readCapability?: ReadCapability;
|
||||
}
|
||||
|
||||
/** A transcript entry as the client renders it. */
|
||||
export interface PiggyTranscriptMessage {
|
||||
id: string;
|
||||
seq: number;
|
||||
role: PiggyMessageRole;
|
||||
content: string;
|
||||
reasoning: string | null;
|
||||
model: string | null;
|
||||
mode: PiggyMode | null;
|
||||
inputTokens: number | null;
|
||||
outputTokens: number | null;
|
||||
costMicroCents: number | null;
|
||||
finishReason: string | null;
|
||||
tool: {
|
||||
callId: string;
|
||||
name: string;
|
||||
arguments: Record<string, unknown> | null;
|
||||
result: Record<string, unknown> | null;
|
||||
ok: boolean | null;
|
||||
} | null;
|
||||
approval: {
|
||||
id: string;
|
||||
change: PiggyProposedChange;
|
||||
decision: 'apply' | 'reject' | null;
|
||||
decidedAt: string | null;
|
||||
} | null;
|
||||
error: string | null;
|
||||
createdAt: string;
|
||||
}
|
||||
|
||||
export interface PiggyConversationDetail {
|
||||
id: string;
|
||||
title: string;
|
||||
/** The last turn's, so reopening restores the picker rather than the default. */
|
||||
model: string | null;
|
||||
mode: PiggyMode | null;
|
||||
context: PiggyChatContext | null;
|
||||
createdAt: string;
|
||||
/**
|
||||
* When the conversation last SAID something, matching
|
||||
* `PiggyConversationSummary.updatedAt`. A rename does not move it, so the
|
||||
* sidebar does not reorder under someone who is tidying up.
|
||||
*/
|
||||
updatedAt: string;
|
||||
messages: PiggyTranscriptMessage[];
|
||||
}
|
||||
|
||||
export interface PiggyConversationCreateInput {
|
||||
/**
|
||||
* The id to open it under, when the caller already has one to keep.
|
||||
*
|
||||
* The relay needs this. A turn's conversation id is minted before the store
|
||||
* is consulted, it is echoed to the browser on the `meta` event, and an
|
||||
* approval posted mid-turn travels with it — so a store that insisted on
|
||||
* generating its own would rename the thread underneath a card the user is
|
||||
* about to press Apply on. Omitted, the column's default mints one.
|
||||
*
|
||||
* Not a way to write into somebody else's thread: the id is the primary key,
|
||||
* so an id that is already taken fails the insert rather than joining it, and
|
||||
* the caller sees the same failure as any other unrecordable turn.
|
||||
*/
|
||||
id?: string;
|
||||
title?: string;
|
||||
/** Supplied when the conversation is opened by sending a message. */
|
||||
firstMessage?: string;
|
||||
model?: string | null;
|
||||
mode?: PiggyMode | null;
|
||||
context?: PiggyChatContext | null;
|
||||
readCapability?: ReadCapability;
|
||||
}
|
||||
|
||||
/**
|
||||
* A title from the first thing the user said.
|
||||
*
|
||||
* Deliberately not a model call: naming a conversation is not worth a round
|
||||
* trip to inference, and a title that arrives half a second after the answer
|
||||
* makes the sidebar jump. Newlines collapse because a pasted block of text
|
||||
* would otherwise become a title with a paragraph in it, and the cut lands on
|
||||
* a word boundary so the rendered row does not end mid-word.
|
||||
*/
|
||||
export function derivePiggyTitle(message: string | undefined): string {
|
||||
const collapsed = (message ?? '').replace(/\s+/g, ' ').trim();
|
||||
if (collapsed.length === 0) return PIGGY_UNTITLED;
|
||||
if (collapsed.length <= PIGGY_TITLE_MAX) return collapsed;
|
||||
// One short of the budget: the ellipsis has to fit inside it too.
|
||||
const cut = collapsed.slice(0, PIGGY_TITLE_MAX - 1);
|
||||
const lastSpace = cut.lastIndexOf(' ');
|
||||
// Below half the budget the "word" is longer than a title, so cutting on the
|
||||
// boundary would throw most of the line away. Take the hard cut instead.
|
||||
return `${(lastSpace > PIGGY_TITLE_MAX / 2 ? cut.slice(0, lastSpace) : cut).trimEnd()}…`;
|
||||
}
|
||||
|
||||
/**
|
||||
* The five methods a live turn needs from the store.
|
||||
*
|
||||
* Named as an interface so the chat relay depends on the capability rather than
|
||||
* on a Postgres-backed class: `piggy-chat.test.ts` drives the whole relay
|
||||
* against a store that records what it was told, which is the only way to
|
||||
* assert "a failed write never reaches the stream" without a database that can
|
||||
* be made to fail on demand. `PiggyConversationService` is the one production
|
||||
* implementation and says so with `implements`, so a signature that drifts here
|
||||
* stops compiling there.
|
||||
*/
|
||||
export interface PiggyTranscriptStore {
|
||||
create(
|
||||
owner: PiggyConversationOwner,
|
||||
input?: PiggyConversationCreateInput,
|
||||
): Promise<PiggyConversationDetail>;
|
||||
readCapabilityFor(owner: PiggyConversationOwner, id: string): Promise<ReadCapability | null>;
|
||||
promptHistory(
|
||||
principal: Principal,
|
||||
id: string,
|
||||
limit?: number,
|
||||
): Promise<{ role: 'user' | 'assistant'; content: string }[]>;
|
||||
appendMessage(
|
||||
owner: PiggyConversationOwner,
|
||||
conversationId: string,
|
||||
message: PiggyMessageInput,
|
||||
): Promise<PiggyTranscriptMessage | null>;
|
||||
linkAgentRuns(owner: PiggyConversationOwner, conversationId: string): Promise<void>;
|
||||
}
|
||||
|
||||
export class PiggyConversationService implements PiggyTranscriptStore {
|
||||
constructor(private readonly db: Database) {}
|
||||
|
||||
/** My conversations, most recent activity first. */
|
||||
async list(owner: PiggyConversationOwner): Promise<PiggyConversationSummary[]> {
|
||||
const rows = await this.db
|
||||
.select({
|
||||
id: piggyConversations.id,
|
||||
title: piggyConversations.title,
|
||||
lastMessageAt: piggyConversations.lastMessageAt,
|
||||
/*
|
||||
* Counted rather than kept in a column on the conversation. A stored
|
||||
* counter is one failed append away from disagreeing with the
|
||||
* transcript it describes, and this is a grouped scan of an index the
|
||||
* table already has.
|
||||
*/
|
||||
messageCount: sql<number>`count(${piggyMessages.id})::int`,
|
||||
})
|
||||
.from(piggyConversations)
|
||||
.leftJoin(piggyMessages, eq(piggyMessages.conversationId, piggyConversations.id))
|
||||
.where(eq(piggyConversations.userId, owner.userId))
|
||||
.groupBy(piggyConversations.id)
|
||||
.orderBy(desc(piggyConversations.lastMessageAt))
|
||||
.limit(PIGGY_CONVERSATION_LIST_LIMIT);
|
||||
|
||||
return rows.map((row) => ({
|
||||
id: row.id,
|
||||
title: row.title,
|
||||
// The wire's `updatedAt` is when the conversation last SAID something.
|
||||
// A rename is not activity and must not reorder somebody's history.
|
||||
updatedAt: row.lastMessageAt.toISOString(),
|
||||
messageCount: row.messageCount,
|
||||
}));
|
||||
}
|
||||
|
||||
async create(
|
||||
owner: PiggyConversationOwner,
|
||||
input: PiggyConversationCreateInput = {},
|
||||
): Promise<PiggyConversationDetail> {
|
||||
const title = input.title?.trim() ? input.title.trim() : derivePiggyTitle(input.firstMessage);
|
||||
const [created] = await this.db
|
||||
.insert(piggyConversations)
|
||||
.values({
|
||||
// Spread rather than `id: input.id ?? undefined`, so that an omitted id
|
||||
// leaves the column to its own default instead of naming it null.
|
||||
...(input.id ? { id: input.id } : {}),
|
||||
userId: owner.userId,
|
||||
title: title.slice(0, PIGGY_TITLE_MAX),
|
||||
model: input.model ?? null,
|
||||
mode: input.mode ?? null,
|
||||
context: input.context ?? null,
|
||||
readCapability: input.readCapability ?? 'book:read',
|
||||
})
|
||||
.returning();
|
||||
if (!created) throw new Error('Piggy conversation insert returned no row.');
|
||||
return { ...toDetail(created), messages: [] };
|
||||
}
|
||||
|
||||
/**
|
||||
* The whole transcript, when it is yours and you may still see what it says.
|
||||
*
|
||||
* The capability check is here rather than only in READ_RULES because a
|
||||
* path-keyed table cannot know what a particular conversation was told. A
|
||||
* person demoted out of `economics:read` keeps their history; they do not
|
||||
* keep the margin figures inside it.
|
||||
*/
|
||||
async detail(principal: Principal, id: string): Promise<PiggyConversationDetail | null> {
|
||||
const conversation = await this.own(principal, id);
|
||||
if (!conversation) return null;
|
||||
requireReadCapability(principal, conversation.readCapability);
|
||||
|
||||
const messages = await this.db
|
||||
.select()
|
||||
.from(piggyMessages)
|
||||
.where(eq(piggyMessages.conversationId, conversation.id))
|
||||
.orderBy(piggyMessages.seq);
|
||||
|
||||
return { ...toDetail(conversation), messages: messages.map(toTranscriptMessage) };
|
||||
}
|
||||
|
||||
/**
|
||||
* What the next turn replays into the prompt.
|
||||
*
|
||||
* Same gate as `detail`, and for a sharper reason: without it, a demoted
|
||||
* user could not READ yesterday's margin answer but could have it fed back
|
||||
* into a fresh prompt and read aloud to them by the model.
|
||||
*/
|
||||
async promptHistory(
|
||||
principal: Principal,
|
||||
id: string,
|
||||
limit: number = PIGGY_PROMPT_HISTORY_LIMIT,
|
||||
): Promise<{ role: 'user' | 'assistant'; content: string }[]> {
|
||||
const conversation = await this.own(principal, id);
|
||||
if (!conversation) return [];
|
||||
requireReadCapability(principal, conversation.readCapability);
|
||||
|
||||
const rows = await this.db
|
||||
.select({ role: piggyMessages.role, content: piggyMessages.content })
|
||||
.from(piggyMessages)
|
||||
.where(
|
||||
and(
|
||||
eq(piggyMessages.conversationId, conversation.id),
|
||||
// Tool rows are evidence for a reader, not context for a model: the
|
||||
// assistant text that follows already says what the tool returned,
|
||||
// and replaying the raw payloads would spend the window twice.
|
||||
inArray(piggyMessages.role, ['user', 'assistant']),
|
||||
ne(piggyMessages.content, ''),
|
||||
),
|
||||
)
|
||||
// Newest first, then reversed: the tail is what a prompt wants, and a
|
||||
// limit on an ascending scan would hand back the oldest instead.
|
||||
.orderBy(desc(piggyMessages.seq))
|
||||
.limit(limit);
|
||||
|
||||
// Narrowed rather than cast: the predicate above already excludes `tool`,
|
||||
// but the column's type does not know that and widening it by assertion is
|
||||
// how a third role would later arrive in a prompt unnoticed.
|
||||
const turns: { role: 'user' | 'assistant'; content: string }[] = [];
|
||||
for (const row of rows) {
|
||||
if (row.role === 'user' || row.role === 'assistant') {
|
||||
turns.push({ role: row.role, content: row.content });
|
||||
}
|
||||
}
|
||||
return turns.reverse();
|
||||
}
|
||||
|
||||
/**
|
||||
* The capability a conversation's contents require, or null when it is not
|
||||
* this caller's. The relay calls this before starting a turn on an existing
|
||||
* thread; `detail` and `promptHistory` enforce it themselves.
|
||||
*/
|
||||
async readCapabilityFor(
|
||||
owner: PiggyConversationOwner,
|
||||
id: string,
|
||||
): Promise<ReadCapability | null> {
|
||||
const [row] = await this.db
|
||||
.select({ readCapability: piggyConversations.readCapability })
|
||||
.from(piggyConversations)
|
||||
.where(and(eq(piggyConversations.id, id), eq(piggyConversations.userId, owner.userId)))
|
||||
.limit(1);
|
||||
return row?.readCapability ?? null;
|
||||
}
|
||||
|
||||
/** Rename. Null when the conversation is not this caller's. */
|
||||
async rename(
|
||||
owner: PiggyConversationOwner,
|
||||
id: string,
|
||||
title: string,
|
||||
): Promise<PiggyConversationDetail | null> {
|
||||
const [updated] = await this.db
|
||||
.update(piggyConversations)
|
||||
.set({ title: title.trim().slice(0, PIGGY_TITLE_MAX), updatedAt: new Date() })
|
||||
.where(and(eq(piggyConversations.id, id), eq(piggyConversations.userId, owner.userId)))
|
||||
.returning();
|
||||
return updated ? { ...toDetail(updated), messages: [] } : null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete, taking the messages with it — by the foreign key's `ON DELETE
|
||||
* CASCADE` rather than by a second statement, so a transcript can never
|
||||
* outlive the conversation that framed it.
|
||||
*/
|
||||
async remove(owner: PiggyConversationOwner, id: string): Promise<boolean> {
|
||||
const deleted = await this.db
|
||||
.delete(piggyConversations)
|
||||
.where(and(eq(piggyConversations.id, id), eq(piggyConversations.userId, owner.userId)))
|
||||
.returning({ id: piggyConversations.id });
|
||||
return deleted.length > 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Append one transcript entry.
|
||||
*
|
||||
* Everything happens in one transaction against a locked conversation row.
|
||||
* `seq` is derived from the rows already there, and two appends racing on the
|
||||
* same conversation — the stream writing an assistant delta while the
|
||||
* approval endpoint settles a card — would otherwise both read the same
|
||||
* maximum and collide on the unique key.
|
||||
*
|
||||
* Returns null when the conversation is not this caller's, which is also
|
||||
* what a deleted conversation looks like: a turn whose thread was closed
|
||||
* mid-answer writes nothing rather than resurrecting it.
|
||||
*/
|
||||
async appendMessage(
|
||||
owner: PiggyConversationOwner,
|
||||
conversationId: string,
|
||||
message: PiggyMessageInput,
|
||||
): Promise<PiggyTranscriptMessage | null> {
|
||||
return this.db.transaction(async (tx) => {
|
||||
const [conversation] = await tx
|
||||
.select()
|
||||
.from(piggyConversations)
|
||||
.where(
|
||||
and(
|
||||
eq(piggyConversations.id, conversationId),
|
||||
eq(piggyConversations.userId, owner.userId),
|
||||
),
|
||||
)
|
||||
.limit(1)
|
||||
.for('update');
|
||||
if (!conversation) return null;
|
||||
|
||||
const [tail] = await tx
|
||||
.select({ next: sql<number>`coalesce(max(${piggyMessages.seq}), -1) + 1` })
|
||||
.from(piggyMessages)
|
||||
.where(eq(piggyMessages.conversationId, conversation.id));
|
||||
const seq = tail?.next ?? 0;
|
||||
|
||||
const content = message.content ?? '';
|
||||
const [inserted] = await tx
|
||||
.insert(piggyMessages)
|
||||
.values({
|
||||
conversationId: conversation.id,
|
||||
seq,
|
||||
role: message.role,
|
||||
content,
|
||||
reasoning: message.reasoning ?? null,
|
||||
model: message.model ?? null,
|
||||
mode: message.mode ?? null,
|
||||
inputTokens: message.inputTokens ?? null,
|
||||
outputTokens: message.outputTokens ?? null,
|
||||
costMicroCents: message.costMicroCents ?? null,
|
||||
finishReason: message.finishReason ?? null,
|
||||
toolCallId: message.tool?.callId ?? null,
|
||||
toolName: message.tool?.name ?? null,
|
||||
toolArguments: message.tool?.arguments ?? null,
|
||||
toolResult: message.tool?.result ?? null,
|
||||
toolOk: message.tool?.ok ?? null,
|
||||
approvalId: message.approval?.change.id ?? null,
|
||||
approvalChange: message.approval?.change ?? null,
|
||||
approvalDecision: message.approval?.decision ?? null,
|
||||
approvalDecidedAt: message.approval?.decidedAt ?? null,
|
||||
error: message.error ?? null,
|
||||
})
|
||||
.returning();
|
||||
if (!inserted) throw new Error('Piggy message insert returned no row.');
|
||||
|
||||
const now = new Date();
|
||||
await tx
|
||||
.update(piggyConversations)
|
||||
.set({
|
||||
lastMessageAt: now,
|
||||
updatedAt: now,
|
||||
model: message.model ?? conversation.model,
|
||||
mode: message.mode ?? conversation.mode,
|
||||
readCapability: strongerCapability(
|
||||
conversation.readCapability,
|
||||
message.readCapability,
|
||||
),
|
||||
// The first thing anyone said names the thread. Only while it is
|
||||
// still unnamed: a rename must survive the next message.
|
||||
title:
|
||||
seq === 0 && message.role === 'user' && conversation.title === PIGGY_UNTITLED
|
||||
? derivePiggyTitle(content)
|
||||
: conversation.title,
|
||||
})
|
||||
.where(eq(piggyConversations.id, conversation.id));
|
||||
|
||||
return toTranscriptMessage(inserted);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Point this turn's ledger rows at the conversation they answered.
|
||||
*
|
||||
* The relay is the only hop that holds both ends. `agent_runs` is opened by
|
||||
* the agent, which knows the conversation id but writes it into the run's
|
||||
* `input` blob; the FK column beside it is what makes "everything this thread
|
||||
* cost" one indexed query instead of a JSON scan the planner cannot use.
|
||||
*
|
||||
* Stated as an UPDATE over the user's own unstamped runs rather than by run
|
||||
* id, because the relay never learns the run id — the agent mints it on the
|
||||
* far side of the hop. That shape is also what backfills the earlier turns of
|
||||
* a thread whose first attempts predate this stamping, and it is idempotent:
|
||||
* `piggy_conversation_id IS NULL` means a second call touches nothing.
|
||||
*
|
||||
* `principal_user_id = $me` is the safety predicate, not an optimisation. The
|
||||
* conversation id travels through the browser, so without it a crafted id
|
||||
* would let one member re-point another member's spend at their own thread.
|
||||
*
|
||||
* This does not go through `executeMutation` for the reason the file header
|
||||
* gives, and one more: nothing here is a claim about the book. It links two
|
||||
* rows PIG has already written to each other.
|
||||
*/
|
||||
async linkAgentRuns(owner: PiggyConversationOwner, conversationId: string): Promise<void> {
|
||||
await this.db
|
||||
.update(agentRuns)
|
||||
.set({ piggyConversationId: conversationId })
|
||||
.where(
|
||||
and(
|
||||
eq(agentRuns.principalUserId, owner.userId),
|
||||
isNull(agentRuns.piggyConversationId),
|
||||
// The agent's own record of which thread it was answering. Compared
|
||||
// as text: `input` is jsonb, and `->>` on a key that is absent is
|
||||
// NULL rather than an error, so a task run simply does not match.
|
||||
sql`${agentRuns.input}->>'conversationId' = ${conversationId}`,
|
||||
),
|
||||
);
|
||||
}
|
||||
|
||||
/** The ownership predicate every read shares. */
|
||||
private async own(owner: PiggyConversationOwner, id: string) {
|
||||
const [row] = await this.db
|
||||
.select()
|
||||
.from(piggyConversations)
|
||||
.where(and(eq(piggyConversations.id, id), eq(piggyConversations.userId, owner.userId)))
|
||||
.limit(1);
|
||||
return row ?? null;
|
||||
}
|
||||
}
|
||||
|
||||
function strongerCapability(
|
||||
current: ReadCapability,
|
||||
candidate: ReadCapability | undefined,
|
||||
): ReadCapability {
|
||||
if (!candidate) return current;
|
||||
return READ_CAPABILITY_RANK[candidate] > READ_CAPABILITY_RANK[current] ? candidate : current;
|
||||
}
|
||||
|
||||
type ConversationRow = typeof piggyConversations.$inferSelect;
|
||||
|
||||
function toDetail(row: ConversationRow): Omit<PiggyConversationDetail, 'messages'> {
|
||||
return {
|
||||
id: row.id,
|
||||
title: row.title,
|
||||
model: row.model,
|
||||
mode: row.mode,
|
||||
context: row.context ?? null,
|
||||
createdAt: row.createdAt.toISOString(),
|
||||
updatedAt: row.lastMessageAt.toISOString(),
|
||||
};
|
||||
}
|
||||
|
||||
function toTranscriptMessage(row: PiggyMessage): PiggyTranscriptMessage {
|
||||
return {
|
||||
id: row.id,
|
||||
seq: row.seq,
|
||||
role: row.role,
|
||||
content: row.content,
|
||||
reasoning: row.reasoning,
|
||||
model: row.model,
|
||||
mode: row.mode,
|
||||
inputTokens: row.inputTokens,
|
||||
outputTokens: row.outputTokens,
|
||||
costMicroCents: row.costMicroCents,
|
||||
finishReason: row.finishReason,
|
||||
// A tool call without its name is not evidence of anything, so the whole
|
||||
// record is present or absent together.
|
||||
tool:
|
||||
row.toolCallId && row.toolName
|
||||
? {
|
||||
callId: row.toolCallId,
|
||||
name: row.toolName,
|
||||
arguments: row.toolArguments ?? null,
|
||||
result: row.toolResult ?? null,
|
||||
ok: row.toolOk,
|
||||
}
|
||||
: null,
|
||||
approval: row.approvalChange
|
||||
? {
|
||||
id: row.approvalId ?? row.approvalChange.id,
|
||||
change: row.approvalChange,
|
||||
decision: row.approvalDecision,
|
||||
decidedAt: row.approvalDecidedAt?.toISOString() ?? null,
|
||||
}
|
||||
: null,
|
||||
error: row.error,
|
||||
createdAt: row.createdAt.toISOString(),
|
||||
};
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------- the live turn
|
||||
|
||||
/**
|
||||
* Every event the protocol can stream, each of which this recorder reads.
|
||||
*
|
||||
* A total record on purpose: adding an arm to `PiggyChatEvent` stops this file
|
||||
* compiling, and a new kind of transcript entry that nobody remembers to
|
||||
* persist is exactly the failure this recorder was written to end.
|
||||
*/
|
||||
const RECORDED_EVENTS: Readonly<Record<PiggyChatEventType, true>> = {
|
||||
meta: true,
|
||||
reasoning_delta: true,
|
||||
content_delta: true,
|
||||
tool_call: true,
|
||||
tool_result: true,
|
||||
approval_required: true,
|
||||
approval_resolved: true,
|
||||
done: true,
|
||||
error: true,
|
||||
};
|
||||
|
||||
export interface PiggyTurnRecorderInput {
|
||||
store: PiggyTranscriptStore;
|
||||
owner: PiggyConversationOwner;
|
||||
conversationId: string;
|
||||
/** The mode the relay authorised, until a `meta` event confirms it. */
|
||||
mode: PiggyMode;
|
||||
/** The model the relay asked for, until `meta` says which one answered. */
|
||||
model?: string | null;
|
||||
/**
|
||||
* The capability this turn's context required. Every row carries it, and the
|
||||
* conversation keeps the strongest — so a thread that asked one margin
|
||||
* question is closed to its author the day they lose `economics:read`.
|
||||
*/
|
||||
capability: ReadCapability;
|
||||
/** Where a swallowed failure goes. Injected by the tests. */
|
||||
log?: (message: string, error: unknown) => void;
|
||||
}
|
||||
|
||||
/**
|
||||
* One turn, written to the transcript as it streams.
|
||||
*
|
||||
* The relay is the only hop that sees a whole turn — the browser renders it and
|
||||
* forgets it on reload, the agent streams it and keeps nothing — so this is
|
||||
* where the record is made. It exists because `piggy_messages` was never
|
||||
* written: the sidebar listed twelve conversations against zero messages, and a
|
||||
* thread reopened the next day was a title and nothing else.
|
||||
*
|
||||
* Three rules hold in here, and each one is a bug that would otherwise be
|
||||
* shipped.
|
||||
*
|
||||
* **Nothing thrown here may reach the stream.** Every append is swallowed and
|
||||
* logged. The answer is what the user asked for; losing the filing is a
|
||||
* disappointment, losing the answer to a failed INSERT is an outage. `absorb`
|
||||
* and `observe` are therefore synchronous and total: they mutate local state
|
||||
* and enqueue, and cannot reject into the pipe loop.
|
||||
*
|
||||
* **Writes are serialised.** `appendMessage` assigns `seq` inside a transaction
|
||||
* against a locked conversation row, so racing appends cannot collide — but
|
||||
* they could still land in the wrong ORDER, and a transcript whose tool
|
||||
* evidence sorts above the question it answered is not a transcript. One
|
||||
* promise chain, appended to, keeps the order the stream had.
|
||||
*
|
||||
* **Tool rows are evidence, and evidence is written when it lands.** The
|
||||
* product's claim is that you can see the records behind an answer. A tool row
|
||||
* is flushed at its result rather than held until the end, so a turn whose
|
||||
* connection dies half-way still leaves what it read behind. The assistant's
|
||||
* text is the one row written last, because it is assembled from deltas.
|
||||
*/
|
||||
export class PiggyTurnRecorder {
|
||||
private readonly decoder = new TextDecoder();
|
||||
/** The tail of a chunk that did not end on a newline. */
|
||||
private pending = '';
|
||||
/** The serialising chain. Every append is `.then`-ed onto it. */
|
||||
private queue: Promise<void> = Promise.resolve();
|
||||
|
||||
private model: string | null;
|
||||
private mode: PiggyMode;
|
||||
private answer = '';
|
||||
private reasoning = '';
|
||||
private inputTokens: number | null = null;
|
||||
private outputTokens: number | null = null;
|
||||
private costMicroCents: number | null = null;
|
||||
private finishReason: string | null = null;
|
||||
private error: string | null = null;
|
||||
|
||||
/** Calls seen but not yet resolved, keyed by the id the protocol gave them. */
|
||||
private readonly openTools = new Map<string, PiggyToolRecord>();
|
||||
/** Changes proposed but not yet answered, keyed by change id. */
|
||||
private readonly openApprovals = new Map<string, PiggyProposedChange>();
|
||||
private closed = false;
|
||||
|
||||
constructor(private readonly input: PiggyTurnRecorderInput) {
|
||||
this.model = input.model ?? null;
|
||||
this.mode = input.mode;
|
||||
}
|
||||
|
||||
/**
|
||||
* File the question.
|
||||
*
|
||||
* Enqueued rather than awaited: the user is waiting on inference, and making
|
||||
* them wait on an INSERT first would put the database's latency in front of
|
||||
* every answer. It is also why this is called before the upstream hop rather
|
||||
* than after — a turn the agent never accepts still leaves the question in
|
||||
* the thread, with the failure recorded beneath it.
|
||||
*/
|
||||
question(content: string): void {
|
||||
this.append({ role: 'user', content });
|
||||
}
|
||||
|
||||
/**
|
||||
* A failure the relay itself saw — a dead agent, a refused hop.
|
||||
*
|
||||
* `??=` because the first failure is the true one: an error frame from the
|
||||
* agent already carries the sanitised reason, and overwriting it with the
|
||||
* transport's account of the same event loses the specific for the generic.
|
||||
*/
|
||||
fail(message: string): void {
|
||||
this.error ??= message;
|
||||
}
|
||||
|
||||
/**
|
||||
* Read one chunk of the NDJSON the agent is streaming.
|
||||
*
|
||||
* The bytes are relayed to the browser untouched; this is a second, silent
|
||||
* reader of the same chunk. Frames arrive split across chunk boundaries as a
|
||||
* matter of course, so the tail is held until its newline arrives, and the
|
||||
* decoder is told the stream continues so a multi-byte character cut in half
|
||||
* is not decoded as two question marks into somebody's transcript.
|
||||
*/
|
||||
absorb(chunk: Uint8Array): void {
|
||||
this.pending += this.decoder.decode(chunk, { stream: true });
|
||||
let newline = this.pending.indexOf('\n');
|
||||
while (newline >= 0) {
|
||||
this.line(this.pending.slice(0, newline));
|
||||
this.pending = this.pending.slice(newline + 1);
|
||||
newline = this.pending.indexOf('\n');
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Close the turn and settle everything still open.
|
||||
*
|
||||
* Idempotent, because it is called from a `finally` that a client abort also
|
||||
* runs through. Resolves once every enqueued write has settled, so the caller
|
||||
* can stamp the ledger knowing the conversation is on disk.
|
||||
*/
|
||||
async finish(): Promise<void> {
|
||||
if (this.closed) return this.queue;
|
||||
this.closed = true;
|
||||
|
||||
// A frame the agent wrote without a trailing newline. Rare, and it is
|
||||
// usually the `done` event carrying the whole turn's cost.
|
||||
if (this.pending.trim()) this.line(this.pending);
|
||||
this.pending = '';
|
||||
|
||||
/*
|
||||
* A call the stream never resolved: the turn was aborted, or the agent died
|
||||
* mid-tool. Written with `ok` left null, which the transcript renders as a
|
||||
* step with its arguments and no outcome — the honest reading. Dropping it
|
||||
* would hide that Piggy touched the book at all.
|
||||
*/
|
||||
for (const tool of this.openTools.values()) this.append({ role: 'tool', tool });
|
||||
this.openTools.clear();
|
||||
|
||||
// A proposal nobody answered. `decision: null` is what the renderer reads
|
||||
// as "the turn that offered this has ended", which beats a card that offers
|
||||
// an Apply button no agent is still listening for.
|
||||
for (const change of this.openApprovals.values()) {
|
||||
this.append({ role: 'tool', approval: { change, decision: null, decidedAt: null } });
|
||||
}
|
||||
this.openApprovals.clear();
|
||||
|
||||
if (this.answer || this.reasoning || this.error || this.hasUsage()) {
|
||||
this.append({
|
||||
role: 'assistant',
|
||||
content: this.answer,
|
||||
reasoning: this.reasoning || null,
|
||||
inputTokens: this.inputTokens,
|
||||
outputTokens: this.outputTokens,
|
||||
costMicroCents: this.costMicroCents,
|
||||
finishReason: this.finishReason,
|
||||
error: this.error,
|
||||
});
|
||||
}
|
||||
|
||||
return this.queue;
|
||||
}
|
||||
|
||||
private hasUsage(): boolean {
|
||||
return (
|
||||
this.inputTokens !== null ||
|
||||
this.outputTokens !== null ||
|
||||
this.costMicroCents !== null ||
|
||||
this.finishReason !== null
|
||||
);
|
||||
}
|
||||
|
||||
/** One NDJSON line. A frame that will not parse is dropped, never thrown. */
|
||||
private line(text: string): void {
|
||||
const trimmed = text.trim();
|
||||
if (!trimmed) return;
|
||||
let frame: unknown;
|
||||
try {
|
||||
frame = JSON.parse(trimmed);
|
||||
} catch {
|
||||
// The pipe is the product; a frame this build cannot read is not worth
|
||||
// failing a turn over, and the bytes reached the browser regardless.
|
||||
return;
|
||||
}
|
||||
if (isRecord(frame)) this.observe(frame);
|
||||
}
|
||||
|
||||
private observe(frame: Record<string, unknown>): void {
|
||||
const type = frame.type;
|
||||
if (typeof type !== 'string' || !Object.hasOwn(RECORDED_EVENTS, type)) return;
|
||||
|
||||
if (type === 'meta') {
|
||||
// Which model actually answered, which is not always the one asked for.
|
||||
this.model = asString(frame.model) ?? this.model;
|
||||
const mode = frame.mode;
|
||||
if (isMode(mode)) this.mode = mode;
|
||||
return;
|
||||
}
|
||||
if (type === 'reasoning_delta') {
|
||||
this.reasoning += asString(frame.delta) ?? '';
|
||||
return;
|
||||
}
|
||||
if (type === 'content_delta') {
|
||||
this.answer += asString(frame.delta) ?? '';
|
||||
return;
|
||||
}
|
||||
if (type === 'tool_call') {
|
||||
const callId = asString(frame.id);
|
||||
const name = asString(frame.name);
|
||||
if (!callId || !name) return;
|
||||
this.openTools.set(callId, { callId, name, arguments: asPayload(frame.arguments) });
|
||||
return;
|
||||
}
|
||||
if (type === 'tool_result') {
|
||||
const callId = asString(frame.id);
|
||||
if (!callId) return;
|
||||
const opened = this.openTools.get(callId);
|
||||
this.openTools.delete(callId);
|
||||
const ok = typeof frame.ok === 'boolean' ? frame.ok : null;
|
||||
this.append({
|
||||
role: 'tool',
|
||||
// A result whose call was never seen is still evidence. The name on the
|
||||
// result frame is what names it; without either, the row would be a
|
||||
// payload attached to nothing, and `toTranscriptMessage` drops it.
|
||||
tool: {
|
||||
callId,
|
||||
name: opened?.name ?? asString(frame.name) ?? '',
|
||||
arguments: opened?.arguments ?? null,
|
||||
result: asPayload(frame.result),
|
||||
ok,
|
||||
},
|
||||
error: ok === false ? (asString(frame.error) ?? null) : null,
|
||||
});
|
||||
return;
|
||||
}
|
||||
if (type === 'approval_required') {
|
||||
const change = asProposedChange(frame.change);
|
||||
if (change) this.openApprovals.set(change.id, change);
|
||||
return;
|
||||
}
|
||||
if (type === 'approval_resolved') {
|
||||
const changeId = asString(frame.changeId);
|
||||
const decision = frame.decision;
|
||||
if (!changeId || (decision !== 'apply' && decision !== 'reject')) return;
|
||||
const change = this.openApprovals.get(changeId);
|
||||
if (!change) return;
|
||||
this.openApprovals.delete(changeId);
|
||||
/*
|
||||
* The change and its answer share a row deliberately — see the table.
|
||||
* Written on resolution rather than on proposal, so a reload can never
|
||||
* show the offer without what the person decided about it.
|
||||
*
|
||||
* Kept separate from the tool row it belongs to, though, because that is
|
||||
* what reads back correctly: the transcript renders tool steps and
|
||||
* approval cards as two lists, and a row carrying both is folded into a
|
||||
* tool step with its card silently dropped.
|
||||
*/
|
||||
this.append({
|
||||
role: 'tool',
|
||||
approval: { change, decision, decidedAt: new Date() },
|
||||
// An approved write that failed anyway. The card says applied; without
|
||||
// this the transcript would agree with it.
|
||||
error: frame.ok === false ? (asString(frame.error) ?? 'The write did not succeed.') : null,
|
||||
});
|
||||
return;
|
||||
}
|
||||
if (type === 'done') {
|
||||
this.inputTokens = asInteger(frame.inputTokens);
|
||||
this.outputTokens = asInteger(frame.outputTokens);
|
||||
this.costMicroCents = asInteger(frame.costMicroCents);
|
||||
this.finishReason = asString(frame.finishReason);
|
||||
return;
|
||||
}
|
||||
// 'error'. Never overwritten, for the reason `fail` gives.
|
||||
this.error ??= asString(frame.message);
|
||||
}
|
||||
|
||||
/**
|
||||
* Enqueue one row, and swallow whatever it does.
|
||||
*
|
||||
* `void` on purpose: nothing upstream awaits this, and the whole point is
|
||||
* that the pipe loop cannot be made to reject by the database.
|
||||
*/
|
||||
private append(message: PiggyMessageInput): void {
|
||||
const row: PiggyMessageInput = {
|
||||
model: this.model,
|
||||
mode: this.mode,
|
||||
readCapability: this.input.capability,
|
||||
...message,
|
||||
};
|
||||
this.queue = this.queue.then(async () => {
|
||||
try {
|
||||
await this.input.store.appendMessage(this.input.owner, this.input.conversationId, row);
|
||||
} catch (error) {
|
||||
this.report(`could not append a ${row.role} message`, error);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private report(message: string, error: unknown): void {
|
||||
const log =
|
||||
this.input.log ??
|
||||
((text: string, cause: unknown) =>
|
||||
console.error(`[piggy] ${text} (${this.input.conversationId}):`, cause));
|
||||
log(message, error);
|
||||
}
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === 'object' && value !== null && !Array.isArray(value);
|
||||
}
|
||||
|
||||
function isMode(value: unknown): value is PiggyMode {
|
||||
return typeof value === 'string' && (PIGGY_MODES as readonly string[]).includes(value);
|
||||
}
|
||||
|
||||
function asString(value: unknown): string | null {
|
||||
return typeof value === 'string' && value.length > 0 ? value : null;
|
||||
}
|
||||
|
||||
/**
|
||||
* A finite integer, or null. `null` and a missing key mean the same thing here:
|
||||
* the provider reported no usage for this turn, which is not zero — a zero
|
||||
* would be added into the spend panel as a turn that cost nothing.
|
||||
*/
|
||||
function asInteger(value: unknown): number | null {
|
||||
return typeof value === 'number' && Number.isFinite(value) ? Math.round(value) : null;
|
||||
}
|
||||
|
||||
/**
|
||||
* A tool's arguments or result, in the shape the column holds.
|
||||
*
|
||||
* The column is a jsonb object and a tool may well answer with an array — the
|
||||
* pipeline list, the accounts it found. Wrapping it rather than discarding it
|
||||
* keeps the evidence a reader came for; storing null would leave a tool step
|
||||
* that says it ran and shows nothing.
|
||||
*/
|
||||
function asPayload(value: unknown): Record<string, unknown> | null {
|
||||
if (value === undefined || value === null) return null;
|
||||
return isRecord(value) ? value : { value };
|
||||
}
|
||||
|
||||
/**
|
||||
* A proposed change, validated structurally and kept whole.
|
||||
*
|
||||
* Rebuilt field by field it would be safer to type and worse as evidence: the
|
||||
* card is stored as it was SHOWN, so a field a newer agent adds has to survive
|
||||
* the trip. What is checked is what the renderer dereferences.
|
||||
*/
|
||||
function asProposedChange(value: unknown): PiggyProposedChange | null {
|
||||
if (!isRecord(value)) return null;
|
||||
if (typeof value.id !== 'string' || value.id.length === 0) return null;
|
||||
if (typeof value.tool !== 'string' || typeof value.kind !== 'string') return null;
|
||||
if (typeof value.summary !== 'string') return null;
|
||||
if (!Array.isArray(value.fields)) return null;
|
||||
const fields = value.fields.every(
|
||||
(field) => isRecord(field) && typeof field.label === 'string' && typeof field.value === 'string',
|
||||
);
|
||||
return fields ? (value as unknown as PiggyProposedChange) : null;
|
||||
}
|
||||
Reference in New Issue
Block a user