fix(db): add run_events(run_id,seq) unique index + wrap appendRunEvent in tx (P1-15)

This commit is contained in:
ZacharyZhang-NY committed 2026-04-16 22:53:06 -04:00
1 parent 8dfa69cacf
commit 12279005fe
2 files changed
+25 -11

No files matched your search

+13 -11
View File
@@ -303,17 +303,19 @@ export function buildRunMethods(
message?: string | null; payload?: Record<string, unknown> | null;
}): void {
const ts = nowIso();
const lastSeq = getDb().prepare(
"SELECT MAX(seq) as maxSeq FROM run_events WHERE run_id = ?"
).get(input.runId) as { maxSeq: number | null } | undefined;
const seq = (lastSeq?.maxSeq ?? 0) + 1;
getDb().prepare(
"INSERT INTO run_events (company_id, run_id, agent_id, seq, event_type, stream, level, message, payload, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
).run(
input.companyId, input.runId, input.agentId, seq,
input.eventType, input.stream ?? null, input.level ?? null,
input.message ?? null, input.payload ? JSON.stringify(input.payload) : null, ts
);
transaction(() => {
const lastSeq = getDb().prepare(
"SELECT MAX(seq) as maxSeq FROM run_events WHERE run_id = ?"
).get(input.runId) as { maxSeq: number | null } | undefined;
const seq = (lastSeq?.maxSeq ?? 0) + 1;
getDb().prepare(
"INSERT INTO run_events (company_id, run_id, agent_id, seq, event_type, stream, level, message, payload, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
).run(
input.companyId, input.runId, input.agentId, seq,
input.eventType, input.stream ?? null, input.level ?? null,
input.message ?? null, input.payload ? JSON.stringify(input.payload) : null, ts
);
});
},
listRunEvents(runId: string, companyId: string, opts?: { offset?: number; limit?: number }): Array<{
+12
View File
@@ -421,6 +421,18 @@ export function runMigrations(
CREATE INDEX IF NOT EXISTS idx_activity_company_actor_action ON activity(company_id, actor, action);
CREATE INDEX IF NOT EXISTS idx_tasks_company_assignee_status ON tasks(company_id, assignee_agent_id, status)`,
},
{
version: 29, name: "add_run_events_seq_unique_index",
sql: `CREATE UNIQUE INDEX IF NOT EXISTS idx_run_events_run_seq_unique ON run_events(run_id, seq)`,
},
{
// Supports checkpoint-engine `shouldPauseHeartbeat` join + filter on
// (company_id, state, type) -> assignee_agent_id lookup. Without this
// SQLite scans the full approvals table on every heartbeat tick.
version: 30, name: "add_checkpoint_pause_indexes",
sql: `CREATE INDEX IF NOT EXISTS idx_approvals_company_state_type ON approvals(company_id, state, type);
CREATE INDEX IF NOT EXISTS idx_approvals_related_task ON approvals(related_task_id)`,
},
];
for (const migration of migrations) {