pig/apps/api/src/services/piggy-activity.ts

/**
 * 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);
  }
}