diff --git a/src/main/automation/automation-run-completion.ts b/src/main/automation/automation-run-completion.ts index 94794944..f1ba9934 100644 --- a/src/main/automation/automation-run-completion.ts +++ b/src/main/automation/automation-run-completion.ts @@ -355,7 +355,7 @@ export function processRunCompletion( // // Park the task as `blocked` rather than leaving it at // `in_progress`. finishRun has already cleared the run's - // checkout_run_id and active_run_id, so a task still tagged + // task_leases, so a task still tagged // `in_progress` has no lock, no run, and no path back to // `todo` — heartbeat picks it up, checkoutTask rejects it // (status ∉ todo/backlog), and the assignee is stuck in a diff --git a/src/main/database-base.ts b/src/main/database-base.ts index c313b204..4f9dab76 100644 --- a/src/main/database-base.ts +++ b/src/main/database-base.ts @@ -32,7 +32,6 @@ import { buildModelCapabilityMethods } from "./database/model-capabilities-crud" import { MODEL_CAPABILITY_SEED } from "./model-capability-seed"; import { initializeDatabase } from "./database/schema"; import { SecretVault } from "./vault"; -import { STALE_LOCK_TIMEOUT_MS } from "./constants"; function nowIso() { return new Date().toISOString(); @@ -307,7 +306,7 @@ export class AppDatabaseBase { .run(ts, ...runIds); this.db .prepare( - `update tasks set active_run_id = null where active_run_id in (${placeholders})`, + `delete from task_leases where run_id in (${placeholders})`, ) .run(...runIds); this.db.prepare("update agents set status = 'idle', updated_at = ? where status = 'running'").run(ts); @@ -317,16 +316,23 @@ export class AppDatabaseBase { } recoverStaleLocks(): Array<{ taskId: string; companyId: string; title: string }> { - const cutoff = new Date(Date.now() - STALE_LOCK_TIMEOUT_MS).toISOString(); + const now = new Date().toISOString(); const stale = this.readAll<{ id: string; companyId: string; title: string }>( - `SELECT id, company_id AS companyId, title FROM tasks WHERE execution_locked_at IS NOT NULL AND execution_locked_at < ? AND status = 'in_progress' AND checkout_run_id NOT IN (SELECT id FROM runs WHERE status = 'running')`, - cutoff, + `SELECT t.id, t.company_id AS companyId, t.title + FROM task_leases l + JOIN tasks t ON t.id = l.task_id + LEFT JOIN runs r ON r.id = l.run_id + WHERE l.expires_at < ? + AND t.status = 'in_progress' + AND (r.id IS NULL OR r.status != 'running')`, + now, ); const ts = nowIso(); for (const task of stale) { + this.db.prepare("delete from task_leases where task_id = ?").run(task.id); this.db .prepare( - `UPDATE tasks SET checkout_run_id = NULL, execution_locked_at = NULL, status = 'todo', updated_at = ? WHERE id = ?`, + `UPDATE tasks SET status = 'todo', updated_at = ? WHERE id = ?`, ) .run(ts, task.id); } diff --git a/src/main/database/crash-recovery-helpers.ts b/src/main/database/crash-recovery-helpers.ts index df868b16..9f1eadca 100644 --- a/src/main/database/crash-recovery-helpers.ts +++ b/src/main/database/crash-recovery-helpers.ts @@ -15,8 +15,8 @@ export interface StaleRunDb { interruptRunningRuns(ts: string): number; /** Mark tasks associated with interrupted runs as 'blocked'. */ blockInterruptedTasks(ts: string): void; - /** Clear active_run_id on tasks whose run is now interrupted. */ - clearActiveRunIds(): void; + /** Delete task lease rows whose run is now interrupted. */ + clearInterruptedTaskLeases(): void; /** Reset all running agents to idle. */ resetAgentsToIdle(ts: string): void; /** Release all workspace locks. */ @@ -51,7 +51,7 @@ export function interruptStaleRunsLogic( if (interruptedCount > 0) { db.blockInterruptedTasks(ts); } - db.clearActiveRunIds(); + db.clearInterruptedTaskLeases(); db.resetAgentsToIdle(ts); db.releaseAllWorkspaceLocks(); }); diff --git a/src/main/database/entity-crud.ts b/src/main/database/entity-crud.ts index c305f3e3..7451831d 100644 --- a/src/main/database/entity-crud.ts +++ b/src/main/database/entity-crud.ts @@ -53,7 +53,7 @@ export function buildEntityCrudMethods( getTask(id: string): TaskRecord { const row = getDb() .prepare( - "select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, active_run_id as activeRunId, cost_usd as costUsd, task_type as taskType, requires_user_review as requiresUserReview, requires_filesystem as requiresFilesystem, started_at as startedAt, completed_at as completedAt, cancelled_at as cancelledAt, created_at as createdAt, updated_at as updatedAt from tasks where id = ?", + "select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, (select l.run_id from task_leases l where l.task_id = tasks.id) as activeRunId, cost_usd as costUsd, task_type as taskType, requires_user_review as requiresUserReview, requires_filesystem as requiresFilesystem, started_at as startedAt, completed_at as completedAt, cancelled_at as cancelledAt, created_at as createdAt, updated_at as updatedAt from tasks where id = ?", ) .get(id) as TaskRecord | undefined; if (!row) throw new Error(`Task not found: ${id}`); diff --git a/src/main/database/migrations/v57-task-leases.ts b/src/main/database/migrations/v57-task-leases.ts new file mode 100644 index 00000000..3c60dfd6 --- /dev/null +++ b/src/main/database/migrations/v57-task-leases.ts @@ -0,0 +1,121 @@ +import type { DatabaseSync } from "node:sqlite"; +import { STALE_LOCK_TIMEOUT_MS } from "../../constants"; +import { columnExists, nowIso, tableExists } from "./types"; + +export const TASK_LEASE_SCHEMA_VERSION = 57; +export const TASK_LEASE_SCHEMA_NAME = "task_leases_runtime_ownership"; + +const LEASE_SECONDS = Math.floor(STALE_LOCK_TIMEOUT_MS / 1000); + +export function migrateTaskLeasesV57(db: DatabaseSync): void { + installTaskLeasesTable(db); + backfillTaskLeases(db); + dropTaskOwnershipColumns(db); + db + .prepare("insert or replace into schema_migrations (version, name, applied_at) values (?, ?, ?)") + .run(TASK_LEASE_SCHEMA_VERSION, TASK_LEASE_SCHEMA_NAME, nowIso()); + assertTaskLeasesV57(db); +} + +export function assertTaskLeasesV57(db: DatabaseSync): void { + if (!tableExists(db, "task_leases")) { + throw new Error("Schema v57 drift: task_leases table is missing."); + } + if (!tableExists(db, "tasks")) { + throw new Error("Schema v57 drift: tasks table is missing."); + } + + const leaseColumns = db + .prepare("pragma table_info(task_leases)") + .all() + .map((row) => row as { name: string; type: string; notnull: number; pk: number }); + const expectedLeaseColumns = [ + ["task_id", "TEXT", 0, 1], + ["company_id", "TEXT", 1, 0], + ["run_id", "TEXT", 1, 0], + ["agent_id", "TEXT", 1, 0], + ["acquired_at", "TEXT", 1, 0], + ["renewed_at", "TEXT", 1, 0], + ["expires_at", "TEXT", 1, 0], + ["created_at", "TEXT", 1, 0], + ["updated_at", "TEXT", 1, 0], + ]; + const actualLeaseColumns = leaseColumns.map((column) => [ + column.name, + column.type.toUpperCase(), + column.notnull, + column.pk, + ]); + if (JSON.stringify(actualLeaseColumns) !== JSON.stringify(expectedLeaseColumns)) { + throw new Error("Schema v57 drift: task_leases columns do not match the canonical schema."); + } + + const columns = new Set( + db.prepare("pragma table_info(tasks)").all().map((row) => (row as { name: string }).name), + ); + for (const column of ["active_run_id", "checkout_run_id", "execution_locked_at", "active_session_id"]) { + if (columns.has(column)) { + throw new Error(`Schema v57 drift: tasks.${column} should be stored in task_leases.`); + } + } +} + +function installTaskLeasesTable(db: DatabaseSync): void { + db.exec(` + create table if not exists task_leases ( + task_id text primary key references tasks(id) on delete cascade, + company_id text not null, + run_id text not null unique references runs(id) on delete cascade, + agent_id text not null references agents(id) on delete cascade, + acquired_at text not null, + renewed_at text not null, + expires_at text not null, + created_at text not null, + updated_at text not null + ); + create index if not exists idx_task_leases_company_agent on task_leases(company_id, agent_id); + create index if not exists idx_task_leases_expires_at on task_leases(expires_at); + `); +} + +function backfillTaskLeases(db: DatabaseSync): void { + if (!tableExists(db, "tasks") || !tableExists(db, "runs")) return; + if (!columnExists(db, "tasks", "checkout_run_id") && !columnExists(db, "tasks", "active_run_id")) return; + + const runExpr = [ + columnExists(db, "tasks", "checkout_run_id") ? "t.checkout_run_id" : "NULL", + columnExists(db, "tasks", "active_run_id") ? "t.active_run_id" : "NULL", + ].join(", "); + const lockExpr = columnExists(db, "tasks", "execution_locked_at") + ? "coalesce(t.execution_locked_at, t.updated_at)" + : "t.updated_at"; + const ts = nowIso(); + + db.exec(` + insert or ignore into task_leases ( + task_id, company_id, run_id, agent_id, acquired_at, renewed_at, expires_at, created_at, updated_at + ) + select t.id, + t.company_id, + coalesce(${runExpr}) as run_id, + r.agent_id, + ${lockExpr} as acquired_at, + ${lockExpr} as renewed_at, + strftime('%Y-%m-%dT%H:%M:%fZ', ${lockExpr}, '+${LEASE_SECONDS} seconds') as expires_at, + '${ts}' as created_at, + '${ts}' as updated_at + from tasks t + join runs r on r.id = coalesce(${runExpr}) + where coalesce(${runExpr}) is not null + and r.status in ('queued', 'running') + `); +} + +function dropTaskOwnershipColumns(db: DatabaseSync): void { + if (!tableExists(db, "tasks")) return; + for (const column of ["active_run_id", "checkout_run_id", "execution_locked_at", "active_session_id"]) { + if (columnExists(db, "tasks", column)) { + db.exec(`alter table tasks drop column ${column}`); + } + } +} diff --git a/src/main/database/runs.ts b/src/main/database/runs.ts index ca64889a..45e444b8 100644 --- a/src/main/database/runs.ts +++ b/src/main/database/runs.ts @@ -215,35 +215,26 @@ export function buildRunMethods( input.status, input.summary ?? "", input.errorMessage, input.exitCode, input.signal, input.model, runCostUsd, input.unattributedCost ? 1 : 0, usageStatus, ts, ts, input.runId, - ); + ); if (run.taskId !== null) { - getDb().prepare("update tasks set active_run_id = null, updated_at = ? where id = ?").run(ts, run.taskId); getDb() - .prepare("UPDATE tasks SET checkout_run_id = NULL, execution_locked_at = NULL, updated_at = ? WHERE id = ? AND checkout_run_id = ?") - .run(ts, run.taskId, input.runId); + .prepare("delete from task_leases where task_id = ? and run_id = ?") + .run(run.taskId, input.runId); if (runNeverStarted) { // Roll the task back to `todo` when no *other* live run claims // it. Invariant: `task.status = 'in_progress'` ⇒ a queued or - // running run owns it. We drop the previous - // `checkout_run_id IS NULL AND active_run_id IS NULL` guard - // because the conditional UPDATE two lines up can miss (race: - // checkout_run_id already swapped to a newer run id), which - // would leave the task silently stuck at in_progress. The - // NOT EXISTS clause is the canonical "no live owner" check. + // running run owns it through `task_leases`. getDb() .prepare( `UPDATE tasks SET status = 'todo', - checkout_run_id = NULL, - execution_locked_at = NULL, updated_at = ? WHERE id = ? AND status = 'in_progress' AND NOT EXISTS ( - SELECT 1 FROM runs r - WHERE r.task_id = tasks.id - AND r.id != ? - AND r.status IN ('queued', 'running') + SELECT 1 FROM task_leases l + WHERE l.task_id = tasks.id + AND l.run_id != ? )`, ) .run(ts, run.taskId, input.runId); @@ -291,10 +282,9 @@ export function buildRunMethods( ) .run(summary, "aborted_before_usage", ts, ts, runId); if (taskId !== null) { - getDb().prepare("update tasks set active_run_id = null, updated_at = ? where id = ?").run(ts, taskId); getDb() - .prepare("UPDATE tasks SET checkout_run_id = NULL, execution_locked_at = NULL, updated_at = ? WHERE id = ? AND checkout_run_id = ?") - .run(ts, taskId, runId); + .prepare("delete from task_leases where task_id = ? and run_id = ?") + .run(taskId, runId); } if (workspaceId) { getDb() @@ -308,12 +298,29 @@ export function buildRunMethods( }, hasActiveRun(filter: { companyId: string; taskId?: string; agentId?: string }) { + if (filter.taskId) { + const whereClauses = [ + "l.company_id = ?", + "l.task_id = ?", + "r.status in ('queued', 'running')", + ]; + const params: SQLInputValue[] = [filter.companyId, filter.taskId]; + if (filter.agentId) { + whereClauses.push("r.agent_id = ?"); + params.push(filter.agentId); + } + const row = getDb().prepare( + `select 1 as present + from task_leases l + join runs r on r.id = l.run_id + where ${whereClauses.join(" and ")} + limit 1`, + ).get(...params) as { present: number } | undefined; + return Boolean(row?.present); + } + const whereClauses = ["company_id = ?", "status in ('queued', 'running')"]; const params: SQLInputValue[] = [filter.companyId]; - if (filter.taskId) { - whereClauses.push("task_id = ?"); - params.push(filter.taskId); - } if (filter.agentId) { whereClauses.push("agent_id = ?"); params.push(filter.agentId); diff --git a/src/main/database/schema.ts b/src/main/database/schema.ts index 6f0f1673..77213086 100644 --- a/src/main/database/schema.ts +++ b/src/main/database/schema.ts @@ -9,11 +9,18 @@ import { markBaselineSchema, upgradeLegacyProfileToBaselineV56, } from "./migrations/legacy-to-v56"; +import { + TASK_LEASE_SCHEMA_VERSION, + assertTaskLeasesV57, + migrateTaskLeasesV57, +} from "./migrations/v57-task-leases"; import { type ReadAll, type Transaction, } from "./migrations/types"; +const LATEST_SCHEMA_VERSION = TASK_LEASE_SCHEMA_VERSION; + export function initializeDatabase( db: DatabaseSync, ra: ReadAll, @@ -23,8 +30,16 @@ export function initializeDatabase( db.exec("create table if not exists schema_migrations (version integer primary key, name text not null, applied_at text not null)"); const currentVersion = currentSchemaVersion(db); - if (currentVersion > BASELINE_SCHEMA_VERSION) { - throw new Error(`Profile schema version ${currentVersion} is newer than this app baseline ${BASELINE_SCHEMA_VERSION}.`); + if (currentVersion > LATEST_SCHEMA_VERSION) { + throw new Error(`Profile schema version ${currentVersion} is newer than this app schema ${LATEST_SCHEMA_VERSION}.`); + } + + if (currentVersion === LATEST_SCHEMA_VERSION) { + tx(() => { + seedSettings(db); + assertTaskLeasesV57(db); + }); + return; } if (currentVersion === BASELINE_SCHEMA_VERSION) { @@ -33,6 +48,7 @@ export function initializeDatabase( markBaselineSchema(db); seedSettings(db); assertBaselineSchema(db); + migrateTaskLeasesV57(db); }); return; } @@ -44,6 +60,7 @@ export function initializeDatabase( upgradeLegacyProfileToBaselineV56({ db, readAll: ra }); seedSettings(db); assertBaselineSchema(db); + migrateTaskLeasesV57(db); }); return; } @@ -53,6 +70,7 @@ export function initializeDatabase( markBaselineSchema(db); seedSettings(db); assertBaselineSchema(db); + migrateTaskLeasesV57(db); }); } diff --git a/src/main/database/snapshot.ts b/src/main/database/snapshot.ts index 1b40a4fe..7d2d3810 100644 --- a/src/main/database/snapshot.ts +++ b/src/main/database/snapshot.ts @@ -83,7 +83,7 @@ export function buildSnapshotMethods( ...companyArgs, ), tasks: readAll( - `select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, active_run_id as activeRunId, cost_usd as costUsd, task_type as taskType, requires_user_review as requiresUserReview, requires_filesystem as requiresFilesystem, started_at as startedAt, completed_at as completedAt, cancelled_at as cancelledAt, created_at as createdAt, updated_at as updatedAt from tasks${companyWhere} order by created_at desc limit 2000`, + `select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, (select l.run_id from task_leases l where l.task_id = tasks.id) as activeRunId, cost_usd as costUsd, task_type as taskType, requires_user_review as requiresUserReview, requires_filesystem as requiresFilesystem, started_at as startedAt, completed_at as completedAt, cancelled_at as cancelledAt, created_at as createdAt, updated_at as updatedAt from tasks${companyWhere} order by created_at desc limit 2000`, ...companyArgs, ).map((record) => ({ ...record, requiresUserReview: toBoolean(record.requiresUserReview), requiresFilesystem: toBoolean(record.requiresFilesystem), startedAt: record.startedAt ?? null, completedAt: record.completedAt ?? null, cancelledAt: record.cancelledAt ?? null })) as TaskRecord[], approvals: readAll( diff --git a/src/main/database/task-checkout.ts b/src/main/database/task-checkout.ts index 710e718b..6003a161 100644 --- a/src/main/database/task-checkout.ts +++ b/src/main/database/task-checkout.ts @@ -1,16 +1,25 @@ import type { DatabaseSync } from "node:sqlite"; +import { STALE_LOCK_TIMEOUT_MS } from "../constants"; function isoFromMs(ms: number): string { return new Date(ms).toISOString(); } -function assertRunExists(db: DatabaseSync, runId: string): void { +function loadRunForCheckout( + db: DatabaseSync, + taskId: string, + runId: string, +): { companyId: string; agentId: string; taskId: string | null } { const row = db - .prepare("SELECT 1 AS present FROM runs WHERE id = ?") - .get(runId) as { present: number } | undefined; + .prepare("SELECT company_id as companyId, agent_id as agentId, task_id as taskId FROM runs WHERE id = ?") + .get(runId) as { companyId: string; agentId: string; taskId: string | null } | undefined; if (!row) { throw new Error(`Run not found for task checkout: ${runId}`); } + if (row.taskId !== null && row.taskId !== taskId) { + throw new Error(`Run ${runId} is bound to task ${row.taskId}, not ${taskId}.`); + } + return row; } export function checkoutTaskInTransaction( @@ -18,26 +27,50 @@ export function checkoutTaskInTransaction( taskId: string, runId: string, ): boolean { - assertRunExists(db, runId); + const run = loadRunForCheckout(db, taskId, runId); const nowMs = Date.now(); const ts = isoFromMs(nowMs); const result = db .prepare( `UPDATE tasks - SET checkout_run_id = ?, - active_run_id = ?, - execution_locked_at = ?, - status = 'in_progress', + SET status = 'in_progress', + started_at = coalesce(started_at, ?), updated_at = ? WHERE id = ? - AND checkout_run_id IS NULL - AND status IN ('todo', 'backlog')`, + AND company_id = ? + AND status IN ('todo', 'backlog') + AND NOT EXISTS ( + SELECT 1 FROM task_leases + WHERE task_leases.task_id = tasks.id + )`, ) - .run(runId, runId, ts, ts, taskId); + .run(ts, ts, taskId, run.companyId); if (result.changes === 0) { return false; } + if (run.taskId === null) { + db.prepare("update runs set task_id = ?, updated_at = ? where id = ? and task_id is null").run(taskId, ts, runId); + } + + db + .prepare( + `insert into task_leases ( + task_id, company_id, run_id, agent_id, acquired_at, renewed_at, expires_at, created_at, updated_at + ) values (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .run( + taskId, + run.companyId, + runId, + run.agentId, + ts, + ts, + isoFromMs(nowMs + STALE_LOCK_TIMEOUT_MS), + ts, + ts, + ); + return true; } diff --git a/src/main/database/task-queries.ts b/src/main/database/task-queries.ts index 02e2eebf..52f6aeab 100644 --- a/src/main/database/task-queries.ts +++ b/src/main/database/task-queries.ts @@ -25,7 +25,7 @@ export function buildTaskQueryMethods( ): TaskRecord[] { const placeholders = statuses.map(() => "?").join(","); return readAll( - `select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, active_run_id as activeRunId, cost_usd as costUsd, task_type as taskType, created_at as createdAt, updated_at as updatedAt + `select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, (select l.run_id from task_leases l where l.task_id = tasks.id) as activeRunId, cost_usd as costUsd, task_type as taskType, created_at as createdAt, updated_at as updatedAt from tasks where company_id = ? and assignee_agent_id = ? and status in (${placeholders}) order by @@ -91,7 +91,7 @@ export function buildTaskQueryMethods( ): TaskListByStatusesRow[] { const placeholders = statuses.map(() => "?").join(", "); return readAll( - `select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, active_run_id as activeRunId, cost_usd as costUsd, requires_user_review as requiresUserReview, started_at as startedAt, completed_at as completedAt, cancelled_at as cancelledAt, created_at as createdAt, updated_at as updatedAt from tasks where company_id = ? and status in (${placeholders}) order by created_at desc`, + `select id, company_id as companyId, project_id as projectId, goal_id as goalId, parent_id as parentId, title, description, assignee_agent_id as assigneeAgentId, workspace_id as workspaceId, status, priority, (select l.run_id from task_leases l where l.task_id = tasks.id) as activeRunId, cost_usd as costUsd, requires_user_review as requiresUserReview, started_at as startedAt, completed_at as completedAt, cancelled_at as cancelledAt, created_at as createdAt, updated_at as updatedAt from tasks where company_id = ? and status in (${placeholders}) order by created_at desc`, companyId, ...statuses, ).map((t) => ({ diff --git a/src/main/database/task-sessions-crud.ts b/src/main/database/task-sessions-crud.ts index b6b00bb9..0331ff27 100644 --- a/src/main/database/task-sessions-crud.ts +++ b/src/main/database/task-sessions-crud.ts @@ -7,9 +7,9 @@ import { type ReadAll, type Transaction, nowIso } from "./helpers"; * Agent-scoped session lifecycle. Sessions belong to agents, not * tasks — `agents.session_id` is the canonical pointer, and * `task_sessions` rows are looked up by their own primary key only. - * The pre-migration `tasks.active_session_id` / `task_sessions.task_id` - * columns were dropped by v41; any read that still needs a session - * given a task id resolves it through the task's assignee agent. + * Legacy task/session pointer columns were dropped by v41; any read + * that still needs a session given a task id resolves it through the + * task's assignee agent. */ export function buildTaskSessionMethods( getDb: () => DatabaseSync, diff --git a/src/main/database/task-status-rules.ts b/src/main/database/task-status-rules.ts index 1eb0a161..788a086a 100644 --- a/src/main/database/task-status-rules.ts +++ b/src/main/database/task-status-rules.ts @@ -37,12 +37,5 @@ export function buildTaskStatusTransition( } else { sets.push("cancelled_at = NULL"); } - if (newStatus !== "in_progress") { - sets.push("checkout_run_id = NULL"); - } - if (newStatus === "todo" || newStatus === "backlog") { - sets.push("active_run_id = NULL"); - } - return { sets, values }; } diff --git a/src/main/database/tasks-crud.ts b/src/main/database/tasks-crud.ts index 74271a2c..accd8de0 100644 --- a/src/main/database/tasks-crud.ts +++ b/src/main/database/tasks-crud.ts @@ -158,13 +158,8 @@ export function buildTaskCrudMethods( ) .run("Task deleted.", "aborted_before_usage", ts, ts, run.id); getDb() - .prepare("update tasks set active_run_id = null, updated_at = ? where id = ?") - .run(ts, run.taskId); - getDb() - .prepare( - "UPDATE tasks SET checkout_run_id = NULL, execution_locked_at = NULL, updated_at = ? WHERE id = ? AND checkout_run_id = ?", - ) - .run(ts, run.taskId, run.id); + .prepare("delete from task_leases where task_id = ? and run_id = ?") + .run(run.taskId, run.id); if (run.workspaceId) { getDb() .prepare( diff --git a/src/main/database/tasks-status.ts b/src/main/database/tasks-status.ts index 832ae805..468c8ad8 100644 --- a/src/main/database/tasks-status.ts +++ b/src/main/database/tasks-status.ts @@ -10,6 +10,7 @@ import { chooseBestTaskAssignee } from "./task-assignment"; import { checkoutTaskInTransaction } from "./task-checkout"; import { guardInReviewHasReviewer } from "./task-review-guard"; import { assertValidTaskStatus, buildTaskStatusTransition } from "./task-status-rules"; +import { STALE_LOCK_TIMEOUT_MS } from "../constants"; export function buildTaskStatusMethods( getDb: () => DatabaseSync, @@ -52,6 +53,9 @@ export function buildTaskStatusMethods( `UPDATE tasks SET ${transition.sets.join(", ")} WHERE id = ?`, ) .run(...transition.values, taskId); + if (newStatus !== "in_progress") { + getDb().prepare("delete from task_leases where task_id = ?").run(taskId); + } }); if (comment) { @@ -77,7 +81,8 @@ export function buildTaskStatusMethods( `select t.id, t.company_id as companyId, t.project_id as projectId, t.goal_id as goalId, t.parent_id as parentId, t.title, t.description, t.assignee_agent_id as assigneeAgentId, t.workspace_id as workspaceId, - t.status, t.priority, t.active_run_id as activeRunId, + t.status, t.priority, + (select l.run_id from task_leases l where l.task_id = t.id) as activeRunId, t.cost_usd as costUsd, t.created_at as createdAt, t.updated_at as updatedAt from tasks t @@ -94,21 +99,19 @@ export function buildTaskStatusMethods( }, releaseTaskLock(taskId: string, runId: string): void { - const ts = nowIso(); getDb() - .prepare( - "UPDATE tasks SET checkout_run_id = NULL, execution_locked_at = NULL, updated_at = ? WHERE id = ? AND checkout_run_id = ?", - ) - .run(ts, taskId, runId); + .prepare("delete from task_leases where task_id = ? and run_id = ?") + .run(taskId, runId); }, renewTaskLease(taskId: string, runId: string): boolean { const ts = nowIso(); + const expiresAt = new Date(Date.now() + STALE_LOCK_TIMEOUT_MS).toISOString(); const result = getDb() .prepare( - "UPDATE tasks SET execution_locked_at = ?, updated_at = ? WHERE id = ? AND checkout_run_id = ?", + "update task_leases set renewed_at = ?, expires_at = ?, updated_at = ? where task_id = ? and run_id = ?", ) - .run(ts, ts, taskId, runId); + .run(ts, expiresAt, ts, taskId, runId); return result.changes > 0; }, diff --git a/tests/integration/process-recovery.spec.ts b/tests/integration/process-recovery.spec.ts index 678762ab..9e4cea78 100644 --- a/tests/integration/process-recovery.spec.ts +++ b/tests/integration/process-recovery.spec.ts @@ -184,12 +184,12 @@ function createAndStartRun( * Verifies database consistency after crash recovery: * - No runs in "running" status * - No agents in "running" status - * - No orphaned checkout_run_ids pointing to non-existent runs + * - No orphaned task leases pointing to non-existent runs */ function verifyDbConsistency(db: AppDatabase): { runningRuns: number; runningAgents: number; - orphanedCheckouts: number; + orphanedTaskLeases: number; } { const runningRuns = ( (db as unknown as RawDb).db @@ -203,17 +203,17 @@ function verifyDbConsistency(db: AppDatabase): { .get() as { cnt: number } ).cnt; - const orphanedCheckouts = ( + const orphanedTaskLeases = ( (db as unknown as RawDb).db .prepare( - `SELECT count(*) as cnt FROM tasks t - LEFT JOIN runs r ON t.checkout_run_id = r.id - WHERE t.checkout_run_id IS NOT NULL AND r.id IS NULL`, + `SELECT count(*) as cnt FROM task_leases l + LEFT JOIN runs r ON l.run_id = r.id + WHERE r.id IS NULL`, ) .get() as { cnt: number } ).cnt; - return { runningRuns, runningAgents, orphanedCheckouts }; + return { runningRuns, runningAgents, orphanedTaskLeases }; } // --------------------------------------------------------------------------- @@ -292,20 +292,20 @@ describe("Process Recovery -- Worker Crash Detection", () => { }); // Verify lock is held before crash - const taskBefore = (db as unknown as RawDb).db + const leaseBefore = (db as unknown as RawDb).db .prepare( - "SELECT checkout_run_id FROM tasks WHERE id = ?", + "SELECT run_id FROM task_leases WHERE task_id = ?", ) - .get(taskId) as { checkout_run_id: string | null }; - expect(taskBefore.checkout_run_id).toBe(runId); + .get(taskId) as { run_id: string | null }; + expect(leaseBefore.run_id).toBe(runId); simulateWorkerCrash(db); - // interruptStaleRuns clears active_run_id but does not directly clear checkout_run_id. - // The checkout_run_id is cleared by finishRun or releaseTaskLock. - // However, after crash, the task is set to "blocked" and a subsequent - // stale lock recovery pass handles the checkout_run_id. - // For this test, verify the task status is blocked (the recovery mechanism). + const leaseAfter = (db as unknown as RawDb).db + .prepare("SELECT run_id FROM task_leases WHERE task_id = ?") + .get(taskId); + expect(leaseAfter).toBeUndefined(); + const task = db.core.entity.getTask(taskId); expect(task.status).toBe("blocked"); }); @@ -483,7 +483,7 @@ describe("Process Recovery -- Context Preservation (D-02)", () => { const consistency = verifyDbConsistency(db); expect(consistency.runningRuns).toBe(0); expect(consistency.runningAgents).toBe(0); - expect(consistency.orphanedCheckouts).toBe(0); + expect(consistency.orphanedTaskLeases).toBe(0); }); }); @@ -579,6 +579,6 @@ describe("Process Recovery -- Multiple Concurrent Runs", () => { const consistency = verifyDbConsistency(db); expect(consistency.runningRuns).toBe(0); expect(consistency.runningAgents).toBe(0); - expect(consistency.orphanedCheckouts).toBe(0); + expect(consistency.orphanedTaskLeases).toBe(0); }); }); diff --git a/tests/unit/atomic-checkout.test.ts b/tests/unit/atomic-checkout.test.ts index 09616e68..47abc284 100644 --- a/tests/unit/atomic-checkout.test.ts +++ b/tests/unit/atomic-checkout.test.ts @@ -24,6 +24,12 @@ type RawDb = { }; }; +type TaskLeaseRow = { + task_id: string; + run_id: string; + expires_at: string; +}; + // --------------------------------------------------------------------------- // Helpers // --------------------------------------------------------------------------- @@ -84,6 +90,14 @@ function createTestTask(db: AppDatabase, companyId: string, agentId: string | nu }); } +function getTaskLease(db: AppDatabase, taskId: string): TaskLeaseRow | null { + return ( + ((db as unknown as RawDb).db + .prepare("SELECT task_id, run_id, expires_at FROM task_leases WHERE task_id = ?") + .get(taskId) as TaskLeaseRow | undefined) ?? null + ); +} + // --------------------------------------------------------------------------- // Tests // --------------------------------------------------------------------------- @@ -157,11 +171,29 @@ describe("CAS checkout", () => { const task = db.core.entity.getTask(taskId); expect(task.activeRunId).toBeNull(); expect(task.status).toBe("todo"); - const lock = (db as unknown as RawDb).db - .prepare("SELECT checkout_run_id, execution_locked_at FROM tasks WHERE id = ?") - .get(taskId) as { checkout_run_id: string | null; execution_locked_at: string | null }; - expect(lock.checkout_run_id).toBeNull(); - expect(lock.execution_locked_at).toBeNull(); + expect(getTaskLease(db, taskId)).toBeNull(); + }); + + it("failed checkout keeps a taskless run unbound", () => { + const companyId = createTestCompany(db); + createTestConnector(db); + const workspaceId = createTestWorkspace(db, companyId); + const agentId = createTestAgent(db, companyId, workspaceId); + const taskId = createTestTask(db, companyId, null, workspaceId); + + (db as unknown as RawDb).db.prepare("UPDATE tasks SET status = 'in_progress' WHERE id = ?").run(taskId); + const { id: runId } = db.core.runs.createRun({ + companyId, + taskId: null, + agentId, + workspaceId, + connectorId: "openai-codex", + }); + + expect(db.core.tasks.checkoutTask(taskId, runId)).toBe(false); + expect(db.core.runs.getRun(runId).taskId).toBeNull(); + expect(getTaskLease(db, taskId)).toBeNull(); + expect(db.core.runs.hasActiveRun({ companyId, taskId })).toBe(false); }); it("createTaskRun atomically creates the run and claims the task row lock", () => { @@ -183,10 +215,7 @@ describe("CAS checkout", () => { const task = db.core.entity.getTask(taskId); expect(task.status).toBe("in_progress"); expect(task.activeRunId).toBe(run?.id); - const lock = (db as unknown as RawDb).db - .prepare("SELECT checkout_run_id FROM tasks WHERE id = ?") - .get(taskId) as { checkout_run_id: string | null }; - expect(lock.checkout_run_id).toBe(run?.id); + expect(getTaskLease(db, taskId)?.run_id).toBe(run?.id); }); it("createTaskRun rolls back the run row when checkout loses the race", () => { @@ -269,7 +298,7 @@ describe("CAS checkout", () => { expect(db.core.tasks.checkoutTask(taskId, runId)).toBe(false); }); - it("lock release in finishRun clears checkout fields", () => { + it("lock release in finishRun clears the task lease", () => { const companyId = createTestCompany(db); createTestConnector(db); const workspaceId = createTestWorkspace(db, companyId); @@ -297,15 +326,10 @@ describe("CAS checkout", () => { unattributedCost: false, }); - const row = (db as unknown as RawDb).db - .prepare("SELECT checkout_run_id, execution_locked_at FROM tasks WHERE id = ?") - .get(taskId) as { checkout_run_id: string | null; execution_locked_at: string | null }; - - expect(row.checkout_run_id).toBeNull(); - expect(row.execution_locked_at).toBeNull(); + expect(getTaskLease(db, taskId)).toBeNull(); }); - it("lock release in cancelRunForDeletion clears checkout fields", () => { + it("lock release in cancelRunForDeletion clears the task lease", () => { const companyId = createTestCompany(db); createTestConnector(db); const workspaceId = createTestWorkspace(db, companyId); @@ -326,12 +350,7 @@ describe("CAS checkout", () => { // Deleting the agent triggers cancelRunForDeletion for active runs db.core.agents.deleteAgent(agentId, companyId); - const row = (db as unknown as RawDb).db - .prepare("SELECT checkout_run_id, execution_locked_at FROM tasks WHERE id = ?") - .get(taskId) as { checkout_run_id: string | null; execution_locked_at: string | null }; - - expect(row.checkout_run_id).toBeNull(); - expect(row.execution_locked_at).toBeNull(); + expect(getTaskLease(db, taskId)).toBeNull(); }); it("stale lock recovery resets tasks older than 15 minutes", () => { @@ -351,10 +370,10 @@ describe("CAS checkout", () => { db.core.tasks.checkoutTask(taskId, runId); - // Backdate execution_locked_at to 20 minutes ago + // Backdate the lease expiry to 20 minutes ago. const backdated = new Date(Date.now() - 20 * 60 * 1000).toISOString(); (db as unknown as RawDb).db - .prepare("UPDATE tasks SET execution_locked_at = ? WHERE id = ?") + .prepare("UPDATE task_leases SET expires_at = ? WHERE task_id = ?") .run(backdated, taskId); const recovered = db.recoverStaleLocks(); @@ -366,12 +385,7 @@ describe("CAS checkout", () => { const task = db.core.entity.getTask(taskId); expect(task.status).toBe("todo"); - const row = (db as unknown as RawDb).db - .prepare("SELECT checkout_run_id, execution_locked_at FROM tasks WHERE id = ?") - .get(taskId) as { checkout_run_id: string | null; execution_locked_at: string | null }; - - expect(row.checkout_run_id).toBeNull(); - expect(row.execution_locked_at).toBeNull(); + expect(getTaskLease(db, taskId)).toBeNull(); }); it("stale lock recovery ignores tasks within 15 minute window", () => { @@ -404,10 +418,8 @@ describe("CAS checkout", () => { it("failed dispatch (run never started) rolls task back to 'todo' so heartbeat can retry", () => { // Regression: workspace-lock contention calls finishRun('failed') on a - // run that never reached 'running'. The old code cleared checkout_run_id - // but left task.status='in_progress' — the next checkoutTask pass then - // refused (WHERE status IN ('todo','backlog')), and the scheduler - // deadlocked while logging "lost checkout race" every 15 seconds. + // run that never reached 'running'. The lease must clear and the task + // status must return to 'todo' so the heartbeat retry path can reclaim it. const companyId = createTestCompany(db); createTestConnector(db); const workspaceId = createTestWorkspace(db, companyId); diff --git a/tests/unit/crash-recovery.test.ts b/tests/unit/crash-recovery.test.ts index 4d503b91..d20058e1 100644 --- a/tests/unit/crash-recovery.test.ts +++ b/tests/unit/crash-recovery.test.ts @@ -31,9 +31,9 @@ function makeStaleRunDb(runningCount: number): StaleRunDb & { calls: string[] } blockInterruptedTasks: vi.fn((_ts: string) => { calls.push("blockInterruptedTasks"); }) as unknown as StaleRunDb["blockInterruptedTasks"], - clearActiveRunIds: vi.fn(() => { - calls.push("clearActiveRunIds"); - }) as unknown as StaleRunDb["clearActiveRunIds"], + clearInterruptedTaskLeases: vi.fn(() => { + calls.push("clearInterruptedTaskLeases"); + }) as unknown as StaleRunDb["clearInterruptedTaskLeases"], resetAgentsToIdle: vi.fn((_ts: string) => { calls.push("resetAgentsToIdle"); }) as unknown as StaleRunDb["resetAgentsToIdle"], @@ -66,16 +66,16 @@ describe("Crash Recovery Detection", () => { expect(db.calls).toContain("blockInterruptedTasks"); }); - it("does NOT mark tasks as blocked when no runs were interrupted", () => { + it("leaves task blocking step idle when no runs were interrupted", () => { const db = makeStaleRunDb(0); interruptStaleRunsLogic(db, transaction, nowIso); expect(db.calls).not.toContain("blockInterruptedTasks"); }); - it("always clears active_run_id and resets agents to idle", () => { + it("always clears interrupted task leases and resets agents to idle", () => { const db = makeStaleRunDb(0); interruptStaleRunsLogic(db, transaction, nowIso); - expect(db.calls).toContain("clearActiveRunIds"); + expect(db.calls).toContain("clearInterruptedTaskLeases"); expect(db.calls).toContain("resetAgentsToIdle"); }); diff --git a/tests/unit/database.test.ts b/tests/unit/database.test.ts index a8672f70..c84c82ad 100644 --- a/tests/unit/database.test.ts +++ b/tests/unit/database.test.ts @@ -1,6 +1,7 @@ import { join } from "node:path"; import { tmpdir } from "node:os"; import { mkdtemp, rm } from "node:fs/promises"; +import { DatabaseSync } from "node:sqlite"; import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; beforeAll(() => { @@ -17,6 +18,8 @@ vi.mock("electron", () => ({ })); import { AppDatabase } from "../../src/main/database"; +import { installBaselineSchema } from "../../src/main/database/migrations/baseline-v56"; +import { markBaselineSchema } from "../../src/main/database/migrations/legacy-to-v56"; describe("AppDatabase", () => { let tempDir: string; @@ -33,18 +36,25 @@ describe("AppDatabase", () => { await rm(tempDir, { recursive: true, force: true }); }); - it("keeps task runtime state on the tasks row", () => { - const shadowTables = db.queryAll<{ name: string }>( + it("keeps task runtime ownership in task_leases", () => { + const leaseTables = db.queryAll<{ name: string }>( "select name from sqlite_master where type = 'table' and name = 'task_leases'", ); - const phaseColumns = db.queryAll<{ name: string }>("pragma table_info(tasks)") - .filter((column) => column.name === "phase"); + const removedTaskColumns = new Set([ + "phase", + "active_run_id", + "checkout_run_id", + "execution_locked_at", + "active_session_id", + ]); + const taskColumns = db.queryAll<{ name: string }>("pragma table_info(tasks)") + .filter((column) => removedTaskColumns.has(column.name)); const phaseTriggers = db.queryAll<{ name: string }>( "select name from sqlite_master where type = 'trigger' and name like 'trg_tasks_phase_from_status_%'", ); - expect(shadowTables).toEqual([]); - expect(phaseColumns).toEqual([]); + expect(leaseTables).toEqual([{ name: "task_leases" }]); + expect(taskColumns).toEqual([]); expect(phaseTriggers).toEqual([]); }); @@ -462,11 +472,14 @@ describe("AppDatabase", () => { expect(db.core.agents.isAgentPendingApproval(pendingAgentId)).toBe(true); }); - it("installs the v56 baseline once on a fresh profile", () => { + it("installs the v56 baseline and v57 task leases once on a fresh profile", () => { const migrations = db.queryAll<{ version: number; name: string }>( "select version, name from schema_migrations order by version", ); - expect(migrations).toEqual([{ version: 56, name: "baseline_v56" }]); + expect(migrations).toEqual([ + { version: 56, name: "baseline_v56" }, + { version: 57, name: "task_leases_runtime_ownership" }, + ]); db.close(); db = new AppDatabase(tempDir); @@ -475,7 +488,51 @@ describe("AppDatabase", () => { const migrationsAfterReinit = db.queryAll<{ version: number; name: string }>( "select version, name from schema_migrations order by version", ); - expect(migrationsAfterReinit).toEqual([{ version: 56, name: "baseline_v56" }]); + expect(migrationsAfterReinit).toEqual([ + { version: 56, name: "baseline_v56" }, + { version: 57, name: "task_leases_runtime_ownership" }, + ]); + }); + + it("migrates v56 task ownership columns into task_leases", async () => { + const legacyDir = await mkdtemp(join(tmpdir(), "agentcompany-v56-lease-")); + const rawDb = new DatabaseSync(join(legacyDir, "profile.sqlite")); + const ts = "2026-04-24T00:00:00.000Z"; + try { + installBaselineSchema(rawDb); + markBaselineSchema(rawDb); + rawDb + .prepare("insert into companies (id, name, description, status, created_at, updated_at) values (?, ?, ?, ?, ?, ?)") + .run("company-1", "Lease Co", "", "active", ts, ts); + rawDb + .prepare("insert into agents (id, company_id, name, role, title, status, reports_to, connector_id, workspace_id, capabilities, created_at, updated_at) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)") + .run("agent-1", "company-1", "Worker", "IC", "Worker", "active", null, "openai-codex", null, "Executes", ts, ts); + rawDb + .prepare("insert into tasks (id, company_id, title, description, assignee_agent_id, status, priority, active_run_id, checkout_run_id, execution_locked_at, created_at, updated_at) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)") + .run("task-1", "company-1", "Lease task", "", "agent-1", "in_progress", "medium", "run-1", "run-1", ts, ts, ts); + rawDb + .prepare("insert into runs (id, company_id, task_id, agent_id, connector_id, status, summary, created_at, updated_at) values (?, ?, ?, ?, ?, ?, ?, ?, ?)") + .run("run-1", "company-1", "task-1", "agent-1", "openai-codex", "queued", "", ts, ts); + } finally { + rawDb.close(); + } + + const legacyAppDb = new AppDatabase(legacyDir); + try { + legacyAppDb.init(); + expect( + legacyAppDb.queryAll<{ taskId: string; runId: string; agentId: string }>( + "select task_id as taskId, run_id as runId, agent_id as agentId from task_leases", + ), + ).toEqual([{ taskId: "task-1", runId: "run-1", agentId: "agent-1" }]); + const columns = legacyAppDb.queryAll<{ name: string }>("pragma table_info(tasks)").map((column) => column.name); + expect(columns).not.toContain("active_run_id"); + expect(columns).not.toContain("checkout_run_id"); + expect(columns).not.toContain("execution_locked_at"); + } finally { + legacyAppDb.close(); + await rm(legacyDir, { recursive: true, force: true }); + } }); it("acquires and releases workspace locks", () => { diff --git a/tests/unit/edge-case-concurrent.test.ts b/tests/unit/edge-case-concurrent.test.ts index 8efab2d6..e3e5c759 100644 --- a/tests/unit/edge-case-concurrent.test.ts +++ b/tests/unit/edge-case-concurrent.test.ts @@ -23,6 +23,12 @@ type RawDb = { }; }; +type TaskLeaseRow = { + task_id: string; + run_id: string; + expires_at: string; +}; + // --------------------------------------------------------------------------- // Helpers // --------------------------------------------------------------------------- @@ -83,6 +89,14 @@ function createTestTask(db: AppDatabase, companyId: string, workspaceId: string) }); } +function getTaskLease(db: AppDatabase, taskId: string): TaskLeaseRow | null { + return ( + ((db as unknown as RawDb).db + .prepare("SELECT task_id, run_id, expires_at FROM task_leases WHERE task_id = ?") + .get(taskId) as TaskLeaseRow | undefined) ?? null + ); +} + // --------------------------------------------------------------------------- // Tests — Edge Case: Concurrent Task Checkout (D-05.1) // --------------------------------------------------------------------------- @@ -234,15 +248,10 @@ describe("Edge Case: Concurrent Task Checkout (D-05.1)", () => { }); expect(db.core.tasks.checkoutTask(taskId, runIdA)).toBe(true); - // Release the lock (sets checkout_run_id = NULL) + // Release the lease while leaving the task lifecycle unchanged. db.core.tasks.releaseTaskLock(taskId, runIdA); - // Verify lock fields are cleared - const row = (db as unknown as RawDb).db - .prepare("SELECT checkout_run_id, execution_locked_at FROM tasks WHERE id = ?") - .get(taskId) as { checkout_run_id: string | null; execution_locked_at: string | null }; - expect(row.checkout_run_id).toBeNull(); - expect(row.execution_locked_at).toBeNull(); + expect(getTaskLease(db, taskId)).toBeNull(); // Task is still in_progress though, so a new checkout fails // (CAS checks status IN ('todo', 'backlog')) @@ -273,9 +282,9 @@ describe("Edge Case: Concurrent Task Checkout (D-05.1)", () => { }); expect(db.core.tasks.checkoutTask(taskId, runIdA)).toBe(true); - // Backdate execution_locked_at to well past the stale timeout + // Backdate the lease expiry past the stale timeout. (db as unknown as RawDb).db - .prepare("UPDATE tasks SET execution_locked_at = '2020-01-01T00:00:00Z' WHERE id = ?") + .prepare("UPDATE task_leases SET expires_at = '2020-01-01T00:00:00Z' WHERE task_id = ?") .run(taskId); const recovered = db.recoverStaleLocks(); @@ -283,12 +292,7 @@ describe("Edge Case: Concurrent Task Checkout (D-05.1)", () => { expect(recovered[0].taskId).toBe(taskId); expect(recovered[0].companyId).toBe(companyId); - // Checkout fields should be cleared - const row = (db as unknown as RawDb).db - .prepare("SELECT checkout_run_id, execution_locked_at FROM tasks WHERE id = ?") - .get(taskId) as { checkout_run_id: string | null; execution_locked_at: string | null }; - expect(row.checkout_run_id).toBeNull(); - expect(row.execution_locked_at).toBeNull(); + expect(getTaskLease(db, taskId)).toBeNull(); }); it("after stale recovery, different agent can checkout the recovered task", () => { @@ -309,9 +313,9 @@ describe("Edge Case: Concurrent Task Checkout (D-05.1)", () => { }); expect(db.core.tasks.checkoutTask(taskId, runIdA)).toBe(true); - // Backdate to make stale + // Backdate the lease to make it stale. (db as unknown as RawDb).db - .prepare("UPDATE tasks SET execution_locked_at = '2020-01-01T00:00:00Z' WHERE id = ?") + .prepare("UPDATE task_leases SET expires_at = '2020-01-01T00:00:00Z' WHERE task_id = ?") .run(taskId); // Recover stale locks (sets status back to "todo") diff --git a/tests/unit/legacy-baseline-migration.test.ts b/tests/unit/legacy-baseline-migration.test.ts index 4e0d0210..98686e2b 100644 --- a/tests/unit/legacy-baseline-migration.test.ts +++ b/tests/unit/legacy-baseline-migration.test.ts @@ -171,6 +171,7 @@ describe("legacy to v56 baseline migration", () => { try { expect(appDb.queryAll("select version, name from schema_migrations")).toEqual([ { version: 56, name: "baseline_v56" }, + { version: 57, name: "task_leases_runtime_ownership" }, ]); expect(appDb.queryAll<{ companyId: string }>("select company_id as companyId from secrets where id = 'secret-1'")[0]) .toEqual({ companyId: "company-1" }); @@ -202,6 +203,7 @@ describe("legacy to v56 baseline migration", () => { try { expect(appDb.queryAll("select version, name from schema_migrations")).toEqual([ { version: 56, name: "baseline_v56" }, + { version: 57, name: "task_leases_runtime_ownership" }, ]); expect(appDb.queryAll("select id from budget_policies where scope_type = 'agent' and scope_id = 'agent-1'")) .toHaveLength(1); @@ -360,6 +362,7 @@ describe("legacy to v56 baseline migration", () => { .toEqual([{ agentId: "agent-1", adapterType: "openai-codex" }]); expect(appDb.queryAll("select version, name from schema_migrations")).toEqual([ { version: 56, name: "baseline_v56" }, + { version: 57, name: "task_leases_runtime_ownership" }, ]); } finally { appDb.close(); diff --git a/tests/unit/requires-filesystem-migration.test.ts b/tests/unit/requires-filesystem-migration.test.ts index 457ddb34..338a949c 100644 --- a/tests/unit/requires-filesystem-migration.test.ts +++ b/tests/unit/requires-filesystem-migration.test.ts @@ -115,8 +115,13 @@ describe("legacy to v56 baseline migration", () => { { id: "task-cognitive", requiresFilesystem: 0 }, { id: "task-inflight", requiresFilesystem: 1 }, ]); - const migration = appDb.queryAll<{ version: number }>("select version from schema_migrations where version = 56")[0]; - expect(migration?.version).toBe(56); + const migrations = appDb.queryAll<{ version: number; name: string }>( + "select version, name from schema_migrations where version in (56, 57) order by version", + ); + expect(migrations).toEqual([ + { version: 56, name: "baseline_v56" }, + { version: 57, name: "task_leases_runtime_ownership" }, + ]); } finally { appDb.close(); }