spb/sp-wallet

1// V7 research runtime — tables and state machine of research jobs. Own tables, created here (not in db.ts): nothing of the2// money code reads them. A job is a request to an AI coding agent on ANOTHER machine; what comes back is untrusted.3import { createHash, randomBytes, timingSafeEqual } from 'node:crypto';4import { homedir } from 'node:os';5import path from 'node:path';6import { canMove, RESEARCH_FINAL, type ResearchBudgets, type ResearchState, type RunnerUsage } from '@spw/core';7import { UserError } from '../config.ts';8import { sql } from '../db.ts';910const SCHEMA = `11create table if not exists research_jobs (12  id uuid primary key, agent_id uuid references agents(id), kind text not null, mission text not null, universe jsonb not null, interval text not null,13  capital_usd double precision not null, budgets jsonb not null, tools jsonb not null,14  state text not null default 'queued', reason text,15  strategy_name text not null unique, strategy_version text not null default '1', lab_commit text,16  dataset jsonb, token_hash text, runner_id text, claimed_at timestamptz, heartbeat_at timestamptz, cancel_requested boolean not null default false,17  usage jsonb, ledger text, submission jsonb, server jsonb, report jsonb, version_id uuid,18  created_at timestamptz not null default now(), updated_at timestamptz not null default now(), finished_at timestamptz);19create index if not exists research_jobs_state on research_jobs (state, created_at);20create index if not exists research_jobs_agent on research_jobs (agent_id, created_at desc);21create table if not exists research_job_log (22  id bigserial primary key, job_id uuid not null references research_jobs(id) on delete cascade, ts timestamptz not null default now(), stream text not null, line text not null);23create index if not exists research_job_log_job on research_job_log (job_id, id);24`;25let migrated: Promise<unknown> | null = null;26export function migrateResearch(force = false): Promise<unknown> { if (force || !migrated) migrated = sql.unsafe(SCHEMA); return migrated; }2728export interface Holdout { start: string; devEnd: string; end: string; devBars: number; testBars: number }29export interface Dataset { ready: boolean; error?: string; dir?: string; source?: string | null; holdout?: Holdout }30export interface JobRow {31  id: string; agent_id: string | null; kind: 'spot' | 'perp'; mission: string; universe: string[]; interval: string; capital_usd: number; budgets: ResearchBudgets; tools: string[];32  state: ResearchState; reason: string | null; strategy_name: string; strategy_version: string; lab_commit: string | null;33  dataset: Dataset | null; token_hash: string | null; runner_id: string | null; claimed_at: Date | null; heartbeat_at: Date | null; cancel_requested: boolean;34  usage: RunnerUsage | null; ledger: string | null; submission: Record<string, unknown> | null; server: Record<string, unknown> | null; report: Record<string, unknown> | null; version_id: string | null;35  created_at: Date; updated_at: Date; finished_at: Date | null;36}37/** columns of a list (no ledger, no full report, no file bodies) */38export const jobList = () => sql`id, agent_id, kind, mission, universe, interval, capital_usd, budgets, tools, state, reason, strategy_name, strategy_version, lab_commit, runner_id, claimed_at, heartbeat_at, cancel_requested, usage, version_id, created_at, updated_at, finished_at`;3940export const loadJob = async (id: string): Promise<JobRow> => {41  const [j] = /^[0-9a-f-]{36}$/.test(id) ? await sql<JobRow[]>`select * from research_jobs where id = ${id}` : [];42  if (!j) throw new UserError('not_found', 'research job not found', 404);43  return j;44};4546/** Move a job along the only transitions that exist. Returns false when the job was not in `from` any more (someone else moved it). */47export async function move(id: string, from: ResearchState, to: ResearchState, set: { reason?: string | null } = {}): Promise<boolean> {48  if (!canMove(from, to)) throw new Error(`research job: ${from} → ${to} is not a transition`);49  const final = RESEARCH_FINAL.includes(to);50  const r = await sql`update research_jobs set state = ${to}, reason = ${set.reason ?? null}, updated_at = now(), finished_at = ${final ? sql`now()` : sql`finished_at`}, token_hash = ${final || to === 'evaluating' ? null : sql`token_hash`}51    where id = ${id} and state = ${from} returning id`;52  return r.length === 1;53}5455export const sha256Hex = (s: string): string => createHash('sha256').update(s).digest('hex');56export const newToken = (): string => randomBytes(32).toString('hex');57export const sameSecret = (a: string, b: string): boolean => { const x = createHash('sha256').update(a).digest(), y = createHash('sha256').update(b).digest(); return timingSafeEqual(x, y); };5859export const researchDataDir = (): string => path.join(process.env.SPW_DATA_DIR || path.join(homedir(), '.config', 'sp-wallet'), 'research-jobs');60export async function appendLog(jobId: string, lines: { s: string; t: string }[]): Promise<void> {61  if (!lines.length) return;62  await sql`insert into research_job_log ${sql(lines.map((l) => ({ job_id: jobId, stream: l.s, line: l.t })))}`;63}64export const sysLog = (jobId: string, text: string): Promise<void> => appendLog(jobId, [{ s: 'api', t: text.slice(0, 2000) }]).catch(() => {});65