fix(api): make task updates transactional

This commit is contained in:
ZacharyZhang-NY committed 2026-04-22 14:00:24 -04:00
1 parent 19f8494d09
commit 7a1dd647aa
8 files changed
+584 -178

No files matched your search

+18
View File
@@ -0,0 +1,18 @@
import type { ReadAll } from "./helpers";
export function guardInReviewHasReviewer(readAll: ReadAll, taskId: string): void {
const row = readAll<{ requiresUserReview: number; reportsTo: string | null }>(
`select t.requires_user_review as requiresUserReview,
a.reports_to as reportsTo
from tasks t
left join agents a on a.id = t.assignee_agent_id
where t.id = ?`,
taskId,
)[0];
if (!row) return;
if (row.requiresUserReview === 1) return;
if (row.reportsTo !== null) return;
throw new Error(
`Task ${taskId} has a top-level assignee and no reviewer: cannot enter in_review without requires_user_review=1.`,
);
}
+60
View File
@@ -0,0 +1,60 @@
export const VALID_TASK_STATUSES = new Set([
"backlog",
"todo",
"in_progress",
"in_review",
"done",
"blocked",
"cancelled",
]);
const STATUSES_THAT_CLEAR_LEASE = new Set<string>([
"done",
"cancelled",
"blocked",
"todo",
"backlog",
]);
export function assertValidTaskStatus(status: string): void {
if (!VALID_TASK_STATUSES.has(status)) {
throw new Error(`Invalid task status: ${status}`);
}
}
export function buildTaskStatusTransition(
newStatus: string,
timestamp: string,
): { sets: string[]; values: (string | null)[]; clearsLease: boolean } {
const sets: string[] = ["status = ?", "updated_at = ?"];
const values: (string | null)[] = [newStatus, timestamp];
if (newStatus === "in_progress") {
sets.push("started_at = COALESCE(started_at, ?)");
values.push(timestamp);
}
if (newStatus === "done") {
sets.push("completed_at = ?");
values.push(timestamp);
} else {
sets.push("completed_at = NULL");
}
if (newStatus === "cancelled") {
sets.push("cancelled_at = ?");
values.push(timestamp);
} 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,
clearsLease: STATUSES_THAT_CLEAR_LEASE.has(newStatus),
};
}
+9 -97
View File
@@ -8,6 +8,8 @@ import {
} from "./helpers";
import { chooseBestTaskAssignee } from "./task-assignment";
import { TASK_LEASE_TTL_MS, checkoutTaskInTransaction } from "./task-checkout";
import { guardInReviewHasReviewer } from "./task-review-guard";
import { assertValidTaskStatus, buildTaskStatusTransition } from "./task-status-rules";
export function buildTaskStatusMethods(
getDb: () => DatabaseSync,
@@ -18,20 +20,6 @@ export function buildTaskStatusMethods(
addActivity: (input: { companyId: string; actor: string; action: string; entityType: string; entityId: string; detail: string }) => void;
},
) {
// Terminal phases and "returned to queue" phases release any existing
// lease: once a task is done/cancelled/blocked or bounced back to
// todo/backlog the execution grant no longer has a referent. We mirror
// that in `task_leases` explicitly — the status→phase trigger only
// touches the phase column and leaves the lease row behind, producing
// an orphan that the dev-mode consistency probe surfaces as a
// divergence. Production would silently drift instead.
const STATUSES_THAT_CLEAR_LEASE = new Set<string>([
"done",
"cancelled",
"blocked",
"todo",
"backlog",
]);
function incrementActiveSprintVelocity(companyId: string, delta: number): void {
const ts = nowIso();
const today = ts.slice(0, 10);
@@ -42,106 +30,30 @@ export function buildTaskStatusMethods(
.run(delta, ts, companyId, today, today);
}
const VALID_TASK_STATUSES = new Set([
"backlog",
"todo",
"in_progress",
"in_review",
"done",
"blocked",
"cancelled",
]);
/**
* Guard against the "in_review dead letter": a task whose assignee is
* a top-level agent (reports_to IS NULL, the DB invariant CEO marker)
* and that does not require user review has no reachable reviewer —
* no manager to walk up to, and no human gate. Without this guard the
* CEO heartbeat loops forever re-proposing the same approval.
*
* The right root fix is to reject the transition at source: a caller
* that moves such a task into `in_review` must first set
* requires_user_review=1 (escalate to human) or route differently
* (advance to done, hire a reviewer, bounce to todo). Previous
* versions silently resurrected these tasks at every DB init; this
* replaces the startup side-effect with a clear runtime error.
*/
function guardInReviewHasReviewer(taskId: string): void {
const row = readAll<{ requiresUserReview: number; reportsTo: string | null }>(
`select t.requires_user_review as requiresUserReview,
a.reports_to as reportsTo
from tasks t
left join agents a on a.id = t.assignee_agent_id
where t.id = ?`,
taskId,
)[0];
if (!row) return;
if (row.requiresUserReview === 1) return;
if (row.reportsTo !== null) return;
throw new Error(
`Task ${taskId} has a top-level assignee and no reviewer: cannot enter in_review without requires_user_review=1.`,
);
}
return {
advanceTaskStatus(
taskId: string,
newStatus: string,
comment?: string,
): { previousStatus: string } {
if (!VALID_TASK_STATUSES.has(newStatus)) {
throw new Error(`Invalid task status: ${newStatus}`);
}
assertValidTaskStatus(newStatus);
const task = getDb()
.prepare("select status, company_id from tasks where id = ?")
.get(taskId) as { status: string; company_id: string } | undefined;
if (!task) throw new Error(`Task not found: ${taskId}`);
const previousStatus = task.status;
if (previousStatus === newStatus) return { previousStatus };
if (newStatus === "in_review") guardInReviewHasReviewer(taskId);
if (newStatus === "in_review") guardInReviewHasReviewer(readAll, taskId);
const ts = nowIso();
// Status side-effect timestamps (Paperclip pattern)
const timestampSets: string[] = ["status = ?", "updated_at = ?"];
const timestampValues: (string | null)[] = [newStatus, ts];
if (newStatus === "in_progress") {
// Set startedAt only once (first time entering in_progress)
timestampSets.push("started_at = COALESCE(started_at, ?)");
timestampValues.push(ts);
}
if (newStatus === "done") {
timestampSets.push("completed_at = ?");
timestampValues.push(ts);
} else {
timestampSets.push("completed_at = NULL");
}
if (newStatus === "cancelled") {
timestampSets.push("cancelled_at = ?");
timestampValues.push(ts);
} else {
timestampSets.push("cancelled_at = NULL");
}
if (newStatus !== "in_progress") {
timestampSets.push("checkout_run_id = NULL");
}
// Returning a task to todo / backlog drops any active-run claim
// as well — heartbeat picks candidate tasks by !activeRunId, so
// leaving a stale pointer here (e.g. after startup recovery of a
// crashed run) silently hides the task from every future wake.
if (newStatus === "todo" || newStatus === "backlog") {
timestampSets.push("active_run_id = NULL");
}
const shouldClearLease = STATUSES_THAT_CLEAR_LEASE.has(newStatus);
const transition = buildTaskStatusTransition(newStatus, ts);
transaction(() => {
getDb()
.prepare(
// eslint-disable-next-line no-restricted-syntax -- B2 Phase 1 grandfathered: advanceTaskStatus is the sanctioned dynamic-SQL helper; timestampSets[0] is always the literal "status = ?". Status→phase trigger plus the explicit DELETE FROM task_leases below keep the shadow surface coherent.
`UPDATE tasks SET ${timestampSets.join(", ")} WHERE id = ?`,
// eslint-disable-next-line no-restricted-syntax -- B2 Phase 1 grandfathered: advanceTaskStatus is the sanctioned dynamic-SQL helper; transition.sets[0] is always the literal "status = ?". Status→phase trigger plus the explicit DELETE FROM task_leases below keep the shadow surface coherent.
`UPDATE tasks SET ${transition.sets.join(", ")} WHERE id = ?`,
)
.run(...timestampValues, taskId);
if (shouldClearLease) {
.run(...transition.values, taskId);
if (transition.clearsLease) {
// Drop any live lease row in the same transaction so no
// caller observes the intermediate state of "task in terminal
// or backlog phase, lease still present". `releaseTaskLock`
+261
View File
@@ -0,0 +1,261 @@
import type { DatabaseSync } from "node:sqlite";
import type { ReadAll, Transaction } from "./helpers";
import { normalizeOptionalId, nowIso } from "./helpers";
import { guardInReviewHasReviewer } from "./task-review-guard";
import { assertValidTaskStatus, buildTaskStatusTransition } from "./task-status-rules";
type TaskUpdatePatch = {
title: string;
description: string;
status: string;
priority: string;
projectId?: string | null;
goalId?: string | null;
parentId?: string | null;
assigneeAgentId?: string | null;
workspaceId?: string | null;
requiresFilesystem: boolean;
};
type ReviewDocument = { id: string; title: string } & Record<string, unknown>;
type TaskReviewRequest = {
reviewerAgentName: string;
documents: ReviewDocument[];
};
type TaskUpdateActivity = {
action: string;
detail: string;
};
type ApprovalRequestInput = {
companyId: string;
relatedTaskId?: string | null;
requestedByAgentId?: string | null;
relatedAgentId?: string | null;
type: string;
payloadSummary: string;
impactSummary: string;
payloadJson?: string | null;
};
type TaskUpdateResult = {
previousStatus: string;
newStatus: string;
previousAssigneeAgentId: string | null;
newAssigneeAgentId: string | null;
parentId: string | null;
reviewRequested: boolean;
statusChanged: boolean;
};
export function buildTaskUpdateMethods(
getDb: () => DatabaseSync,
readAll: ReadAll,
transaction: Transaction,
deps: {
addComment: (input: { companyId: string; taskId: string; authorAgentId: string | null; authorName: string; body: string }) => void;
addActivity: (input: { companyId: string; actor: string; action: string; entityType: string; entityId: string; detail: string }) => void;
requestApproval: (input: ApprovalRequestInput) => string;
setHeartbeat: (agentId: string, enabled: boolean, intervalSec: number) => void;
},
) {
function incrementActiveSprintVelocity(companyId: string, delta: number): void {
const ts = nowIso();
const today = ts.slice(0, 10);
getDb()
.prepare(
"update sprints set completed_points = max(0, completed_points + ?), updated_at = ? where company_id = ? and status = 'active' and start_date <= ? and end_date >= ?",
)
.run(delta, ts, companyId, today, today);
}
function pauseAssigneeHeartbeat(agentId: string | null): void {
if (!agentId) return;
const agent = getDb()
.prepare("select heartbeat_enabled as heartbeatEnabled, heartbeat_interval_sec as heartbeatIntervalSec from agents where id = ?")
.get(agentId) as { heartbeatEnabled: number; heartbeatIntervalSec: number | null } | undefined;
if (!agent?.heartbeatEnabled) return;
deps.setHeartbeat(agentId, false, agent.heartbeatIntervalSec || 120);
}
function requestDeliverableReview(input: {
companyId: string;
taskId: string;
actorAgentId: string;
taskTitle: string;
currentAssigneeAgentId: string | null;
reviewRequest: TaskReviewRequest;
}): void {
const docTitles = input.reviewRequest.documents.map((document) => document.title).join(", ");
deps.requestApproval({
companyId: input.companyId,
relatedTaskId: input.taskId,
requestedByAgentId: input.actorAgentId,
type: "deliverable_review",
payloadSummary: `Review deliverable: ${docTitles || input.taskTitle}`.slice(0, 280),
impactSummary: `${input.reviewRequest.reviewerAgentName} reviewed and approved. Awaiting your final review.`.slice(0, 400),
payloadJson: JSON.stringify({
documentIds: input.reviewRequest.documents.map((document) => document.id),
documents: input.reviewRequest.documents,
}),
});
getDb()
.prepare("update tasks set requires_user_review = 0, updated_at = ? where id = ? and company_id = ?")
.run(nowIso(), input.taskId, input.companyId);
pauseAssigneeHeartbeat(input.currentAssigneeAgentId);
deps.addComment({
companyId: input.companyId,
taskId: input.taskId,
authorAgentId: null,
authorName: "System",
body: "Deliverable submitted for board review.",
});
}
return {
updateTaskWithReviewSideEffects(input: {
taskId: string;
companyId: string;
actorAgentId: string;
actorName: string;
task: TaskUpdatePatch;
comment?: string;
reviewRequest?: TaskReviewRequest | null;
activity: TaskUpdateActivity;
}): TaskUpdateResult {
assertValidTaskStatus(input.task.status);
const assigneeAgentId = normalizeOptionalId(input.task.assigneeAgentId);
if (input.task.status === "in_progress" && !assigneeAgentId) {
throw new Error("Tasks in progress require an assignee.");
}
return transaction(() => {
const current = getDb()
.prepare(
"select id, company_id as companyId, status, title, assignee_agent_id as assigneeAgentId, parent_id as parentId, requires_user_review as requiresUserReview from tasks where id = ? and company_id = ?",
)
.get(input.taskId, input.companyId) as
| {
id: string;
companyId: string;
status: string;
title: string;
assigneeAgentId: string | null;
parentId: string | null;
requiresUserReview: number;
}
| undefined;
if (!current) {
throw new Error(`Task not found: ${input.taskId}`);
}
const reviewRequested =
input.task.status === "done" &&
current.requiresUserReview === 1 &&
input.reviewRequest !== null &&
input.reviewRequest !== undefined;
if (input.task.status === "done" && current.requiresUserReview === 1 && !input.reviewRequest) {
throw new Error(`Task ${input.taskId} requires a deliverable review request before it can be marked done.`);
}
const newStatus = reviewRequested ? "in_review" : input.task.status;
assertValidTaskStatus(newStatus);
if (reviewRequested) {
requestDeliverableReview({
companyId: input.companyId,
taskId: input.taskId,
actorAgentId: input.actorAgentId,
taskTitle: current.title,
currentAssigneeAgentId: current.assigneeAgentId,
reviewRequest: input.reviewRequest!,
});
}
const statusChanged = current.status !== newStatus;
if (statusChanged && newStatus === "in_review") {
guardInReviewHasReviewer(readAll, input.taskId);
}
const ts = nowIso();
const sets = [
"project_id = ?",
"goal_id = ?",
"parent_id = ?",
"title = ?",
"description = ?",
"assignee_agent_id = ?",
"workspace_id = ?",
"priority = ?",
"requires_filesystem = ?",
];
const values: Array<string | number | null> = [
normalizeOptionalId(input.task.projectId),
normalizeOptionalId(input.task.goalId),
normalizeOptionalId(input.task.parentId),
input.task.title,
input.task.description,
assigneeAgentId,
normalizeOptionalId(input.task.workspaceId),
input.task.priority,
input.task.requiresFilesystem ? 1 : 0,
];
const transition = statusChanged ? buildTaskStatusTransition(newStatus, ts) : null;
if (transition) {
sets.push(...transition.sets);
values.push(...transition.values);
} else {
sets.push("status = ?", "updated_at = ?");
values.push(newStatus, ts);
}
const updateResult = getDb()
// eslint-disable-next-line no-restricted-syntax -- R-03 sanctioned domain update helper: SET clauses come from fixed literals only; status->phase trigger and explicit task_leases delete keep the dual-write surface coherent inside this transaction.
.prepare(`update tasks set ${sets.join(", ")} where id = ? and company_id = ?`)
.run(...values, input.taskId, input.companyId);
if (updateResult.changes === 0) {
throw new Error(`Task not found: ${input.taskId}`);
}
if (transition?.clearsLease) {
getDb()
.prepare("DELETE FROM task_leases WHERE task_id = ?")
.run(input.taskId);
}
if (newStatus === "done" && current.status !== "done") {
incrementActiveSprintVelocity(input.companyId, 1);
}
if (current.status === "done" && newStatus !== "done") {
incrementActiveSprintVelocity(input.companyId, -1);
}
if (input.comment) {
deps.addComment({
companyId: input.companyId,
taskId: input.taskId,
authorAgentId: input.actorAgentId,
authorName: input.actorName,
body: input.comment,
});
}
deps.addActivity({
companyId: input.companyId,
actor: input.actorAgentId,
action: input.activity.action,
entityType: "task",
entityId: input.taskId,
detail: input.activity.detail,
});
return {
previousStatus: current.status,
newStatus,
previousAssigneeAgentId: current.assigneeAgentId,
newAssigneeAgentId: assigneeAgentId,
parentId: normalizeOptionalId(input.task.parentId),
reviewRequested,
statusChanged,
};
});
},
};
}
+13
View File
@@ -5,6 +5,7 @@ import { buildTaskQueryMethods } from "./task-queries";
import { buildTaskSessionMethods } from "./task-sessions-crud";
import { buildTaskStatusMethods } from "./tasks-status";
import { buildTasksPhaseMethods } from "./tasks-phase";
import { buildTaskUpdateMethods } from "./tasks-update";
export function buildTaskMethods(
getDb: () => DatabaseSync,
@@ -13,6 +14,17 @@ export function buildTaskMethods(
deps: {
addComment: (input: { companyId: string; taskId: string; authorAgentId: string | null; authorName: string; body: string }) => void;
addActivity: (input: { companyId: string; actor: string; action: string; entityType: string; entityId: string; detail: string }) => void;
requestApproval: (input: {
companyId: string;
relatedTaskId?: string | null;
requestedByAgentId?: string | null;
relatedAgentId?: string | null;
type: string;
payloadSummary: string;
impactSummary: string;
payloadJson?: string | null;
}) => string;
setHeartbeat: (agentId: string, enabled: boolean, intervalSec: number) => void;
},
) {
return {
@@ -20,6 +32,7 @@ export function buildTaskMethods(
...buildTaskQueryMethods(getDb, readAll, transaction),
...buildTaskSessionMethods(getDb, readAll, transaction),
...buildTaskStatusMethods(getDb, readAll, transaction, deps),
...buildTaskUpdateMethods(getDb, readAll, transaction, deps),
...buildTasksPhaseMethods(getDb),
};
}