command hq: per-turn cost ledger, watcher fires as their own table
The owner asked for individual turn token/dollar costs, a running total, cost over time, and cost per trigger — derived, not stored, following the same rule `running` already uses. Two new tables carry it: - command_hq_watcher_fires: one row per trigger (sessionRowId, taskId, firedAt). Replaces the single watcherLastFiredAt column, which lost the previous fire's timestamp every time a new one landed — fine for "when did it last fire", useless for "how many times, and what did each one cost". - command_hq_turns: one row per completed turn (sessionRowId, fireId nullable, inputTokens, outputTokens, totalUsd). Nothing pre-summed — running total, cost-over-time and cost-per-trigger are all queries over this table. Turn count needed no column either: COUNT(*) already answers it, confirmed explicitly with the owner rather than assumed. command_hq_sessions gains watcherArmedAt/watcherStoppedAt in place of any "last fired" cache — set only on the false<->true transitions, so duration (current run, or the last completed one) is computable without a full transition history table. Two new read routes, GET /commanders/:sessionId/turns and .../fires, expose the raw series behind the cost summary and watcherLastFiredAt (now computed via MAX(fired_at), not read off a column — the field survives in the JSON response under the same name). Bug found and fixed while wiring this up, recorded in docs/command-hq.md: recordCommandHqWatcherFired's signature changed from (userId, sessionId) to (sessionRowId, taskId) and one call site kept the old argument order — tsgo stayed clean because number/string still matched structurally. Caught by re-reading every call site by hand, not by any tool. Needs bun db:push before any route answers with real data. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
@@ -13,6 +13,11 @@ export {
|
||||
armCommandHqWatcher,
|
||||
recordCommandHqWatcherFired,
|
||||
disarmCommandHqWatcher,
|
||||
recordCommandHqTurn,
|
||||
getCommandHqCostSummary,
|
||||
listCommandHqTurns,
|
||||
listCommandHqWatcherFires,
|
||||
getLastFiredAtBySessionRowIds,
|
||||
listCommandHqTags,
|
||||
createCommandHqTag,
|
||||
deleteCommandHqTag,
|
||||
@@ -20,4 +25,13 @@ export {
|
||||
addCommandHqSessionTag,
|
||||
removeCommandHqSessionTag,
|
||||
} from './queries';
|
||||
export type { CommandHqCategory, CommandHqSessionRow, CommandHqTag } from './queries';
|
||||
export type {
|
||||
CommandHqCategory,
|
||||
CommandHqSessionRow,
|
||||
CommandHqTag,
|
||||
CommandHqWatcherFire,
|
||||
CommandHqTurnCost,
|
||||
CommandHqCostSummary,
|
||||
CommandHqTurnRow,
|
||||
CommandHqFireRow,
|
||||
} from './queries';
|
||||
|
||||
@@ -1,6 +1,13 @@
|
||||
import { and, eq, inArray, isNull } from 'drizzle-orm';
|
||||
import { and, desc, eq, inArray, isNull, sql } from 'drizzle-orm';
|
||||
import { db } from '../db';
|
||||
import { commandHqCategories, commandHqSessionTags, commandHqSessions, commandHqTags } from './schema';
|
||||
import {
|
||||
commandHqCategories,
|
||||
commandHqSessionTags,
|
||||
commandHqSessions,
|
||||
commandHqTags,
|
||||
commandHqTurns,
|
||||
commandHqWatcherFires,
|
||||
} from './schema';
|
||||
|
||||
// ── Categories ──
|
||||
|
||||
@@ -50,7 +57,8 @@ export type CommandHqSessionRow = {
|
||||
categoryId: number | null;
|
||||
watcherArmed: boolean;
|
||||
watcherTaskId: string | null;
|
||||
watcherLastFiredAt: Date | null;
|
||||
watcherArmedAt: Date | null;
|
||||
watcherStoppedAt: Date | null;
|
||||
createdAt: Date;
|
||||
};
|
||||
|
||||
@@ -61,7 +69,8 @@ const SESSION_ROW_COLUMNS = {
|
||||
categoryId: commandHqSessions.categoryId,
|
||||
watcherArmed: commandHqSessions.watcherArmed,
|
||||
watcherTaskId: commandHqSessions.watcherTaskId,
|
||||
watcherLastFiredAt: commandHqSessions.watcherLastFiredAt,
|
||||
watcherArmedAt: commandHqSessions.watcherArmedAt,
|
||||
watcherStoppedAt: commandHqSessions.watcherStoppedAt,
|
||||
createdAt: commandHqSessions.createdAt,
|
||||
};
|
||||
|
||||
@@ -136,46 +145,175 @@ export async function getCommandHqSession(userId: number, sessionId: string): Pr
|
||||
* Arm (or re-arm) a commander's watcher — called on the first `task:started` after the dialog submits a
|
||||
* `handoff` job, and again on every SUBSEQUENT `task:started` while already armed, which is how the
|
||||
* AGENT's own self-re-arm (same turn, per `handoffContract`) is picked up: the platform does not drive
|
||||
* the re-arm, it only notices the new task id the agent's own next watch produced. Idempotent in effect:
|
||||
* arming with the same task id twice just writes the same values again.
|
||||
* the re-arm, it only notices the new task id the agent's own next watch produced.
|
||||
*
|
||||
* `watcherArmedAt` is set ONLY on the false→true transition — a `CASE` inside the same UPDATE rather than
|
||||
* a read-then-write, so there is no race between checking the current value and writing the new one. Every
|
||||
* later re-arm call leaves it alone, which is what makes it "since when has the CURRENT run been going"
|
||||
* rather than "since the last re-arm". `watcherStoppedAt` is unconditionally cleared, because arming from a
|
||||
* previously-disarmed state means a stale stop time would otherwise sit there describing a run that is no
|
||||
* longer the current one.
|
||||
*/
|
||||
export async function armCommandHqWatcher(userId: number, sessionId: string, taskId: string): Promise<boolean> {
|
||||
const rows = await db
|
||||
.update(commandHqSessions)
|
||||
.set({ watcherArmed: true, watcherTaskId: taskId })
|
||||
.set({
|
||||
watcherArmed: true,
|
||||
watcherTaskId: taskId,
|
||||
watcherArmedAt: sql`CASE WHEN ${commandHqSessions.watcherArmed} = false THEN now() ELSE ${commandHqSessions.watcherArmedAt} END`,
|
||||
watcherStoppedAt: null,
|
||||
})
|
||||
.where(and(eq(commandHqSessions.userId, userId), eq(commandHqSessions.sessionId, sessionId)))
|
||||
.returning({ id: commandHqSessions.id });
|
||||
return rows.length > 0;
|
||||
}
|
||||
|
||||
export type CommandHqWatcherFire = { id: number; firedAt: Date };
|
||||
|
||||
/**
|
||||
* The watcher fired. Records WHEN for display — does not touch `watcherArmed` or `watcherTaskId`, which
|
||||
* stay exactly as they were through the fired-then-acting window, right up until the next `armCommandHqWatcher`
|
||||
* replaces the task id on re-arm.
|
||||
* The watcher fired — a "trigger". Inserts a row in `command_hq_watcher_fires` rather than updating a
|
||||
* scalar, so cost can later be attributed per trigger rather than only in aggregate — see
|
||||
* `commandHqWatcherFires`'s own docstring. Does not touch `watcherArmed` or `watcherTaskId` on the
|
||||
* commander row, which stay exactly as they were through the fired-then-acting window, right up until the
|
||||
* next `armCommandHqWatcher` replaces the task id on re-arm.
|
||||
*
|
||||
* Takes the commander's own row id, not `(userId, sessionId)` — every caller already has the row in hand
|
||||
* (it had to check `watcherTaskId` matched before deciding this WAS the tracked watcher firing), so this
|
||||
* avoids resolving it a second time.
|
||||
*/
|
||||
export async function recordCommandHqWatcherFired(userId: number, sessionId: string): Promise<boolean> {
|
||||
const rows = await db
|
||||
.update(commandHqSessions)
|
||||
.set({ watcherLastFiredAt: new Date() })
|
||||
.where(and(eq(commandHqSessions.userId, userId), eq(commandHqSessions.sessionId, sessionId)))
|
||||
.returning({ id: commandHqSessions.id });
|
||||
return rows.length > 0;
|
||||
export async function recordCommandHqWatcherFired(sessionRowId: number, taskId: string): Promise<CommandHqWatcherFire> {
|
||||
const [row] = await db
|
||||
.insert(commandHqWatcherFires)
|
||||
.values({ sessionRowId, taskId })
|
||||
.returning({ id: commandHqWatcherFires.id, firedAt: commandHqWatcherFires.firedAt });
|
||||
return row!;
|
||||
}
|
||||
|
||||
/**
|
||||
* Disarm a commander's watcher. No product path calls this yet — there is no UI for it — but the
|
||||
* mechanism needs an off switch to exist before it needs a button, and `watcherTaskId` must be cleared
|
||||
* alongside `watcherArmed`, not left pointing at a task nothing will ever re-arm from again.
|
||||
* Disarm a commander's watcher. No product path calls this yet on the web side — there is no UI for it —
|
||||
* but the mechanism needs an off switch to exist before it needs a button, and `watcherTaskId` must be
|
||||
* cleared alongside `watcherArmed`, not left pointing at a task nothing will ever re-arm from again.
|
||||
* `watcherStoppedAt` is set here and ONLY here — the other end of the pair `watcherArmedAt` began.
|
||||
*/
|
||||
export async function disarmCommandHqWatcher(userId: number, sessionId: string): Promise<boolean> {
|
||||
const rows = await db
|
||||
.update(commandHqSessions)
|
||||
.set({ watcherArmed: false, watcherTaskId: null })
|
||||
.set({ watcherArmed: false, watcherTaskId: null, watcherStoppedAt: new Date() })
|
||||
.where(and(eq(commandHqSessions.userId, userId), eq(commandHqSessions.sessionId, sessionId)))
|
||||
.returning({ id: commandHqSessions.id });
|
||||
return rows.length > 0;
|
||||
}
|
||||
|
||||
export type CommandHqTurnCost = { inputTokens: number; outputTokens: number; totalUsd: number };
|
||||
|
||||
/**
|
||||
* Record one completed turn's cost — called on every `result` a commander's session produces, not only
|
||||
* the ones that follow a fire. The raw ledger `docs/command-hq.md` describes: running total, cost over
|
||||
* time, cost per trigger and turn count are all queries over this table, nothing pre-summed here.
|
||||
*
|
||||
* `fireId` is resolved to whichever trigger is CURRENTLY open for this commander — the most recent fire
|
||||
* row, if any — inside the same statement, so a turn during the setup conversation (before any fire has
|
||||
* ever happened) correctly gets `fireId: null` rather than a query needing to special-case it.
|
||||
*/
|
||||
export async function recordCommandHqTurn(sessionRowId: number, cost: CommandHqTurnCost): Promise<void> {
|
||||
// Two statements rather than one INSERT ... SELECT: a fire landing in the gap between them would
|
||||
// misattribute at most one turn to the setup conversation instead of the trigger that just started,
|
||||
// which is a smaller and more honest error than the complexity of a correlated subquery buys back —
|
||||
// fires happen far less often than turns complete, so the gap is rarely even open.
|
||||
const [currentFire] = await db
|
||||
.select({ id: commandHqWatcherFires.id })
|
||||
.from(commandHqWatcherFires)
|
||||
.where(eq(commandHqWatcherFires.sessionRowId, sessionRowId))
|
||||
.orderBy(desc(commandHqWatcherFires.firedAt))
|
||||
.limit(1);
|
||||
await db.insert(commandHqTurns).values({
|
||||
sessionRowId,
|
||||
fireId: currentFire?.id ?? null,
|
||||
inputTokens: cost.inputTokens,
|
||||
outputTokens: cost.outputTokens,
|
||||
totalUsd: cost.totalUsd,
|
||||
});
|
||||
}
|
||||
|
||||
export type CommandHqCostSummary = {
|
||||
turnCount: number;
|
||||
inputTokens: number;
|
||||
outputTokens: number;
|
||||
totalUsd: number;
|
||||
};
|
||||
|
||||
/** The running total for a commander — every turn it has ever cost, summed. Zeroes, never null, for a commander with no turns recorded yet. */
|
||||
export async function getCommandHqCostSummary(sessionRowId: number): Promise<CommandHqCostSummary> {
|
||||
const [row] = await db
|
||||
.select({
|
||||
turnCount: sql<number>`count(*)::int`,
|
||||
inputTokens: sql<number>`coalesce(sum(${commandHqTurns.inputTokens}), 0)::int`,
|
||||
outputTokens: sql<number>`coalesce(sum(${commandHqTurns.outputTokens}), 0)::int`,
|
||||
totalUsd: sql<number>`coalesce(sum(${commandHqTurns.totalUsd}), 0)::float8`,
|
||||
})
|
||||
.from(commandHqTurns)
|
||||
.where(eq(commandHqTurns.sessionRowId, sessionRowId));
|
||||
return row ?? { turnCount: 0, inputTokens: 0, outputTokens: 0, totalUsd: 0 };
|
||||
}
|
||||
|
||||
export type CommandHqTurnRow = {
|
||||
id: number;
|
||||
fireId: number | null;
|
||||
inputTokens: number;
|
||||
outputTokens: number;
|
||||
totalUsd: number;
|
||||
createdAt: Date;
|
||||
};
|
||||
|
||||
/** Every turn's cost for a commander, oldest first — the raw series a client plots cost-over-time from. */
|
||||
export async function listCommandHqTurns(sessionRowId: number): Promise<CommandHqTurnRow[]> {
|
||||
return db
|
||||
.select({
|
||||
id: commandHqTurns.id,
|
||||
fireId: commandHqTurns.fireId,
|
||||
inputTokens: commandHqTurns.inputTokens,
|
||||
outputTokens: commandHqTurns.outputTokens,
|
||||
totalUsd: commandHqTurns.totalUsd,
|
||||
createdAt: commandHqTurns.createdAt,
|
||||
})
|
||||
.from(commandHqTurns)
|
||||
.where(eq(commandHqTurns.sessionRowId, sessionRowId))
|
||||
.orderBy(commandHqTurns.createdAt);
|
||||
}
|
||||
|
||||
export type CommandHqFireRow = { id: number; taskId: string; firedAt: Date };
|
||||
|
||||
/** Every trigger for a commander, oldest first. */
|
||||
export async function listCommandHqWatcherFires(sessionRowId: number): Promise<CommandHqFireRow[]> {
|
||||
return db
|
||||
.select({
|
||||
id: commandHqWatcherFires.id,
|
||||
taskId: commandHqWatcherFires.taskId,
|
||||
firedAt: commandHqWatcherFires.firedAt,
|
||||
})
|
||||
.from(commandHqWatcherFires)
|
||||
.where(eq(commandHqWatcherFires.sessionRowId, sessionRowId))
|
||||
.orderBy(commandHqWatcherFires.firedAt);
|
||||
}
|
||||
|
||||
/**
|
||||
* The most recent fire for each of the given commanders, batched — one query for a whole grid rather than
|
||||
* one per card, the same shape `listCommandHqSessionTags` already uses. "Last fired" has no column of its
|
||||
* own (dropped deliberately — see the schema file); this is where the grid gets it instead.
|
||||
*/
|
||||
export async function getLastFiredAtBySessionRowIds(sessionRowIds: number[]): Promise<Map<number, Date>> {
|
||||
if (!sessionRowIds.length) return new Map();
|
||||
const rows = await db
|
||||
.select({
|
||||
sessionRowId: commandHqWatcherFires.sessionRowId,
|
||||
lastFiredAt: sql<Date>`max(${commandHqWatcherFires.firedAt})`,
|
||||
})
|
||||
.from(commandHqWatcherFires)
|
||||
.where(inArray(commandHqWatcherFires.sessionRowId, sessionRowIds))
|
||||
.groupBy(commandHqWatcherFires.sessionRowId);
|
||||
return new Map(rows.map((r) => [r.sessionRowId, r.lastFiredAt]));
|
||||
}
|
||||
|
||||
/** Move a commander to a category, or `null` to uncategorise it explicitly (distinct from never having had one). */
|
||||
export async function setCommandHqSessionCategory(
|
||||
userId: number,
|
||||
|
||||
@@ -1,4 +1,14 @@
|
||||
import { boolean, integer, pgTable, serial, text, timestamp, uniqueIndex } from 'drizzle-orm/pg-core';
|
||||
import {
|
||||
boolean,
|
||||
doublePrecision,
|
||||
index,
|
||||
integer,
|
||||
pgTable,
|
||||
serial,
|
||||
text,
|
||||
timestamp,
|
||||
uniqueIndex,
|
||||
} from 'drizzle-orm/pg-core';
|
||||
import { users } from '../auth/schema';
|
||||
|
||||
/**
|
||||
@@ -88,13 +98,76 @@ export const commandHqSessions = pgTable(
|
||||
* exactly the window a card would show as "armed, acting on what just fired" rather than "watching".
|
||||
*/
|
||||
watcherTaskId: text('watcher_task_id'),
|
||||
/** When the watcher last fired. Display only — "Armed, last fired 3h ago". Never read for logic. */
|
||||
watcherLastFiredAt: timestamp('watcher_last_fired_at', { withTimezone: true }),
|
||||
/**
|
||||
* When the CURRENT (or most recent) run started — set the moment `watcherArmed` first flips false→true,
|
||||
* left untouched by every later self-re-arm, so a watcher that has been cycling for weeks still reports
|
||||
* its true start rather than its last re-arm. Answers "how long has this been running" while armed, and
|
||||
* "how long did that run last" (paired with `watcherStoppedAt`) once it isn't. Not a full arm/disarm
|
||||
* history — only the boundaries of the current or most recent run, which is what a card needs; a table
|
||||
* of every transition was rejected as more than this needs, the same call `watcherLastFiredAt` lost for
|
||||
* the opposite reason (not enough — see `commandHqWatcherFires` above).
|
||||
*/
|
||||
watcherArmedAt: timestamp('watcher_armed_at', { withTimezone: true }),
|
||||
/**
|
||||
* When the current run ENDED — set on disarm, cleared back to null on the next arm. Paired with
|
||||
* `watcherArmedAt` so a duration is computable even after the watcher stops: while armed this is null
|
||||
* (`watcherArmedAt` to now); once disarmed both are set and the gap between them is how long that run
|
||||
* lasted. Both null means never armed at all.
|
||||
*/
|
||||
watcherStoppedAt: timestamp('watcher_stopped_at', { withTimezone: true }),
|
||||
createdAt: timestamp('created_at', { withTimezone: true }).notNull().defaultNow(),
|
||||
},
|
||||
(table) => [uniqueIndex('uq_command_hq_sessions_user_session').on(table.userId, table.sessionId)],
|
||||
);
|
||||
|
||||
/**
|
||||
* One row per time a commander's watcher fired — a "trigger". Nothing but the raw fact and when it
|
||||
* happened: no cached "last fired" lives on `commandHqSessions` — that is `MAX(firedAt)` for the commander,
|
||||
* computed at read time, the same "derive it, do not store a copy" rule `running` above already follows.
|
||||
*
|
||||
* This is what makes "cost per trigger" answerable at all. Before this table existed, a fire only ever
|
||||
* updated a single timestamp column, which meant the PREVIOUS fire's time was gone the moment a new one
|
||||
* landed — fine for "when did it last fire", useless for "how many times, and what did each one cost".
|
||||
*/
|
||||
export const commandHqWatcherFires = pgTable(
|
||||
'command_hq_watcher_fires',
|
||||
{
|
||||
id: serial('id').primaryKey(),
|
||||
sessionRowId: integer('session_row_id')
|
||||
.notNull()
|
||||
.references(() => commandHqSessions.id, { onDelete: 'cascade' }),
|
||||
/** Which watcher task fired — `commandHqSessions.watcherTaskId` at the moment this row was written. */
|
||||
taskId: text('task_id').notNull(),
|
||||
firedAt: timestamp('fired_at', { withTimezone: true }).notNull().defaultNow(),
|
||||
},
|
||||
(table) => [index('idx_command_hq_watcher_fires_session').on(table.sessionRowId, table.firedAt)],
|
||||
);
|
||||
|
||||
/**
|
||||
* One row per completed turn — every `result` message a commander's session produces, cost attached.
|
||||
* The raw ledger: running total, cost over time, cost per trigger and turn count are all queries over
|
||||
* this table, nothing pre-summed or cached elsewhere. `totalUsd` is a plain float (`doublePrecision`),
|
||||
* matching `MessageCost.totalUSD` as the harness itself reports it — this is a monitoring number, not a
|
||||
* billing ledger, and the accumulated float error over a realistic turn count is far below what the
|
||||
* harness's own cost estimate is accurate to in the first place.
|
||||
*/
|
||||
export const commandHqTurns = pgTable(
|
||||
'command_hq_turns',
|
||||
{
|
||||
id: serial('id').primaryKey(),
|
||||
sessionRowId: integer('session_row_id')
|
||||
.notNull()
|
||||
.references(() => commandHqSessions.id, { onDelete: 'cascade' }),
|
||||
/** Which trigger this turn belongs to. Null for the setup conversation before the first-ever arm. */
|
||||
fireId: integer('fire_id').references(() => commandHqWatcherFires.id, { onDelete: 'set null' }),
|
||||
inputTokens: integer('input_tokens').notNull(),
|
||||
outputTokens: integer('output_tokens').notNull(),
|
||||
totalUsd: doublePrecision('total_usd').notNull(),
|
||||
createdAt: timestamp('created_at', { withTimezone: true }).notNull().defaultNow(),
|
||||
},
|
||||
(table) => [index('idx_command_hq_turns_session').on(table.sessionRowId, table.createdAt)],
|
||||
);
|
||||
|
||||
/** A user-created label. Many-to-many with commanders, unlike categories — schema only for v1; no UI yet. */
|
||||
export const commandHqTags = pgTable(
|
||||
'command_hq_tags',
|
||||
|
||||
Reference in New Issue
Block a user