Scaffold PIG and model the compute-GTM ontology
PIG is an agent-native CRM for two-sided AI-compute companies: businesses that buy GPU capacity from providers and resell it. Their business is the spread between two pipelines, which is precisely what a generic CRM cannot represent. The load-bearing decision is the `allocations` table, joining a capacity_commitment (what we bought, at a known cost) to a demand_deal (what we sold, at a known price). Margin, utilisation and idle capacity all fall out of that one join. Cost is charged against the full commitment rather than only the hours that sold, because unsold hours are already paid for and any other treatment flatters a block that is losing money. Domain decisions worth noting, each grounded in how this market operates: - Demand stages put `legal` second, not last. Customers do not hand workloads to an infrastructure provider before paper is executed. - Supply qualification splits technical from financial diligence, recorded attributably. Accepting capacity is a two-key decision. - Capacity carries a time SHAPE (intervals + quantities), not a window. Commitments ramp and step down; a rectangle reports availability that does not exist in the month someone wants it. - SLAs model three distinct shapes: none, a reliability tier plus credits policy, and a negotiated agreement. Aggregators generally cannot promise uptime on resold capacity, but negotiate heavyweight paper upstream. Remedies include fee abatement, which is materially better than a capped credit and is not expressible as one. - Export control is a predicate on the allocation edge, evaluated against the ULTIMATE parent's jurisdiction. Country of incorporation is not a valid key, so this cannot live as a flag on an account. - Agent-derived claims land in `facts` with a confidence band and evidence. Only verified claims self-apply; weaker ones await review. - The API never calls the agent. It writes to a leased queue, guarded by a partial unique index on unfinished work. Verified: typechecks clean, migration generates and applies to Postgres 16 (31 tables, 24 enums, 117 indexes). Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,231 @@
|
||||
/**
|
||||
* The agent layer.
|
||||
*
|
||||
* Three ideas here, all of which exist to make an agent safe to point at a
|
||||
* database that people make commercial decisions from.
|
||||
*
|
||||
* 1. **The API never calls the agent.** It writes a row to `agentTasks`. A
|
||||
* worker leases rows and drains them. The queue therefore survives the agent
|
||||
* being offline, a restart replays nothing that already finished, and no
|
||||
* request thread is ever blocked on a model.
|
||||
*
|
||||
* 2. **Every derived claim carries its evidence.** Enrichment writes to `facts`
|
||||
* with a score, a band, a source URL and a status — not directly to the
|
||||
* record. Strong signals apply themselves; weak ones queue for a human. An
|
||||
* agent permitted to write unattributed claims will eventually write a wrong
|
||||
* one, and nobody will be able to tell which.
|
||||
*
|
||||
* 3. **Every outward action is idempotent.** `agentActions` carries a unique
|
||||
* idempotency key, so a retried task cannot send the same message twice.
|
||||
*
|
||||
* The pattern is adapted from Comp AI CRM (MIT); see NOTICE.
|
||||
*/
|
||||
import {
|
||||
index,
|
||||
integer,
|
||||
jsonb,
|
||||
numeric,
|
||||
pgTable,
|
||||
text,
|
||||
timestamp,
|
||||
uniqueIndex,
|
||||
uuid,
|
||||
} from 'drizzle-orm/pg-core';
|
||||
import { sql } from 'drizzle-orm';
|
||||
import {
|
||||
agentTaskKindEnum,
|
||||
agentTaskOutcomeEnum,
|
||||
factBandEnum,
|
||||
factStatusEnum,
|
||||
} from './enums';
|
||||
import { accounts, contacts } from './crm';
|
||||
import { users } from './identity';
|
||||
|
||||
/**
|
||||
* The work queue. Leased rather than locked: a worker claims a row by stamping
|
||||
* `leasedUntil` into the future, and a crashed worker's rows become claimable
|
||||
* again when the lease lapses. No dead-letter babysitting, no stuck jobs.
|
||||
*/
|
||||
export const agentTasks = pgTable(
|
||||
'agent_tasks',
|
||||
{
|
||||
id: uuid('id').primaryKey().defaultRandom(),
|
||||
kind: agentTaskKindEnum('kind').notNull(),
|
||||
|
||||
/**
|
||||
* What the task is about — an account id, a contact id, a commitment id.
|
||||
* Free text because the referent varies by kind, and a foreign key per kind
|
||||
* would mean a column per kind.
|
||||
*/
|
||||
subject: text('subject').notNull(),
|
||||
/** Why this was queued. Shown to the user; keeps agent work explicable. */
|
||||
reason: text('reason'),
|
||||
payload: jsonb('payload').$type<Record<string, unknown>>(),
|
||||
|
||||
/** Higher runs first. */
|
||||
priority: integer('priority').notNull().default(0),
|
||||
/** Model calls this task may spend before giving up. */
|
||||
budget: integer('budget').notNull().default(4),
|
||||
attempts: integer('attempts').notNull().default(0),
|
||||
maxAttempts: integer('max_attempts').notNull().default(3),
|
||||
|
||||
/** Not eligible to run before this time. Used for backoff and scheduling. */
|
||||
dueAt: timestamp('due_at', { withTimezone: true }).notNull().defaultNow(),
|
||||
/** Claimed until. Null means unclaimed; past means the claim has lapsed. */
|
||||
leasedUntil: timestamp('leased_until', { withTimezone: true }),
|
||||
leasedBy: text('leased_by'),
|
||||
|
||||
startedAt: timestamp('started_at', { withTimezone: true }),
|
||||
finishedAt: timestamp('finished_at', { withTimezone: true }),
|
||||
outcome: agentTaskOutcomeEnum('outcome'),
|
||||
error: text('error'),
|
||||
|
||||
requestedByUserId: uuid('requested_by_user_id').references(() => users.id, {
|
||||
onDelete: 'set null',
|
||||
}),
|
||||
|
||||
createdAt: timestamp('created_at', { withTimezone: true }).notNull().defaultNow(),
|
||||
},
|
||||
(t) => [
|
||||
/**
|
||||
* At most one *outstanding* task per (kind, subject).
|
||||
*
|
||||
* Partial on `finished_at IS NULL`, so re-enriching an account next month is
|
||||
* fine while queueing the same enrichment twice today is not. Without this,
|
||||
* a UI that queues work on page view will melt the agent.
|
||||
*/
|
||||
uniqueIndex('agent_tasks_pending_key')
|
||||
.on(t.kind, t.subject)
|
||||
.where(sql`${t.finishedAt} IS NULL`),
|
||||
index('agent_tasks_claimable_idx').on(t.dueAt, t.priority),
|
||||
index('agent_tasks_lease_idx').on(t.leasedUntil),
|
||||
],
|
||||
);
|
||||
|
||||
/**
|
||||
* Evidence-bearing claims produced by the agent.
|
||||
*
|
||||
* A fact never overwrites a record directly. It records what was claimed, about
|
||||
* which field, with what confidence, on what evidence — and only then, if the
|
||||
* band is `verified`, is it applied. Everything weaker waits for a person.
|
||||
*/
|
||||
export const facts = pgTable(
|
||||
'facts',
|
||||
{
|
||||
id: uuid('id').primaryKey().defaultRandom(),
|
||||
|
||||
accountId: uuid('account_id').references(() => accounts.id, { onDelete: 'cascade' }),
|
||||
contactId: uuid('contact_id').references(() => contacts.id, { onDelete: 'cascade' }),
|
||||
|
||||
/** The field being claimed about, e.g. `title`, `supplierType`, `gpuCount`. */
|
||||
field: text('field').notNull(),
|
||||
value: text('value').notNull(),
|
||||
|
||||
/** 0–1. Mapped to a band by `bandForScore` in @pig/core. */
|
||||
score: numeric('score', { precision: 4, scale: 3 }).notNull(),
|
||||
band: factBandEnum('band').notNull(),
|
||||
status: factStatusEnum('status').notNull().default('proposed'),
|
||||
|
||||
/** What the claim rests on: quotes, URLs, the reasoning that produced it. */
|
||||
evidence: jsonb('evidence').$type<Record<string, unknown>>(),
|
||||
sourceUrl: text('source_url'),
|
||||
/** How it was derived — 'web_search', 'prime_api', 'inference'. */
|
||||
method: text('method'),
|
||||
|
||||
/** The agent run that produced this, for tracing back. */
|
||||
agentRunId: uuid('agent_run_id'),
|
||||
|
||||
decidedByUserId: uuid('decided_by_user_id').references(() => users.id, {
|
||||
onDelete: 'set null',
|
||||
}),
|
||||
decidedAt: timestamp('decided_at', { withTimezone: true }),
|
||||
|
||||
observedAt: timestamp('observed_at', { withTimezone: true }).notNull().defaultNow(),
|
||||
supersededAt: timestamp('superseded_at', { withTimezone: true }),
|
||||
createdAt: timestamp('created_at', { withTimezone: true }).notNull().defaultNow(),
|
||||
},
|
||||
(t) => [
|
||||
index('facts_account_idx').on(t.accountId),
|
||||
index('facts_contact_idx').on(t.contactId),
|
||||
index('facts_status_idx').on(t.status),
|
||||
index('facts_field_idx').on(t.field),
|
||||
],
|
||||
);
|
||||
|
||||
/** One execution of the agent, for cost accounting and debugging. */
|
||||
export const agentRuns = pgTable(
|
||||
'agent_runs',
|
||||
{
|
||||
id: uuid('id').primaryKey().defaultRandom(),
|
||||
agentTaskId: uuid('agent_task_id').references(() => agentTasks.id, {
|
||||
onDelete: 'set null',
|
||||
}),
|
||||
|
||||
/** Which agent — 'piggy', or a user's own connected client. */
|
||||
agent: text('agent').notNull().default('piggy'),
|
||||
/** The person on whose behalf this ran. */
|
||||
principalUserId: uuid('principal_user_id').references(() => users.id, {
|
||||
onDelete: 'set null',
|
||||
}),
|
||||
|
||||
status: text('status').notNull().default('running'),
|
||||
model: text('model'),
|
||||
inputTokens: integer('input_tokens'),
|
||||
outputTokens: integer('output_tokens'),
|
||||
costMicroCents: integer('cost_micro_cents'),
|
||||
|
||||
input: jsonb('input').$type<Record<string, unknown>>(),
|
||||
result: jsonb('result').$type<Record<string, unknown>>(),
|
||||
summary: text('summary'),
|
||||
error: text('error'),
|
||||
|
||||
startedAt: timestamp('started_at', { withTimezone: true }).notNull().defaultNow(),
|
||||
finishedAt: timestamp('finished_at', { withTimezone: true }),
|
||||
},
|
||||
(t) => [
|
||||
index('agent_runs_task_idx').on(t.agentTaskId),
|
||||
index('agent_runs_principal_idx').on(t.principalUserId),
|
||||
],
|
||||
);
|
||||
|
||||
/**
|
||||
* Side effects the agent performed. Every row that touches the world outside
|
||||
* PIG lands here first, keyed by an idempotency key, so a retry is a no-op
|
||||
* rather than a second message to a customer.
|
||||
*/
|
||||
export const agentActions = pgTable(
|
||||
'agent_actions',
|
||||
{
|
||||
id: uuid('id').primaryKey().defaultRandom(),
|
||||
agentRunId: uuid('agent_run_id').references(() => agentRuns.id, { onDelete: 'cascade' }),
|
||||
|
||||
/** 'update_record' | 'send_slack' | 'post_buzz' | 'create_task' */
|
||||
type: text('type').notNull(),
|
||||
targetType: text('target_type'),
|
||||
targetId: text('target_id'),
|
||||
summary: text('summary'),
|
||||
|
||||
/**
|
||||
* Unique. This is the whole safety mechanism: an action is attempted at
|
||||
* most once, no matter how many times its task is retried.
|
||||
*/
|
||||
idempotencyKey: text('idempotency_key').notNull(),
|
||||
|
||||
status: text('status').notNull().default('pending'),
|
||||
externalId: text('external_id'),
|
||||
error: text('error'),
|
||||
metadata: jsonb('metadata').$type<Record<string, unknown>>(),
|
||||
|
||||
createdAt: timestamp('created_at', { withTimezone: true }).notNull().defaultNow(),
|
||||
},
|
||||
(t) => [
|
||||
uniqueIndex('agent_actions_idempotency_key').on(t.idempotencyKey),
|
||||
index('agent_actions_run_idx').on(t.agentRunId),
|
||||
],
|
||||
);
|
||||
|
||||
export type AgentTask = typeof agentTasks.$inferSelect;
|
||||
export type NewAgentTask = typeof agentTasks.$inferInsert;
|
||||
export type Fact = typeof facts.$inferSelect;
|
||||
export type NewFact = typeof facts.$inferInsert;
|
||||
export type AgentRun = typeof agentRuns.$inferSelect;
|
||||
Reference in New Issue
Block a user