diff --git a/src/main/database/runs.ts b/src/main/database/runs.ts index 1a76773d..6a1a6986 100644 --- a/src/main/database/runs.ts +++ b/src/main/database/runs.ts @@ -303,17 +303,19 @@ export function buildRunMethods( message?: string | null; payload?: Record | 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<{ diff --git a/src/main/database/schema.ts b/src/main/database/schema.ts index 3b0923a2..316a14db 100644 --- a/src/main/database/schema.ts +++ b/src/main/database/schema.ts @@ -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) {