feat(04-01): implement CAS checkout, lock release, stale recovery, and migration 16

- Rewrite checkoutTask to use runId-based CAS with checkout_run_id IS NULL guard
- Add releaseTaskLock and recoverStaleLocks methods
- Add lock release in finishRun and cancelRunForDeletion exit paths
- Add migration 16: pending_wakes to wakeup_requests cutover with DROP TABLE
- Replace all pending_wakes references with wakeup_requests
- Add wakeup lifecycle CRUD: enqueue, claim, complete, defer, promote, list
- Update api-server, orchestration-service, worker-service callers
- Update existing database.test.ts for new checkoutTask signature
This commit is contained in:
ZacharyZhang-NY committed 2026-03-26 18:56:54 -04:00
1 parent 02e87dc06d
commit 08935b3379
5 files changed
+326 -71

No files matched your search

+14 -7
View File
@@ -145,9 +145,14 @@ export function createWorkerService({
}
for (const agentId of affectedAgentIds) {
const pendingWakes = db.consumePendingWakes(agentId);
for (const wake of pendingWakes) {
wakeAgentIfPossible(agentId, wake.companyId, wake.trigger);
// Promote any deferred wakes back to queued, then process queued wakes
const agentRuns = snapshot.runs.filter((r) => r.agentId === agentId);
for (const companyId of new Set(agentRuns.map((r) => r.companyId))) {
db.promoteWakeupRequests(agentId, companyId);
const wakes = db.listWakeupRequests(companyId, { agentId, status: "queued" });
for (const wake of wakes) {
wakeAgentIfPossible(agentId, wake.companyId, wake.source);
}
}
}
@@ -381,10 +386,12 @@ export function createWorkerService({
}
}
const pendingWakes = db.consumePendingWakes(run.agentId);
if (pendingWakes.length > 0 && !isQuitting() && !destroyed) {
for (const wake of pendingWakes) {
wakeAgentIfPossible(run.agentId, wake.companyId, wake.trigger);
// Promote any deferred wakes back to queued after run completes
db.promoteWakeupRequests(run.agentId, run.companyId);
const queuedWakes = db.listWakeupRequests(run.companyId, { agentId: run.agentId, status: "queued" });
if (queuedWakes.length > 0 && !isQuitting() && !destroyed) {
for (const wake of queuedWakes) {
wakeAgentIfPossible(run.agentId, wake.companyId, wake.source);
}
}