mirror of
https://github.com/stablyai/orca.git
synced 2026-10-02 08:02:02 +00:00
887 lines
32 KiB
TypeScript
887 lines
32 KiB
TypeScript
import { mkdtempSync, rmSync } from 'node:fs'
|
|
import { tmpdir } from 'node:os'
|
|
import { join } from 'node:path'
|
|
import { afterEach, describe, expect, it } from 'vitest'
|
|
import Database from '../../sqlite/sync-database'
|
|
import { LEGACY_RUN_ID, OrchestrationDb } from './db'
|
|
import { createRootDispatch } from './db/root-dispatch-test-fixture'
|
|
|
|
const runId = 'run_legacy_local'
|
|
|
|
// Overwrites the datetime('now')-seeded timestamps with explicit fixture values
|
|
// so stale-detection assertions stay deterministic (no wall clock).
|
|
function setDispatchTimes(
|
|
d: OrchestrationDb,
|
|
id: string,
|
|
dispatchedAt: string,
|
|
heartbeatAt: string | null = null
|
|
): void {
|
|
const sqlite = (d as unknown as { db: Database.Database }).db
|
|
sqlite
|
|
.prepare('UPDATE dispatch_contexts SET dispatched_at = ?, last_heartbeat_at = ? WHERE id = ?')
|
|
.run(dispatchedAt, heartbeatAt, id)
|
|
}
|
|
|
|
describe('OrchestrationDb', () => {
|
|
let db: OrchestrationDb | undefined
|
|
|
|
afterEach(() => {
|
|
db?.close()
|
|
})
|
|
|
|
function createDb(): OrchestrationDb {
|
|
db = new OrchestrationDb(':memory:')
|
|
return db
|
|
}
|
|
|
|
describe('tasks', () => {
|
|
it('creates a task with no deps as ready', () => {
|
|
const d = createDb()
|
|
const task = d.createTask({ runId, spec: 'do something' })
|
|
expect(task.id).toMatch(/^task_/)
|
|
expect(task.status).toBe('ready')
|
|
expect(task.deps).toBe('[]')
|
|
expect(task.task_title).toBe('do something')
|
|
expect(task.display_name).toBe('do something')
|
|
})
|
|
|
|
it('persists explicit task display metadata', () => {
|
|
const d = createDb()
|
|
const task = d.createTask({
|
|
runId,
|
|
spec: 'full details',
|
|
taskTitle: 'Checkout race',
|
|
displayName: 'Fix checkout race'
|
|
})
|
|
|
|
expect(task.task_title).toBe('Checkout race')
|
|
expect(task.display_name).toBe('Fix checkout race')
|
|
expect(d.getTask(task.id)?.display_name).toBe('Fix checkout race')
|
|
})
|
|
|
|
it('persists the creating terminal handle for task-created worktrees', () => {
|
|
const d = createDb()
|
|
const task = d.createTask({
|
|
runId,
|
|
spec: 'spawn related workspace',
|
|
createdByTerminalHandle: 'term_creator'
|
|
})
|
|
|
|
expect(task.created_by_terminal_handle).toBe('term_creator')
|
|
expect(d.getTask(task.id)?.created_by_terminal_handle).toBe('term_creator')
|
|
})
|
|
|
|
it('creates a task with deps as pending', () => {
|
|
const d = createDb()
|
|
const parent = d.createTask({ runId, spec: 'parent' })
|
|
const child = d.createTask({ runId, spec: 'child', deps: [parent.id] })
|
|
expect(child.status).toBe('pending')
|
|
expect(JSON.parse(child.deps)).toEqual([parent.id])
|
|
})
|
|
|
|
it('promotes pending tasks when deps complete', () => {
|
|
const d = createDb()
|
|
const t1 = d.createTask({ runId, spec: 'first' })
|
|
const t2 = d.createTask({ runId, spec: 'second', deps: [t1.id] })
|
|
|
|
expect(d.getTask(t2.id)?.status).toBe('pending')
|
|
|
|
d.updateTaskStatus(t1.id, 'completed')
|
|
|
|
expect(d.getTask(t2.id)?.status).toBe('ready')
|
|
})
|
|
|
|
it('does not promote task until ALL deps complete', () => {
|
|
const d = createDb()
|
|
const t1 = d.createTask({ runId, spec: 'a' })
|
|
const t2 = d.createTask({ runId, spec: 'b' })
|
|
const t3 = d.createTask({ runId, spec: 'c', deps: [t1.id, t2.id] })
|
|
|
|
d.updateTaskStatus(t1.id, 'completed')
|
|
expect(d.getTask(t3.id)?.status).toBe('pending')
|
|
|
|
d.updateTaskStatus(t2.id, 'completed')
|
|
expect(d.getTask(t3.id)?.status).toBe('ready')
|
|
})
|
|
|
|
it('sets completed_at on completion', () => {
|
|
const d = createDb()
|
|
const task = d.createTask({ runId, spec: 'do it' })
|
|
const updated = d.updateTaskStatus(task.id, 'completed', '{"result": true}')
|
|
expect(updated?.completed_at).toBeTruthy()
|
|
expect(updated?.result).toBe('{"result": true}')
|
|
})
|
|
|
|
it('completing a task frees its active dispatch context', () => {
|
|
const d = createDb()
|
|
const task = d.createTask({ runId, spec: 'do it' })
|
|
createRootDispatch(d, task.id, 'term_a')
|
|
|
|
d.updateTaskStatus(task.id, 'completed')
|
|
|
|
expect(d.getActiveDispatchForTerminal('term_a')).toBeUndefined()
|
|
expect(d.getDispatchContext(task.id)?.status).toBe('completed')
|
|
})
|
|
|
|
it('listTasks filters by status', () => {
|
|
const d = createDb()
|
|
d.createTask({ runId, spec: 'ready task' })
|
|
const t2 = d.createTask({ runId, spec: 'another' })
|
|
d.updateTaskStatus(t2.id, 'completed')
|
|
|
|
expect(d.listTasks({ status: 'ready' })).toHaveLength(1)
|
|
expect(d.listTasks({ status: 'completed' })).toHaveLength(1)
|
|
expect(d.listTasks({ ready: true })).toHaveLength(1)
|
|
})
|
|
|
|
it('listTasks returns all when no filter', () => {
|
|
const d = createDb()
|
|
d.createTask({ runId, spec: 'one' })
|
|
d.createTask({ runId, spec: 'two' })
|
|
expect(d.listTasks()).toHaveLength(2)
|
|
})
|
|
|
|
it('listTasksWithDispatch joins active dispatch metadata', () => {
|
|
const d = createDb()
|
|
const ready = d.createTask({ runId, spec: 'ready task' })
|
|
const dispatched = d.createTask({ runId, spec: 'active task' })
|
|
const ctx = createRootDispatch(d, dispatched.id, 'term_worker')
|
|
|
|
const rows = d.listTasksWithDispatch()
|
|
const readyRow = rows.find((r) => r.id === ready.id)
|
|
const dispatchedRow = rows.find((r) => r.id === dispatched.id)
|
|
|
|
expect(readyRow?.assignee_handle).toBeNull()
|
|
expect(readyRow?.dispatch_id).toBeNull()
|
|
expect(dispatchedRow?.assignee_handle).toBe('term_worker')
|
|
expect(dispatchedRow?.dispatch_id).toBe(ctx.id)
|
|
})
|
|
|
|
it('listTasksWithDispatch does not surface completed dispatches', () => {
|
|
const d = createDb()
|
|
const task = d.createTask({ runId, spec: 'work' })
|
|
createRootDispatch(d, task.id, 'term_worker')
|
|
d.updateTaskStatus(task.id, 'completed')
|
|
|
|
const rows = d.listTasksWithDispatch()
|
|
const row = rows.find((r) => r.id === task.id)
|
|
// Task is completed — its dispatch is terminal and should not appear as
|
|
// an "active" assignee.
|
|
expect(row?.assignee_handle).toBeNull()
|
|
expect(row?.dispatch_id).toBeNull()
|
|
})
|
|
|
|
it('supports parent_id for task decomposition', () => {
|
|
const d = createDb()
|
|
const parent = d.createTask({ runId, spec: 'parent' })
|
|
const child = d.createTask({ runId, spec: 'child', parentId: parent.id })
|
|
expect(child.parent_id).toBe(parent.id)
|
|
})
|
|
})
|
|
|
|
describe('dispatch contexts', () => {
|
|
it('creates a dispatch context and marks task as dispatched', () => {
|
|
const d = createDb()
|
|
const task = d.createTask({ runId, spec: 'work' })
|
|
const ctx = createRootDispatch(d, task.id, 'term_worker')
|
|
|
|
expect(ctx.id).toMatch(/^ctx_/)
|
|
expect(ctx.task_id).toBe(task.id)
|
|
expect(ctx.assignee_handle).toBe('term_worker')
|
|
expect(ctx.status).toBe('dispatched')
|
|
expect(d.getTask(task.id)?.status).toBe('dispatched')
|
|
})
|
|
|
|
it('rejects dispatch for non-ready tasks', () => {
|
|
const d = createDb()
|
|
const parent = d.createTask({ runId, spec: 'parent' })
|
|
const child = d.createTask({ runId, spec: 'child', deps: [parent.id] })
|
|
|
|
expect(() => createRootDispatch(d, child.id, 'term_worker')).toThrow(
|
|
/only ready tasks can be dispatched/
|
|
)
|
|
})
|
|
|
|
it('rejects dispatch to an occupied terminal', () => {
|
|
const d = createDb()
|
|
const t1 = d.createTask({ runId, spec: 'first' })
|
|
const t2 = d.createTask({ runId, spec: 'second' })
|
|
createRootDispatch(d, t1.id, 'term_worker')
|
|
|
|
expect(() => createRootDispatch(d, t2.id, 'term_worker')).toThrow(
|
|
/already has an active dispatch/
|
|
)
|
|
})
|
|
|
|
// Real leaf UUIDs: pane keys are `${tabId}:${leafUuid}`; only the leaf is
|
|
// remint-stable identity (tab half can change on pane break-out).
|
|
const LEAF_A = '11111111-1111-1111-8111-111111111111'
|
|
const LEAF_B = '22222222-2222-4222-9222-222222222222'
|
|
|
|
it('rejects dispatch to a reminted handle on a pane with an active dispatch', () => {
|
|
const d = createDb()
|
|
const t1 = d.createTask({ runId, spec: 'first' })
|
|
const t2 = d.createTask({ runId, spec: 'second' })
|
|
createRootDispatch(d, t1.id, 'term_old', `tab_1:${LEAF_A}`)
|
|
|
|
expect(() => createRootDispatch(d, t2.id, 'term_new', `tab_1:${LEAF_A}`)).toThrow(
|
|
/already has an active dispatch/
|
|
)
|
|
})
|
|
|
|
it('rejects dispatch when pane keys share a leaf after break-out', () => {
|
|
const d = createDb()
|
|
const t1 = d.createTask({ runId, spec: 'first' })
|
|
const t2 = d.createTask({ runId, spec: 'second' })
|
|
createRootDispatch(d, t1.id, 'term_old', `tab_1:${LEAF_A}`)
|
|
|
|
expect(() => createRootDispatch(d, t2.id, 'term_new', `tab_2:${LEAF_A}`)).toThrow(
|
|
/already has an active dispatch/
|
|
)
|
|
})
|
|
|
|
it('allows concurrent dispatches to different panes', () => {
|
|
const d = createDb()
|
|
const t1 = d.createTask({ runId, spec: 'first' })
|
|
const t2 = d.createTask({ runId, spec: 'second' })
|
|
createRootDispatch(d, t1.id, 'term_a', `tab_1:${LEAF_A}`)
|
|
|
|
expect(() => createRootDispatch(d, t2.id, 'term_b', `tab_1:${LEAF_B}`)).not.toThrow()
|
|
})
|
|
|
|
it('falls back to handle lock when pane keys are missing', () => {
|
|
const d = createDb()
|
|
const t1 = d.createTask({ runId, spec: 'first' })
|
|
const t2 = d.createTask({ runId, spec: 'second' })
|
|
createRootDispatch(d, t1.id, 'term_worker')
|
|
|
|
// New dispatch has a pane key but the active row is legacy (no pane key):
|
|
// only handle identity can lock; a different handle is free.
|
|
expect(() => createRootDispatch(d, t2.id, 'term_other', `tab_1:${LEAF_A}`)).not.toThrow()
|
|
})
|
|
|
|
it('allows dispatch to a terminal after previous dispatch completes', () => {
|
|
const d = createDb()
|
|
const t1 = d.createTask({ runId, spec: 'first' })
|
|
const t2 = d.createTask({ runId, spec: 'second' })
|
|
const ctx1 = createRootDispatch(d, t1.id, 'term_worker')
|
|
|
|
d.completeDispatch(ctx1.id)
|
|
|
|
expect(() => createRootDispatch(d, t2.id, 'term_worker')).not.toThrow()
|
|
})
|
|
|
|
it('getDispatchContext returns latest for a task', () => {
|
|
const d = createDb()
|
|
const task = d.createTask({ runId, spec: 'work' })
|
|
const ctx = createRootDispatch(d, task.id, 'term_a')
|
|
const found = d.getDispatchContext(task.id)
|
|
expect(found?.id).toBe(ctx.id)
|
|
})
|
|
|
|
it('getDispatchContext uses insertion order when timestamps tie', () => {
|
|
const d = createDb()
|
|
const task = d.createTask({ runId, spec: 'work' })
|
|
const ctx1 = createRootDispatch(d, task.id, 'term_a')
|
|
d.failDispatch(ctx1.id, 'retry')
|
|
const ctx2 = createRootDispatch(d, task.id, 'term_a')
|
|
|
|
expect(d.getDispatchContext(task.id)?.id).toBe(ctx2.id)
|
|
})
|
|
|
|
it('getActiveDispatchForTerminal returns active dispatch', () => {
|
|
const d = createDb()
|
|
const task = d.createTask({ runId, spec: 'work' })
|
|
createRootDispatch(d, task.id, 'term_a')
|
|
|
|
const active = d.getActiveDispatchForTerminal('term_a')
|
|
expect(active?.task_id).toBe(task.id)
|
|
expect(d.getActiveDispatchForTerminal('term_b')).toBeUndefined()
|
|
})
|
|
|
|
it('getLatestDispatchForTerminal returns the most recent completed dispatch', () => {
|
|
const d = createDb()
|
|
const firstTask = d.createTask({
|
|
runId,
|
|
spec: 'first'
|
|
})
|
|
const first = createRootDispatch(d, firstTask.id, 'term_a')
|
|
d.completeDispatch(first.id)
|
|
const secondTask = d.createTask({
|
|
runId,
|
|
spec: 'second'
|
|
})
|
|
const second = createRootDispatch(d, secondTask.id, 'term_a')
|
|
d.completeDispatch(second.id)
|
|
|
|
const latest = d.getLatestDispatchForTerminal('term_a')
|
|
expect(latest?.id).toBe(second.id)
|
|
expect(latest?.status).toBe('completed')
|
|
expect(d.getActiveDispatchForTerminal('term_a')).toBeUndefined()
|
|
})
|
|
|
|
it('circuit breaker trips after 3 failures', () => {
|
|
const d = createDb()
|
|
const task = d.createTask({ runId, spec: 'flaky' })
|
|
const ctx = createRootDispatch(d, task.id, 'term_a')
|
|
|
|
const after1 = d.failDispatch(ctx.id, 'timeout')
|
|
expect(after1?.failure_count).toBe(1)
|
|
expect(after1?.status).toBe('failed')
|
|
expect(d.getTask(task.id)?.status).toBe('ready')
|
|
|
|
const ctx2 = createRootDispatch(d, task.id, 'term_a')
|
|
const after2 = d.failDispatch(ctx2.id, 'timeout')
|
|
expect(after2?.failure_count).toBe(2)
|
|
expect(after2?.status).toBe('failed')
|
|
|
|
const ctx3 = createRootDispatch(d, task.id, 'term_a')
|
|
const after3 = d.failDispatch(ctx3.id, 'timeout')
|
|
expect(after3?.failure_count).toBe(3)
|
|
expect(after3?.status).toBe('circuit_broken')
|
|
expect([after1, after2, after3].every((dispatch) => dispatch?.completed_at)).toBe(true)
|
|
expect(d.getTask(task.id)?.status).toBe('failed')
|
|
})
|
|
|
|
it('completeDispatch sets completed_at', () => {
|
|
const d = createDb()
|
|
const task = d.createTask({ runId, spec: 'work' })
|
|
const ctx = createRootDispatch(d, task.id, 'term_a')
|
|
d.completeDispatch(ctx.id)
|
|
|
|
const updated = d.getDispatchContext(task.id)
|
|
expect(updated?.status).toBe('completed')
|
|
expect(updated?.completed_at).toBeTruthy()
|
|
})
|
|
})
|
|
|
|
describe('decision gates', () => {
|
|
it('creates a gate and blocks the task', () => {
|
|
const d = createDb()
|
|
const task = d.createTask({
|
|
runId,
|
|
spec: 'needs approval'
|
|
})
|
|
createRootDispatch(d, task.id, 'term_a')
|
|
const gate = d.createGate({
|
|
taskId: task.id,
|
|
question: 'Proceed?',
|
|
options: ['yes', 'no']
|
|
})
|
|
|
|
expect(gate.id).toMatch(/^gate_/)
|
|
expect(gate.task_id).toBe(task.id)
|
|
expect(gate.status).toBe('pending')
|
|
expect(JSON.parse(gate.options)).toEqual(['yes', 'no'])
|
|
|
|
const updated = d.getTask(task.id)
|
|
expect(updated?.status).toBe('blocked')
|
|
expect(d.getActiveDispatchForTerminal('term_a')).toBeUndefined()
|
|
})
|
|
|
|
it('resolves a gate and unblocks the task', () => {
|
|
const d = createDb()
|
|
const task = d.createTask({ runId, spec: 'work' })
|
|
const gate = d.createGate({ taskId: task.id, question: 'ok?' })
|
|
|
|
const resolved = d.resolveGate(gate.id, 'yes')
|
|
expect(resolved?.status).toBe('resolved')
|
|
expect(resolved?.resolution).toBe('yes')
|
|
|
|
const updated = d.getTask(task.id)
|
|
expect(updated?.status).toBe('ready')
|
|
})
|
|
|
|
it('times out a gate', () => {
|
|
const d = createDb()
|
|
const task = d.createTask({ runId, spec: 'work' })
|
|
const gate = d.createGate({ taskId: task.id, question: 'ok?' })
|
|
|
|
const timedOut = d.timeoutGate(gate.id)
|
|
expect(timedOut?.status).toBe('timeout')
|
|
})
|
|
|
|
it('lists gates with filters', () => {
|
|
const d = createDb()
|
|
const t1 = d.createTask({ runId, spec: 'a' })
|
|
const t2 = d.createTask({ runId, spec: 'b' })
|
|
d.createGate({ taskId: t1.id, question: 'q1' })
|
|
const g2 = d.createGate({ taskId: t2.id, question: 'q2' })
|
|
d.resolveGate(g2.id, 'done')
|
|
|
|
expect(d.listGates()).toHaveLength(2)
|
|
expect(d.listGates({ status: 'pending' })).toHaveLength(1)
|
|
expect(d.listGates({ taskId: t1.id })).toHaveLength(1)
|
|
expect(d.listGates({ taskId: t2.id, status: 'resolved' })).toHaveLength(1)
|
|
})
|
|
|
|
it('returns undefined for nonexistent gate', () => {
|
|
const d = createDb()
|
|
expect(d.resolveGate('gate_fake', 'yes')).toBeUndefined()
|
|
})
|
|
})
|
|
|
|
describe('coordinator runs', () => {
|
|
it('creates and retrieves a coordinator run', () => {
|
|
const d = createDb()
|
|
const run = d.createCoordinatorRun({
|
|
spec: 'build feature',
|
|
coordinatorHandle: 'coord',
|
|
pollIntervalMs: 1000
|
|
})
|
|
|
|
expect(run.id).toMatch(/^run_/)
|
|
expect(run.status).toBe('running')
|
|
expect(run.coordinator_handle).toBe('coord')
|
|
expect(run.poll_interval_ms).toBe(1000)
|
|
})
|
|
|
|
it('updates coordinator run status', () => {
|
|
const d = createDb()
|
|
const run = d.createCoordinatorRun({
|
|
spec: 'work',
|
|
coordinatorHandle: 'coord'
|
|
})
|
|
|
|
const updated = d.updateCoordinatorRun(run.id, 'completed')
|
|
expect(updated?.status).toBe('completed')
|
|
expect(updated?.completed_at).not.toBeNull()
|
|
})
|
|
|
|
it('finds active coordinator run', () => {
|
|
const d = createDb()
|
|
expect(d.getActiveCoordinatorRun()).toBeUndefined()
|
|
|
|
const run = d.createCoordinatorRun({
|
|
spec: 'work',
|
|
coordinatorHandle: 'coord'
|
|
})
|
|
|
|
expect(d.getActiveCoordinatorRun()?.id).toBe(run.id)
|
|
|
|
d.updateCoordinatorRun(run.id, 'completed')
|
|
expect(d.getActiveCoordinatorRun()).toBeUndefined()
|
|
})
|
|
})
|
|
|
|
describe('lifecycle', () => {
|
|
it('resetAll clears all tables', () => {
|
|
const d = createDb()
|
|
d.insertMessage({
|
|
runId,
|
|
from: 'a',
|
|
to: 'b',
|
|
subject: 'test'
|
|
})
|
|
d.createTask({ runId, spec: 'work' })
|
|
|
|
d.resetAll()
|
|
|
|
expect(d.getInbox()).toHaveLength(0)
|
|
expect(d.listTasks()).toHaveLength(0)
|
|
})
|
|
|
|
it('resetMessages clears only messages', () => {
|
|
const d = createDb()
|
|
d.insertMessage({
|
|
runId,
|
|
from: 'a',
|
|
to: 'b',
|
|
subject: 'test'
|
|
})
|
|
d.createTask({ runId, spec: 'work' })
|
|
|
|
d.resetMessages()
|
|
|
|
expect(d.getInbox()).toHaveLength(0)
|
|
expect(d.listTasks()).toHaveLength(1)
|
|
})
|
|
|
|
it('resetTasks clears tasks and dispatch contexts', () => {
|
|
const d = createDb()
|
|
d.insertMessage({
|
|
runId,
|
|
from: 'a',
|
|
to: 'b',
|
|
subject: 'test'
|
|
})
|
|
const task = d.createTask({ runId, spec: 'work' })
|
|
createRootDispatch(d, task.id, 'term_a')
|
|
|
|
d.resetTasks()
|
|
|
|
expect(d.getInbox()).toHaveLength(1)
|
|
expect(d.listTasks()).toHaveLength(0)
|
|
})
|
|
})
|
|
|
|
describe('heartbeat + thread helpers (fresh schema)', () => {
|
|
it('insertMessage accepts type = heartbeat', () => {
|
|
const d = createDb()
|
|
const msg = d.insertMessage({
|
|
runId,
|
|
from: 'worker',
|
|
to: 'coord',
|
|
subject: 'alive',
|
|
type: 'heartbeat',
|
|
payload: JSON.stringify({ taskId: 'task_x', dispatchId: 'ctx_x' })
|
|
})
|
|
expect(msg.type).toBe('heartbeat')
|
|
})
|
|
|
|
it('recordHeartbeat updates last_heartbeat_at on dispatched rows', () => {
|
|
const d = createDb()
|
|
const task = d.createTask({ runId, spec: 'work' })
|
|
const ctx = createRootDispatch(d, task.id, 'term_a')
|
|
|
|
d.recordHeartbeat(ctx.id, '2026-05-04T00:00:00.000Z')
|
|
const after = d.getDispatchContext(task.id)
|
|
expect(after?.last_heartbeat_at).toBe('2026-05-04T00:00:00.000Z')
|
|
})
|
|
|
|
it('getStaleDispatches returns only dispatched rows past the grace window', () => {
|
|
const d = createDb()
|
|
// Fixture: four rows, SQL-backdated timestamps (no fake clock):
|
|
// (a) dispatched, heartbeated 5 min ago → not stale
|
|
// (b) dispatched, heartbeated 12 min ago → STALE (expected result)
|
|
// (c) dispatched, never heartbeated, dispatched 30s ago → not stale (grace)
|
|
// (d) completed, heartbeated 30 min ago → not stale (status filter)
|
|
const taskA = d.createTask({ runId, spec: 'a' })
|
|
const taskB = d.createTask({ runId, spec: 'b' })
|
|
const taskC = d.createTask({ runId, spec: 'c' })
|
|
const taskD = d.createTask({ runId, spec: 'd' })
|
|
const ctxA = createRootDispatch(d, taskA.id, 'term_a')
|
|
const ctxB = createRootDispatch(d, taskB.id, 'term_b')
|
|
const ctxC = createRootDispatch(d, taskC.id, 'term_c')
|
|
const ctxD = createRootDispatch(d, taskD.id, 'term_d')
|
|
d.completeDispatch(ctxD.id)
|
|
|
|
const now = Date.now()
|
|
const iso = (ms: number) => new Date(now - ms).toISOString()
|
|
|
|
// Backdate dispatched_at for a, b, d to long ago so the grace doesn't
|
|
// shield them. c keeps its default (≈now).
|
|
const sqlite = (d as unknown as { db: Database.Database }).db
|
|
sqlite
|
|
.prepare(
|
|
'UPDATE dispatch_contexts SET dispatched_at = ?, last_heartbeat_at = ? WHERE id = ?'
|
|
)
|
|
.run(iso(60 * 60 * 1000), iso(5 * 60 * 1000), ctxA.id)
|
|
sqlite
|
|
.prepare(
|
|
'UPDATE dispatch_contexts SET dispatched_at = ?, last_heartbeat_at = ? WHERE id = ?'
|
|
)
|
|
.run(iso(60 * 60 * 1000), iso(12 * 60 * 1000), ctxB.id)
|
|
sqlite
|
|
.prepare('UPDATE dispatch_contexts SET dispatched_at = ? WHERE id = ?')
|
|
.run(iso(30_000), ctxC.id)
|
|
sqlite
|
|
.prepare(
|
|
'UPDATE dispatch_contexts SET dispatched_at = ?, last_heartbeat_at = ? WHERE id = ?'
|
|
)
|
|
.run(iso(60 * 60 * 1000), iso(30 * 60 * 1000), ctxD.id)
|
|
|
|
const stale = d.getStaleDispatches(iso(10 * 60 * 1000))
|
|
expect(stale.map((s) => s.id)).toEqual([ctxB.id])
|
|
})
|
|
|
|
// Regression for #8452: dispatched_at / last_heartbeat_at are written by
|
|
// datetime('now') (space-format, e.g. "2026-07-12 12:00:00") while the
|
|
// threshold is ISO ("...T11:55:00.000Z"). Raw TEXT ordering ranks the space
|
|
// (0x20) below the 'T' (0x54) at index 10, flagging fresh same-date rows.
|
|
it('getStaleDispatches ignores fresh SQLite space-format timestamps (#8452)', () => {
|
|
const d = createDb()
|
|
|
|
// Fresh worker: dispatched 12:00, heartbeat 12:05 (space-format), both
|
|
// after the 11:55 threshold → NOT stale.
|
|
const fresh = createRootDispatch(d, d.createTask({ runId, spec: 'fresh' }).id, 'term_fresh')
|
|
setDispatchTimes(d, fresh.id, '2026-07-12 12:00:00', '2026-07-12 12:05:00')
|
|
|
|
// Legacy ISO-format fresh row (mixed-format table) stays fresh too.
|
|
const legacy = createRootDispatch(
|
|
d,
|
|
d.createTask({ runId, spec: 'legacy' }).id,
|
|
'term_legacy'
|
|
)
|
|
setDispatchTimes(d, legacy.id, '2026-07-12T12:00:00.000Z', '2026-07-12T12:05:00.000Z')
|
|
|
|
// Genuinely hung: dispatched + heartbeated at 10:00, ~2h before threshold.
|
|
const hung = createRootDispatch(d, d.createTask({ runId, spec: 'hung' }).id, 'term_hung')
|
|
setDispatchTimes(d, hung.id, '2026-07-12 10:00:00', '2026-07-12 10:00:00')
|
|
|
|
const stale = d.getStaleDispatches('2026-07-12T11:55:00.000Z')
|
|
expect(stale.map((s) => s.id)).toEqual([hung.id])
|
|
})
|
|
|
|
it('getStaleDispatches keeps a just-dispatched space-format row in the grace window (#8452)', () => {
|
|
const d = createDb()
|
|
|
|
// Space-format dispatched_at one minute after the threshold, no heartbeat
|
|
// yet → still inside the grace window, must not be flagged.
|
|
const ctx = createRootDispatch(d, d.createTask({ runId, spec: 'x' }).id, 'term_x')
|
|
setDispatchTimes(d, ctx.id, '2026-07-12 12:00:00')
|
|
|
|
const stale = d.getStaleDispatches('2026-07-12T11:59:00.000Z')
|
|
expect(stale).toEqual([])
|
|
})
|
|
|
|
// Same-UTC-date midnight threshold: keeps the buggy space-vs-'T' compare in
|
|
// play so this guards the fix at a day boundary (#8452; idea from @KMGeon's #8453).
|
|
it('getStaleDispatches keeps a fresh row just after a UTC-midnight threshold (#8452)', () => {
|
|
const d = createDb()
|
|
|
|
const ctx = createRootDispatch(
|
|
d,
|
|
d.createTask({ runId, spec: 'midnight' }).id,
|
|
'term_midnight'
|
|
)
|
|
setDispatchTimes(d, ctx.id, '2026-05-04 00:04:00')
|
|
|
|
const stale = d.getStaleDispatches('2026-05-04T00:00:00.000Z')
|
|
expect(stale).toEqual([])
|
|
})
|
|
|
|
// Guards the last_heartbeat_at half of the fix on its own: a worker
|
|
// dispatched long before the threshold (stale under either format) that
|
|
// just sent a fresh space-format heartbeat must stay fresh (#8452).
|
|
it('getStaleDispatches keeps a live worker with a fresh space-format heartbeat (#8452)', () => {
|
|
const d = createDb()
|
|
|
|
const ctx = createRootDispatch(d, d.createTask({ runId, spec: 'live' }).id, 'term_live')
|
|
setDispatchTimes(d, ctx.id, '2026-07-12 10:00:00', '2026-07-12 11:59:00')
|
|
|
|
const stale = d.getStaleDispatches('2026-07-12T11:55:00.000Z')
|
|
expect(stale).toEqual([])
|
|
})
|
|
|
|
it('getThreadMessagesFor returns only same-thread replies to a handle', () => {
|
|
const d = createDb()
|
|
const outbound = d.insertMessage({
|
|
runId,
|
|
from: 'worker',
|
|
to: 'coord',
|
|
subject: 'Question',
|
|
type: 'decision_gate',
|
|
body: 'yes or no?'
|
|
})
|
|
// Reply in the same thread addressed to the worker
|
|
const reply = d.insertMessage({
|
|
runId,
|
|
from: 'coord',
|
|
to: 'worker',
|
|
subject: 'Re: Question',
|
|
body: 'yes',
|
|
threadId: outbound.id
|
|
})
|
|
// Distractor: different thread, same recipient
|
|
d.insertMessage({
|
|
runId,
|
|
from: 'coord',
|
|
to: 'worker',
|
|
subject: 'other',
|
|
body: 'unrelated',
|
|
threadId: 'thread_other'
|
|
})
|
|
// Distractor: same thread but not addressed to worker
|
|
d.insertMessage({
|
|
runId,
|
|
from: 'coord',
|
|
to: 'someone_else',
|
|
subject: 'cc',
|
|
body: 'not yours',
|
|
threadId: outbound.id
|
|
})
|
|
|
|
const replies = d.getThreadMessagesFor(outbound.id, 'worker', outbound.sequence)
|
|
expect(replies).toHaveLength(1)
|
|
expect(replies[0].id).toBe(reply.id)
|
|
})
|
|
})
|
|
|
|
describe('schema migration from v1 → v2', () => {
|
|
let dbPath: string
|
|
let tempDir: string
|
|
|
|
afterEach(() => {
|
|
// Why: Windows keeps the SQLite file locked until the DB handle closes,
|
|
// so migration temp directories must close before recursive cleanup.
|
|
db?.close()
|
|
db = undefined
|
|
if (tempDir) {
|
|
rmSync(tempDir, { recursive: true, force: true })
|
|
}
|
|
})
|
|
|
|
function createV1Snapshot(): string {
|
|
tempDir = mkdtempSync(join(tmpdir(), 'orca-db-migrate-'))
|
|
dbPath = join(tempDir, 'test.db')
|
|
const raw = new Database(dbPath)
|
|
// v1 schema: pre-heartbeat CHECK, no last_heartbeat_at column.
|
|
raw.exec(`
|
|
CREATE TABLE messages (
|
|
id TEXT NOT NULL,
|
|
from_handle TEXT NOT NULL,
|
|
to_handle TEXT NOT NULL,
|
|
subject TEXT NOT NULL,
|
|
body TEXT NOT NULL DEFAULT '',
|
|
type TEXT NOT NULL DEFAULT 'status'
|
|
CHECK(type IN (
|
|
'status', 'dispatch', 'worker_done', 'merge_ready',
|
|
'escalation', 'handoff', 'decision_gate'
|
|
)),
|
|
priority TEXT NOT NULL DEFAULT 'normal'
|
|
CHECK(priority IN ('normal', 'high', 'urgent')),
|
|
thread_id TEXT,
|
|
payload TEXT,
|
|
read INTEGER NOT NULL DEFAULT 0,
|
|
sequence INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
created_at TEXT NOT NULL DEFAULT (datetime('now'))
|
|
);
|
|
CREATE UNIQUE INDEX idx_messages_id ON messages(id);
|
|
CREATE INDEX idx_inbox ON messages(to_handle, read);
|
|
CREATE INDEX idx_thread ON messages(thread_id);
|
|
|
|
CREATE TABLE tasks (
|
|
id TEXT PRIMARY KEY, parent_id TEXT, spec TEXT NOT NULL,
|
|
status TEXT NOT NULL DEFAULT 'pending'
|
|
CHECK(status IN ('pending','ready','dispatched','completed','failed','blocked')),
|
|
deps TEXT NOT NULL DEFAULT '[]', result TEXT,
|
|
created_at TEXT NOT NULL DEFAULT (datetime('now')),
|
|
completed_at TEXT
|
|
);
|
|
|
|
CREATE TABLE dispatch_contexts (
|
|
id TEXT PRIMARY KEY, task_id TEXT NOT NULL, assignee_handle TEXT,
|
|
status TEXT NOT NULL DEFAULT 'pending'
|
|
CHECK(status IN ('pending','dispatched','completed','failed','circuit_broken')),
|
|
failure_count INTEGER NOT NULL DEFAULT 0, last_failure TEXT,
|
|
dispatched_at TEXT, completed_at TEXT,
|
|
created_at TEXT NOT NULL DEFAULT (datetime('now'))
|
|
);
|
|
|
|
CREATE TABLE decision_gates (
|
|
id TEXT PRIMARY KEY, task_id TEXT NOT NULL, question TEXT NOT NULL,
|
|
options TEXT NOT NULL DEFAULT '[]',
|
|
status TEXT NOT NULL DEFAULT 'pending'
|
|
CHECK(status IN ('pending','resolved','timeout')),
|
|
resolution TEXT,
|
|
created_at TEXT NOT NULL DEFAULT (datetime('now')),
|
|
resolved_at TEXT
|
|
);
|
|
|
|
CREATE TABLE coordinator_runs (
|
|
id TEXT PRIMARY KEY, spec TEXT NOT NULL,
|
|
status TEXT NOT NULL DEFAULT 'idle'
|
|
CHECK(status IN ('idle','running','completed','failed')),
|
|
coordinator_handle TEXT NOT NULL,
|
|
poll_interval_ms INTEGER NOT NULL DEFAULT 2000,
|
|
created_at TEXT NOT NULL DEFAULT (datetime('now')),
|
|
completed_at TEXT
|
|
);
|
|
`)
|
|
// Seed a pre-existing v1 message so migration must preserve data.
|
|
raw
|
|
.prepare(
|
|
`INSERT INTO messages (id, from_handle, to_handle, subject, type) VALUES ('msg_v1', 'a', 'b', 'pre-migration', 'status')`
|
|
)
|
|
.run()
|
|
raw.pragma('user_version = 0')
|
|
raw.close()
|
|
return dbPath
|
|
}
|
|
|
|
it('migrates a v1 snapshot to v2, accepts heartbeat, preserves indexes', () => {
|
|
const path = createV1Snapshot()
|
|
const d = new OrchestrationDb(path)
|
|
db = d
|
|
|
|
// (a) INSERT type='heartbeat' now succeeds
|
|
expect(() =>
|
|
d.insertMessage({
|
|
runId,
|
|
from: 'w',
|
|
to: 'c',
|
|
subject: 'alive',
|
|
type: 'heartbeat',
|
|
payload: '{"taskId":"t","dispatchId":"ctx"}'
|
|
})
|
|
).not.toThrow()
|
|
|
|
// (b) last_heartbeat_at column exists on dispatch_contexts
|
|
const task = d.createTask({ runId, spec: 'work' })
|
|
const ctx = createRootDispatch(d, task.id, 'term_a')
|
|
d.recordHeartbeat(ctx.id, '2026-05-04T00:00:00.000Z')
|
|
expect(d.getDispatchContext(task.id)?.last_heartbeat_at).toBe('2026-05-04T00:00:00.000Z')
|
|
expect(d.getTask(task.id)?.task_title).toBe('work')
|
|
expect(d.getTask(task.id)?.display_name).toBe('work')
|
|
|
|
// (c) Indexes still attached to messages post-rebuild.
|
|
const sqlite = (d as unknown as { db: Database.Database }).db
|
|
const indexes = sqlite
|
|
.prepare(
|
|
`SELECT name FROM sqlite_master WHERE type = 'index' AND tbl_name = 'messages' AND name NOT LIKE 'sqlite_%'`
|
|
)
|
|
.all() as { name: string }[]
|
|
const names = new Set(indexes.map((r) => r.name))
|
|
expect(names.has('idx_messages_id')).toBe(true)
|
|
expect(names.has('idx_inbox')).toBe(true)
|
|
expect(names.has('idx_messages_undelivered_inbox')).toBe(true)
|
|
expect(names.has('idx_thread')).toBe(true)
|
|
|
|
// v1 data preserved
|
|
expect(d.getMessageById('msg_v1')?.subject).toBe('pre-migration')
|
|
expect(d.getMessageById('msg_v1')?.run_id).toBe(d.getLegacyAdoption()?.adopted_run_id)
|
|
expect(d.getRun(LEGACY_RUN_ID)).toMatchObject({ legacy: 1 })
|
|
})
|
|
|
|
it('adds pane-identity columns (v6) and persists them', () => {
|
|
const path = createV1Snapshot()
|
|
const d = new OrchestrationDb(path)
|
|
db = d
|
|
|
|
const task = d.createTask({ runId, spec: 'work' })
|
|
const ctx = createRootDispatch(d, task.id, 'term_a', 'tab_1:leaf_1')
|
|
expect(d.getDispatchContextById(ctx.id)?.assignee_pane_key).toBe('tab_1:leaf_1')
|
|
|
|
const msg = d.insertMessage({
|
|
runId,
|
|
from: 'w',
|
|
to: 'c',
|
|
subject: 'done',
|
|
type: 'worker_done',
|
|
senderPaneKey: 'tab_1:leaf_1'
|
|
})
|
|
expect(d.getMessageById(msg.id)?.sender_pane_key).toBe('tab_1:leaf_1')
|
|
})
|
|
|
|
it('is idempotent: opening an already-migrated DB is a no-op', () => {
|
|
const path = createV1Snapshot()
|
|
const first = new OrchestrationDb(path)
|
|
first.insertMessage({
|
|
runId,
|
|
from: 'w',
|
|
to: 'c',
|
|
subject: 'alive',
|
|
type: 'heartbeat',
|
|
payload: '{}'
|
|
})
|
|
first.close()
|
|
|
|
const second = new OrchestrationDb(path)
|
|
db = second
|
|
expect(() =>
|
|
second.insertMessage({
|
|
runId,
|
|
from: 'w',
|
|
to: 'c',
|
|
subject: 'again',
|
|
type: 'heartbeat',
|
|
payload: '{}'
|
|
})
|
|
).not.toThrow()
|
|
const inbox = second.getInbox(10)
|
|
expect(inbox.length).toBeGreaterThanOrEqual(2)
|
|
})
|
|
})
|
|
})
|