mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-07 07:55:47 +01:00
Drive issue recovery from committed changes and exact deadlines (#18402)
Reviewed and verified at 11b49125db.
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
12b0cc6304
commit
42fd568cf8
38 files changed
+1224
-216
No files matched your search
@@ -4168,6 +4168,7 @@ describe("company portability", () => {
|
||||
const insertedRelationValues: Array<Record<string, unknown>> = [];
|
||||
const monitorUpdates: Array<Record<string, unknown>> = [];
|
||||
const db = {
|
||||
transaction: async (write: (tx: unknown) => Promise<unknown>) => write(db),
|
||||
insert: () => ({
|
||||
values: (rows: Array<Record<string, unknown>>) => ({
|
||||
onConflictDoNothing: async () => {
|
||||
|
||||
@@ -0,0 +1,525 @@
|
||||
import { createServer } from "node:http";
|
||||
import { setImmediate as immediate } from "node:timers/promises";
|
||||
import { idleWorkSnapshot } from "../services/task-admission.js";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { agents, authUsers, companies, companyMemberships, createDb, heartbeatRuns, issueRecoveryActions, issueRelations, issues, issueWatchdogs, issueComments } from "@paperclipai/db";
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { heartbeatService } from "../services/heartbeat.js";
|
||||
import { instanceSettingsService } from "../services/instance-settings.js";
|
||||
import { undeliveredConversationComments } from "../services/agent-conversations.js";
|
||||
import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js";
|
||||
import { issueService } from "../services/issues.js";
|
||||
import { issueRecoveryActionService } from "../services/issue-recovery-actions.js";
|
||||
import { registerIssueRecoveryWork } from "../services/recovery/work.js";
|
||||
import { notifyIssueRecovery } from "../services/issue-recovery-notifications.js";
|
||||
import { taskWatchdogService } from "../services/task-watchdogs.js";
|
||||
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
|
||||
|
||||
vi.setConfig({ testTimeout: 90_000 });
|
||||
|
||||
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_ISSUE_RECOVERY_TEST_DATABASE_URL");
|
||||
const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
|
||||
describe.skipIf(!support.supported)("issue recovery production producers and deadlines", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let heartbeat: ReturnType<typeof heartbeatService>;
|
||||
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator>;
|
||||
let worker: ReturnType<ReturnType<typeof heartbeatService>["registerIssueStateRecoveryWork"]>;
|
||||
let dispatch: ReturnType<ReturnType<typeof heartbeatService>["registerRunDispatchWork"]>;
|
||||
let cancellation: ReturnType<ReturnType<typeof heartbeatService>["registerScheduledRetryCancellationWork"]>;
|
||||
let draining = false;
|
||||
const tasks = new Map<string, { nextRunAt?: () => number | null; hasPending: () => Promise<boolean> }>();
|
||||
const errors: unknown[] = [];
|
||||
const passes = new Set<Promise<unknown>>();
|
||||
const extraStops: Array<() => Promise<void>> = [];
|
||||
const extraDeadlines: Array<() => number | null> = [];
|
||||
beforeAll(async () => {
|
||||
if (!external) database = await startEmbeddedPostgresTestDatabase("issue-recovery-work-");
|
||||
db = createDb(external ?? database!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
|
||||
await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`)));
|
||||
}, 90_000);
|
||||
beforeEach(async () => {
|
||||
await db.execute(sql`TRUNCATE companies CASCADE`);
|
||||
heartbeat = heartbeatService(db); draining = false; errors.length = 0;
|
||||
});
|
||||
afterEach(async () => {
|
||||
for (const stop of extraStops.splice(0)) await stop();
|
||||
extraDeadlines.length = 0;
|
||||
await worker?.stop(); await dispatch?.stop(); await cancellation?.stop(); await coordinator?.stop();
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
vi.useRealTimers(); vi.restoreAllMocks();
|
||||
expect(errors).toEqual([]);
|
||||
});
|
||||
afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); });
|
||||
async function start() {
|
||||
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true,
|
||||
onError: error => errors.push(error) });
|
||||
tasks.clear();
|
||||
const register = coordinator.register.bind(coordinator);
|
||||
coordinator.register = (queue, task) => {
|
||||
tasks.set(queue, task);
|
||||
return register(queue, { ...task, run: signal => {
|
||||
const work = task.run(signal); passes.add(work);
|
||||
void work.then(() => passes.delete(work), () => passes.delete(work));
|
||||
return work;
|
||||
} });
|
||||
};
|
||||
worker = heartbeat.registerIssueStateRecoveryWork(coordinator, () => !draining);
|
||||
dispatch = heartbeat.registerRunDispatchWork(coordinator, () => !draining);
|
||||
cancellation = heartbeat.registerScheduledRetryCancellationWork(coordinator, () => !draining);
|
||||
await Promise.all([worker.ready, dispatch.ready, cancellation.ready]);
|
||||
}
|
||||
async function compete() {
|
||||
const other = heartbeatService(db);
|
||||
const competing = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
|
||||
const register = competing.register.bind(competing);
|
||||
competing.register = (queue, task) => register(queue, { ...task, run: signal => {
|
||||
const pass = task.run(signal); passes.add(pass);
|
||||
void pass.then(() => passes.delete(pass), () => passes.delete(pass));
|
||||
return pass;
|
||||
} });
|
||||
const work = other.registerIssueStateRecoveryWork(competing, () => true);
|
||||
extraStops.push(() => competing.stop()); extraDeadlines.push(competing.nextWakeAt); await work.ready;
|
||||
}
|
||||
async function fixture() {
|
||||
const companyId = randomUUID();
|
||||
await db.insert(authUsers).values({ id: "recovery-user", name: "Recovery operator", email: "recovery-user@example.test", createdAt: new Date(), updatedAt: new Date() }).onConflictDoNothing();
|
||||
await db.insert(companies).values({ id: companyId, name: "Recovery", issuePrefix: `R${companyId.slice(0, 7)}`,
|
||||
requireBoardApprovalForNewAgents: false, defaultResponsibleUserId: "recovery-user" });
|
||||
await db.insert(companyMemberships).values({ companyId, principalId: "recovery-user", principalType: "user", status: "active", membershipRole: "owner" });
|
||||
const [agent] = await db.insert(agents).values({ companyId, name: "Recovery process", role: "engineer", status: "idle",
|
||||
adapterType: "process", adapterConfig: { command: process.execPath, args: ["-e", "console.log('recovery-proof')"], cwd: process.cwd() },
|
||||
runtimeConfig: { heartbeat: { enabled: true, intervalSec: 0, wakeOnDemand: true } },
|
||||
}).returning();
|
||||
return { companyId, agentId: agent.id };
|
||||
}
|
||||
async function completingProvider(f: { companyId: string; agentId: string }, beforeComplete?: () => Promise<unknown>) {
|
||||
const failures: unknown[] = [];
|
||||
const listener = createServer(async (request, response) => {
|
||||
try {
|
||||
let body = "";
|
||||
for await (const chunk of request) body += chunk;
|
||||
const { runId } = JSON.parse(body) as { runId: string };
|
||||
const [run] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId),
|
||||
eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.agentId, f.agentId)));
|
||||
if (!run?.contextSnapshot?.issueId) throw new Error("Unknown controlled task run");
|
||||
await beforeComplete?.();
|
||||
await issueService(db).update(String(run.contextSnapshot.issueId), { status: "done" });
|
||||
response.end("ok");
|
||||
} catch (error) { failures.push(error); response.statusCode = 500; response.end("completion failed"); }
|
||||
});
|
||||
await new Promise<void>(resolve => listener.listen(0, "127.0.0.1", resolve));
|
||||
extraStops.push(async () => { listener.closeAllConnections(); await new Promise<void>(resolve => listener.close(() => resolve())); });
|
||||
const port = (listener.address() as { port: number }).port;
|
||||
const script = `fetch("http://127.0.0.1:${port}/complete", {method:"POST", body:JSON.stringify({runId:process.env.PAPERCLIP_RUN_ID})}).then(async response=>{if(!response.ok)throw new Error(await response.text());console.log("recovery-proof")}).catch(error=>{console.error(error);process.exitCode=1})`;
|
||||
await db.update(agents).set({ adapterConfig: { command: process.execPath, args: ["-e", script], cwd: process.cwd() } })
|
||||
.where(eq(agents.id, f.agentId));
|
||||
return failures;
|
||||
}
|
||||
async function settled() {
|
||||
await vi.waitFor(async () => {
|
||||
await Promise.allSettled([...passes]);
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
const wake = coordinator.nextWakeAt();
|
||||
const detail = wake === null ? "" : JSON.stringify({ now: Date.now(), tasks: await Promise.all([...tasks].map(async ([queue, task]) => ({ queue, at: task.nextRunAt?.(), pending: await task.hasPending() }))),
|
||||
runs: (await db.select().from(heartbeatRuns)).map(run => ({ id: run.id, status: run.status, error: run.errorCode, lease: run.controllerLeaseExpiresAt, reason: run.contextSnapshot?.wakeReason })) });
|
||||
expect(wake, detail).toBeNull();
|
||||
expect(extraDeadlines.map(next => next())).toEqual(extraDeadlines.map(() => null));
|
||||
expect(idleWorkSnapshot().active).toBe(0);
|
||||
}, { timeout: 40_000 });
|
||||
}
|
||||
async function quiet() {
|
||||
await settled();
|
||||
expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.status} in ('queued', 'running')`)).toHaveLength(0);
|
||||
const reads = vi.spyOn(db, "select"); const executes = vi.spyOn(db, "execute");
|
||||
const selected = reads.mock.calls.length, executed = executes.mock.calls.length;
|
||||
const statements: string[] = [], priorDebug = db.$client.options.debug;
|
||||
db.$client.options.debug = (_connection, query) => { statements.push(query); };
|
||||
if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
|
||||
try {
|
||||
await vi.advanceTimersByTimeAsync(20 * 60_000);
|
||||
expect(reads).toHaveBeenCalledTimes(selected); expect(executes).toHaveBeenCalledTimes(executed);
|
||||
expect(statements).toEqual([]);
|
||||
} finally { vi.useRealTimers(); db.$client.options.debug = priorDebug; reads.mockRestore(); executes.mockRestore(); }
|
||||
}
|
||||
async function finish(issueId: string) {
|
||||
const retries = () => db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.status, "scheduled_retry"),
|
||||
sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`));
|
||||
const before = await retries();
|
||||
await issueService(db).update(issueId, { status: "done" });
|
||||
// Closing one blocker can legitimately start its dependent. Wait for this
|
||||
// task's dispatch and execution; quiet() verifies the whole fixture later.
|
||||
await vi.waitFor(async () => {
|
||||
await Promise.allSettled([...passes]);
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
expect(await db.select().from(heartbeatRuns).where(and(
|
||||
sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`,
|
||||
sql`${heartbeatRuns.status} in ('queued', 'running')`,
|
||||
))).toHaveLength(0);
|
||||
}, { timeout: 15_000 });
|
||||
const saved = [...new Map([...before, ...await retries()].map(run => [run.id, run])).values()];
|
||||
if (saved.length > 0) {
|
||||
const trace: unknown[] = [];
|
||||
await vi.waitFor(async () => {
|
||||
await Promise.allSettled([...passes]);
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
const remaining = await retries();
|
||||
if (remaining.length > 0) trace.push({ now: Date.now(),
|
||||
deadlines: [coordinator.nextWakeAt(), ...extraDeadlines.map(next => next())],
|
||||
remaining: remaining.map(run => ({ id: run.id, at: run.scheduledRetryAt })) });
|
||||
expect(remaining, JSON.stringify(trace)).toHaveLength(0);
|
||||
}, { timeout: Math.max(...saved.map(run => run.scheduledRetryAt!.getTime())) - Date.now() + 30_000 });
|
||||
for (const retry of saved) {
|
||||
const [settledRetry] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, retry.id));
|
||||
expect(settledRetry.status).toBe("cancelled");
|
||||
}
|
||||
}
|
||||
}
|
||||
async function completed(issueId: string) {
|
||||
let runId: string | undefined;
|
||||
await vi.waitFor(async () => {
|
||||
const runs = await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`);
|
||||
const run = runs.find(row => row.status === "running" || row.status === "succeeded");
|
||||
expect(run).toBeDefined(); runId = run!.id;
|
||||
}, { timeout: 15_000 });
|
||||
await vi.waitFor(async () => {
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
const [current] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId!));
|
||||
expect(current.status, JSON.stringify({ error: current.error, code: current.errorCode, stderr: current.stderrExcerpt })).toBe("succeeded");
|
||||
}, { timeout: 15_000 });
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId!));
|
||||
expect(run.stdoutExcerpt).toContain("recovery-proof");
|
||||
return run;
|
||||
}
|
||||
async function savedConversation(withMessage = true) {
|
||||
const f = await fixture(), settings = instanceSettingsService(db);
|
||||
await settings.updateExperimental({ enableAgentChat: true, enableConferenceRoomChat: false });
|
||||
const replies: string[] = [], failures: unknown[] = [];
|
||||
const listener = createServer(async (request, response) => {
|
||||
try {
|
||||
let body = "";
|
||||
for await (const chunk of request) body += chunk;
|
||||
const { runId } = JSON.parse(body) as { runId: string };
|
||||
const [run] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId),
|
||||
eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.agentId, f.agentId)));
|
||||
if (!run) throw new Error("Unknown controlled conversation run");
|
||||
await issueService(db).addComment(String(run.contextSnapshot?.issueId), "Saved chat delivered", {
|
||||
agentId: f.agentId, runId: run.id,
|
||||
});
|
||||
replies.push(run.id); response.end("ok");
|
||||
} catch (error) { failures.push(error); response.statusCode = 500; response.end("controlled reply failed"); }
|
||||
});
|
||||
await new Promise<void>(resolve => listener.listen(0, "127.0.0.1", resolve));
|
||||
extraStops.push(async () => { listener.closeAllConnections(); await new Promise<void>(resolve => listener.close(() => resolve())); });
|
||||
const port = (listener.address() as { port: number }).port;
|
||||
const script = `fetch("http://127.0.0.1:${port}/reply", {method:"POST", body:JSON.stringify({runId:process.env.PAPERCLIP_RUN_ID})}).then(async response=>{if(!response.ok)throw new Error(await response.text());console.log("recovery-proof")}).catch(error=>{console.error(error);process.exitCode=1})`;
|
||||
await db.update(agents).set({ adapterConfig: { command: process.execPath, args: ["-e", script], cwd: process.cwd() } })
|
||||
.where(eq(agents.id, f.agentId));
|
||||
const chat = await issueService(db).create(f.companyId, { title: "Saved conversation", status: "in_review",
|
||||
conversationAgentId: f.agentId, conversationUserId: "recovery-user", assigneeAgentId: f.agentId,
|
||||
conversationState: "waiting" });
|
||||
const message = withMessage ? await issueService(db).addComment(chat.id, "Keep this message", { userId: "recovery-user" }, { clientRequestId: randomUUID() }) : null;
|
||||
await settings.updateExperimental({ enableAgentChat: false });
|
||||
await start(); await quiet();
|
||||
expect(replies).toEqual([]);
|
||||
return { ...f, settings, chat, message, replies, failures };
|
||||
}
|
||||
it("dispatches a saved conversation only from the committed Agent Chat enable transition", async () => {
|
||||
const f = await savedConversation();
|
||||
expect(await undeliveredConversationComments(db, f.companyId, f.chat.id)).toHaveLength(1);
|
||||
await f.settings.updateExperimental({ enableAgentChat: true });
|
||||
expect(coordinator.nextWakeAt()).not.toBeNull();
|
||||
await vi.waitFor(async () => expect(await db.select().from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(1), { timeout: 40_000 });
|
||||
const run = await completed(f.chat.id);
|
||||
expect(f.replies).toEqual([run.id]); expect(f.failures).toEqual([]);
|
||||
expect(await undeliveredConversationComments(db, f.companyId, f.chat.id)).toEqual([]);
|
||||
expect((await db.select().from(issues).where(eq(issues.id, f.chat.id)))[0])
|
||||
.toMatchObject({ conversationState: "waiting", executionRunId: null });
|
||||
expect(await db.select().from(issueComments).where(and(eq(issueComments.createdByRunId, run.id),
|
||||
eq(issueComments.authorAgentId, f.agentId)))).toHaveLength(1);
|
||||
await quiet();
|
||||
let hints = 0;
|
||||
const stop = subscribeDeliveryWork(db, DELIVERY_QUEUES.strandedAssignments, () => { hints++; });
|
||||
try { await f.settings.updateExperimental({ enableAgentChat: true }); expect(hints).toBe(0); await quiet(); }
|
||||
finally { stop(); }
|
||||
expect(f.replies).toEqual([run.id]);
|
||||
});
|
||||
it("keeps disabled saved conversations parked after rollback, unrelated flags and unchanged values", async () => {
|
||||
const f = await savedConversation();
|
||||
await expect(db.transaction(async tx => {
|
||||
await instanceSettingsService(tx as unknown as typeof db).updateExperimental({ enableAgentChat: true });
|
||||
throw new Error("chat flag rollback");
|
||||
})).rejects.toThrow("chat flag rollback");
|
||||
await quiet();
|
||||
expect((await f.settings.getExperimental()).enableAgentChat).toBe(false);
|
||||
let hints = 0;
|
||||
const stop = subscribeDeliveryWork(db, DELIVERY_QUEUES.strandedAssignments, () => { hints++; });
|
||||
try {
|
||||
await f.settings.updateExperimental({ enableConferenceRoomChat: true });
|
||||
await f.settings.updateExperimental({ enableAgentChat: false });
|
||||
expect(hints).toBe(0); await quiet();
|
||||
} finally { stop(); }
|
||||
expect(f.replies).toEqual([]); expect(f.failures).toEqual([]);
|
||||
expect(await db.select().from(heartbeatRuns)).toEqual([]);
|
||||
expect(await undeliveredConversationComments(db, f.companyId, f.chat.id)).toHaveLength(1);
|
||||
});
|
||||
it("does not dispatch an accepted saved message after Agent Chat is disabled", async () => {
|
||||
const f = await savedConversation(false);
|
||||
await f.settings.updateExperimental({ enableAgentChat: true }); await quiet();
|
||||
draining = true;
|
||||
await issueService(db).addComment(f.chat.id, "Retain while disabled", { userId: "recovery-user" }, { clientRequestId: randomUUID() });
|
||||
let hints = 0;
|
||||
const stop = subscribeDeliveryWork(db, DELIVERY_QUEUES.strandedAssignments, () => { hints++; });
|
||||
try { await f.settings.updateExperimental({ enableAgentChat: false }); expect(hints).toBe(0); }
|
||||
finally { stop(); }
|
||||
draining = false; await quiet();
|
||||
expect(f.replies).toEqual([]); expect(f.failures).toEqual([]);
|
||||
expect(await db.select().from(heartbeatRuns)).toEqual([]);
|
||||
expect(await undeliveredConversationComments(db, f.companyId, f.chat.id)).toHaveLength(1);
|
||||
});
|
||||
it("keeps empty, configured idle, and terminal history quiet", async () => {
|
||||
await fixture(); await start(); await quiet();
|
||||
});
|
||||
it("creates a blocking child through the ordinary root service transaction", async () => {
|
||||
const f = await fixture();
|
||||
const parent = await issueService(db).create(f.companyId, { title: "Parent", status: "todo" });
|
||||
const child = await issueService(db).createChild(parent.id, { title: "Blocking child", status: "todo", blockParentUntilDone: true });
|
||||
expect(child.parentBlockerAdded).toBe(true);
|
||||
expect(await db.select().from(issueRelations).where(and(eq(issueRelations.companyId, f.companyId),
|
||||
eq(issueRelations.issueId, child.issue.id), eq(issueRelations.relatedIssueId, parent.id)))).toHaveLength(1);
|
||||
});
|
||||
it.each(["startup", "committed"])("restores blocker-only and completed-action debt with no assignment candidates (%s)", async mode => {
|
||||
const f = await fixture();
|
||||
if (mode === "committed") { await start(); await quiet(); }
|
||||
const blocker = await issueService(db).create(f.companyId, { title: "Orphan blocker", status: "blocked", createdByAgentId: f.agentId });
|
||||
const dependent = await issueService(db).create(f.companyId, { title: "Blocked dependent", status: "blocked", blockedByIssueIds: [blocker.id] });
|
||||
const completedSource = await issueService(db).create(f.companyId, { title: "Completed source", status: "done" });
|
||||
const action = await issueRecoveryActionService(db).upsertSourceScoped({ companyId: f.companyId, sourceIssueId: completedSource.id,
|
||||
kind: "stranded_assigned_issue", ownerType: "agent", ownerAgentId: f.agentId, cause: "stranded_assigned_issue",
|
||||
fingerprint: "source-terminal", nextAction: "Restore work", wakePolicy: { type: "wake_owner" }, evidence: {} });
|
||||
if (mode === "startup") await start();
|
||||
await vi.waitFor(async () => {
|
||||
expect((await db.select().from(issues).where(eq(issues.id, blocker.id)))[0].assigneeAgentId).toBe(f.agentId);
|
||||
expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, action.id)))[0].status).toBe("resolved");
|
||||
}, { timeout: 40_000 });
|
||||
await finish(blocker.id); await finish(dependent.id); await quiet();
|
||||
});
|
||||
it("bounds subsidiary scans for100 assignment candidates and coalesces repeated company writes", async () => {
|
||||
const f = await fixture(), foreign = await fixture();
|
||||
await db.update(agents).set({ lifecycleState: "preparing" }).where(eq(agents.id, f.agentId));
|
||||
await db.insert(issues).values(Array.from({ length: 100 }, (_, index) => ({ companyId: f.companyId,
|
||||
title: `Waiting setup ${index}`, status: "todo", assigneeAgentId: f.agentId })));
|
||||
const [source] = await db.insert(issues).values({ companyId: f.companyId, title: "Done source", status: "done" }).returning();
|
||||
const action = await issueRecoveryActionService(db).upsertSourceScoped({ companyId: f.companyId, sourceIssueId: source.id,
|
||||
kind: "stranded_assigned_issue", ownerType: "agent", ownerAgentId: f.agentId, cause: "stranded_assigned_issue",
|
||||
fingerprint: "query-budget", nextAction: "Restore work", wakePolicy: { type: "wake_owner" }, evidence: {} });
|
||||
const scans = { blockers: 0, actions: 0, loads: 0 };
|
||||
const originalSelect = db.select.bind(db);
|
||||
vi.spyOn(db, "select").mockImplementation(((fields?: Record<string, unknown>) => {
|
||||
const query = originalSelect(fields as never); const from = query.from.bind(query);
|
||||
query.from = ((table: unknown) => {
|
||||
if (table === issues && fields?.id === issues.id && fields.companyId === issues.companyId && Object.keys(fields).length === 2) scans.loads++;
|
||||
if (table === issueRelations && fields?.createdByAgentId === issues.createdByAgentId) scans.blockers++;
|
||||
if (table === issueRecoveryActions && fields?.action === issueRecoveryActions) scans.actions++;
|
||||
return from(table as never);
|
||||
}) as typeof query.from;
|
||||
return query;
|
||||
}) as typeof db.select);
|
||||
await start(); await settled();
|
||||
expect(scans.actions).toBe(1); expect(scans.blockers).toBe(0); expect(scans.loads).toBeLessThanOrEqual(8);
|
||||
expect((await originalSelect().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, action.id)))[0].status).toBe("resolved");
|
||||
const [foreignSource] = await db.insert(issues).values({ companyId: foreign.companyId, title: "Foreign done", status: "done" }).returning();
|
||||
const [foreignAction] = await db.insert(issueRecoveryActions).values({ companyId: foreign.companyId, sourceIssueId: foreignSource.id,
|
||||
kind: "stranded_assigned_issue", ownerType: "board", cause: "stranded_assigned_issue", fingerprint: "foreign-query-budget", nextAction: "Inspect" }).returning();
|
||||
const before = { ...scans };
|
||||
for (let index = 0; index < 100; index++) await db.transaction(tx => notifyIssueRecovery(tx, { companyId: f.companyId }));
|
||||
await new Promise(resolve => setTimeout(resolve, 100));
|
||||
expect(scans.loads).toBe(before.loads);
|
||||
await settled();
|
||||
expect(scans.actions).toBe(before.actions); expect(scans.blockers).toBe(before.blockers);
|
||||
expect(scans.loads).toBeLessThanOrEqual(before.loads + 4);
|
||||
expect((await originalSelect().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, foreignAction.id)))[0].status).toBe("active");
|
||||
expect(await originalSelect().from(heartbeatRuns)).toHaveLength(0); await quiet();
|
||||
});
|
||||
it("limits an ordinary comment to its issue among one thousand open assignments", async () => {
|
||||
const f = await fixture();
|
||||
const rows = await db.insert(issues).values(Array.from({ length: 1000 }, (_, index) => ({
|
||||
companyId: f.companyId, title: `Open work ${index}`, status: "todo", assigneeAgentId: f.agentId,
|
||||
}))).returning({ id: issues.id });
|
||||
const checked: string[] = [];
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
|
||||
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) });
|
||||
const register = coordinator.register.bind(coordinator);
|
||||
coordinator.register = (queue, task) => register(queue, { ...task, run: signal => {
|
||||
const pass = task.run(signal); passes.add(pass);
|
||||
void pass.then(() => passes.delete(pass), () => passes.delete(pass));
|
||||
return pass;
|
||||
} });
|
||||
// Observe the actual paged reconstruction boundary without executing one
|
||||
// thousand adapters. The producer, transaction, coordinator and SQL are real;
|
||||
// the other acceptance cases exercise the resulting recovery decisions.
|
||||
worker = registerIssueRecoveryWork(db, coordinator, {
|
||||
stranded: async scope => { checked.push(scope.issueId); },
|
||||
orphanBlockers: async () => {}, recoveryActions: async () => {},
|
||||
dependencies: async () => ({ enqueueFailed: 0 }), locks: async () => {},
|
||||
watchdogs: async () => ({ incomplete: 0 }), schedulingSuppressed: async () => false,
|
||||
}, () => true);
|
||||
await worker.ready;
|
||||
async function settleWork() {
|
||||
for (let step = 0; step < 120; step++) {
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
await Promise.allSettled([...passes]); await immediate();
|
||||
if (coordinator.nextWakeAt() === null && idleWorkSnapshot().active === 0) return;
|
||||
}
|
||||
throw new Error("Recovery did not settle its committed work within two virtual minutes");
|
||||
}
|
||||
await settleWork();
|
||||
expect(new Set(checked).size).toBe(1000);
|
||||
checked.length = 0;
|
||||
await issueService(db).addComment(rows[0].id, "New evidence on this task", { userId: "recovery-user" });
|
||||
await settleWork();
|
||||
expect(checked).toEqual([rows[0].id]);
|
||||
// Company eligibility changes still reconstruct all assignments. Narrowing
|
||||
// comment hints must not hide unrelated work after an operator resumes it.
|
||||
checked.length = 0;
|
||||
await db.transaction(tx => notifyIssueRecovery(tx, { companyId: f.companyId }));
|
||||
await settleWork();
|
||||
expect(new Set(checked).size).toBe(1000);
|
||||
await quiet();
|
||||
});
|
||||
it("dispatches an ordinary new assignment after commit and returns to quiet after completion", async () => {
|
||||
const f = await fixture(); await start(); await quiet();
|
||||
const issue = await issueService(db).create(f.companyId, { title: "Recover assignment", status: "todo", assigneeAgentId: f.agentId });
|
||||
await completed(issue.id);
|
||||
await finish(issue.id); await quiet();
|
||||
}, 90_000);
|
||||
it("reacts to blocker resolution once and dispatches the durable dependency wake", async () => {
|
||||
const f = await fixture();
|
||||
const blocker = await issueService(db).create(f.companyId, { title: "Blocker", status: "todo" });
|
||||
const dependent = await issueService(db).create(f.companyId, { title: "Dependent", status: "blocked", assigneeAgentId: f.agentId,
|
||||
blockedByIssueIds: [blocker.id] });
|
||||
const failures = await completingProvider(f);
|
||||
await start(); await quiet();
|
||||
await compete();
|
||||
await issueService(db).update(blocker.id, { status: "done" });
|
||||
const run = await completed(dependent.id);
|
||||
expect(run.contextSnapshot?.wakeReason).toBe("issue_blockers_resolved");
|
||||
expect(failures).toEqual([]);
|
||||
await finish(dependent.id); await quiet();
|
||||
expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId),
|
||||
sql`${heartbeatRuns.contextSnapshot}->>'wakeReason' = 'issue_blockers_resolved'`))).toHaveLength(1);
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(1);
|
||||
}, 90_000);
|
||||
it("settles saved disposition debt on its original real-clock timers before measuring idle", async () => {
|
||||
const f = await fixture();
|
||||
const blocker = await issueService(db).create(f.companyId, { title: "Saved debt blocker", status: "todo" });
|
||||
const issue = await issueService(db).create(f.companyId, { title: "Saved disposition debt", status: "blocked",
|
||||
assigneeAgentId: f.agentId, blockedByIssueIds: [blocker.id] });
|
||||
await start(); await compete();
|
||||
await issueService(db).update(blocker.id, { status: "done" });
|
||||
await completed(issue.id);
|
||||
let retryId: string | undefined;
|
||||
await vi.waitFor(async () => {
|
||||
const [retry] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.status, "scheduled_retry"),
|
||||
sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issue.id}`));
|
||||
expect(retry).toBeDefined(); retryId = retry!.id;
|
||||
}, { timeout: 40_000 });
|
||||
await finish(issue.id); await quiet();
|
||||
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, retryId!)))[0].status).toBe("cancelled");
|
||||
}, 120_000);
|
||||
it("uses ordinary watchdog configuration and subtree closure to dispatch a reusable review", async () => {
|
||||
const f = await fixture();
|
||||
const source = await issueService(db).create(f.companyId, { title: "Finished subtree", status: "done" });
|
||||
const failures = await completingProvider(f, () => taskWatchdogService(db).disableForIssue(f.companyId, source.id));
|
||||
await start(); await quiet();
|
||||
await compete();
|
||||
const watchdog = await taskWatchdogService(db).upsertForIssue(f.companyId, source.id, { agentId: f.agentId });
|
||||
let reviewId: string | undefined;
|
||||
await vi.waitFor(async () => {
|
||||
const [row] = await db.select().from(issueWatchdogs).where(eq(issueWatchdogs.id, watchdog.watchdog.id));
|
||||
expect(row.watchdogIssueId).toBeTruthy(); reviewId = row.watchdogIssueId!;
|
||||
}, { timeout: 15_000 });
|
||||
await completed(reviewId!);
|
||||
expect(failures).toEqual([]);
|
||||
expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId),
|
||||
sql`${heartbeatRuns.contextSnapshot}->>'wakeReason' = 'task_watchdog_stopped_subtree'`))).toHaveLength(1);
|
||||
await finish(reviewId!);
|
||||
await taskWatchdogService(db).disableForIssue(f.companyId, source.id);
|
||||
if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); await quiet();
|
||||
}, 90_000);
|
||||
it("bounds ordinary writes to their company and reconstructs foreign stale debt after restart", async () => {
|
||||
const selected = await fixture(); await start(); await quiet();
|
||||
const foreign = await fixture();
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId: foreign.companyId, agentId: foreign.agentId,
|
||||
status: "failed", finishedAt: new Date() }).returning();
|
||||
const [stale] = await db.insert(issues).values({ companyId: foreign.companyId, title: "Foreign stale lock", status: "done",
|
||||
checkoutRunId: run.id }).returning();
|
||||
await issueService(db).create(selected.companyId, { title: "Selected change", status: "done" }); await quiet();
|
||||
expect((await db.select().from(issues).where(eq(issues.id, stale.id)))[0].checkoutRunId).toBe(run.id);
|
||||
await worker.stop(); await dispatch.stop(); await cancellation.stop(); await coordinator.stop();
|
||||
heartbeat = heartbeatService(db); await start(); await quiet();
|
||||
expect((await db.select().from(issues).where(eq(issues.id, stale.id)))[0].checkoutRunId).toBeNull();
|
||||
});
|
||||
it("fences nested assignment writes until outer commit and ignores rollback", async () => {
|
||||
const f = await fixture(); await start(); await quiet();
|
||||
let committedId = "";
|
||||
await db.transaction(async outer => {
|
||||
await outer.transaction(async nested => {
|
||||
const issue = await issueService(db).create(f.companyId, { title: "Nested commit", status: "todo", assigneeAgentId: f.agentId }, nested);
|
||||
committedId = issue.id;
|
||||
});
|
||||
await new Promise(resolve => setTimeout(resolve, 100));
|
||||
expect(await db.select().from(heartbeatRuns)).toHaveLength(0);
|
||||
});
|
||||
await completed(committedId); await finish(committedId); await settled();
|
||||
await expect(db.transaction(async outer => {
|
||||
await outer.transaction(async nested => {
|
||||
await issueService(db).create(f.companyId, { title: "Nested rollback", status: "todo", assigneeAgentId: f.agentId }, nested);
|
||||
});
|
||||
throw new Error("rollback control");
|
||||
})).rejects.toThrow("rollback control");
|
||||
await quiet();
|
||||
expect(await db.select().from(issues).where(eq(issues.title, "Nested rollback"))).toHaveLength(0);
|
||||
}, 90_000);
|
||||
it("retains the first committed assignment during drain and executes on resume", async () => {
|
||||
const f = await fixture(); await start(); await quiet(); draining = true;
|
||||
const issue = await issueService(db).create(f.companyId, { title: "Drained assignment", status: "todo", assigneeAgentId: f.agentId });
|
||||
await new Promise(resolve => setTimeout(resolve, 100));
|
||||
const reads = vi.spyOn(db, "select");
|
||||
if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
|
||||
try { await vi.advanceTimersByTimeAsync(90_000); expect(reads).not.toHaveBeenCalled(); }
|
||||
finally { vi.useRealTimers(); reads.mockRestore(); }
|
||||
expect(await db.select().from(heartbeatRuns)).toHaveLength(0);
|
||||
draining = false; worker.wake(); dispatch.wake();
|
||||
await completed(issue.id); await finish(issue.id); await quiet();
|
||||
}, 90_000);
|
||||
it("retains a fresh subtree's exact first-run grace then disarms on disable", async () => {
|
||||
const f = await fixture();
|
||||
const source = await issueService(db).create(f.companyId, { title: "Fresh unassigned subtree", status: "todo" });
|
||||
await start(); await quiet();
|
||||
await taskWatchdogService(db).upsertForIssue(f.companyId, source.id, { agentId: f.agentId });
|
||||
await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(source.createdAt.getTime() + 15_000), { timeout: 5_000 });
|
||||
expect(await db.select().from(issues).where(eq(issues.originKind, "task_watchdog"))).toHaveLength(0);
|
||||
await taskWatchdogService(db).disableForIssue(f.companyId, source.id);
|
||||
if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); await quiet();
|
||||
});
|
||||
it("waits for a live foreign controller's exact stored lease before clearing a terminal lock", async () => {
|
||||
const f = await fixture();
|
||||
const source = await issueService(db).create(f.companyId, { title: "Remote owner", status: "done" });
|
||||
const lease = new Date(Date.now() + 5_000);
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running",
|
||||
runtimeMode: "legacy", controllerBootId: randomUUID(), controllerLeaseExpiresAt: lease,
|
||||
contextSnapshot: { issueId: source.id }, processPid: 2_000_000_000, startedAt: new Date() }).returning();
|
||||
await db.update(issues).set({ executionRunId: run.id }).where(eq(issues.id, source.id));
|
||||
await start();
|
||||
await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(lease.getTime()), { timeout: 3_000 });
|
||||
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].status).toBe("running");
|
||||
await vi.waitFor(async () => expect((await db.select().from(issues).where(eq(issues.id, source.id)))[0].executionRunId).toBeNull(), { timeout: 12_000 });
|
||||
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0].status).toBe("succeeded");
|
||||
if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); await quiet();
|
||||
}, 90_000);
|
||||
});
|
||||
@@ -0,0 +1,149 @@
|
||||
import { PgDialect } from "drizzle-orm/pg-core";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import { issueRecoveryActions, issueWatchdogs } from "@paperclipai/db";
|
||||
import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js";
|
||||
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
|
||||
import { notifyIssueRecovery } from "../services/issue-recovery-notifications.js";
|
||||
import { registerIssueRecoveryWork } from "../services/recovery/work.js";
|
||||
import { idleWorkSnapshot } from "../services/task-admission.js";
|
||||
|
||||
const stops: Array<() => Promise<void>> = [];
|
||||
afterEach(async () => { for (const stop of stops.splice(0)) await stop(); vi.useRealTimers(); });
|
||||
function fixture() {
|
||||
vi.useFakeTimers(); vi.setSystemTime(1_000_000);
|
||||
const dialect = new PgDialect();
|
||||
const rows = { issues: [] as Array<{ id: string; companyId: string }>, watchdogs: [] as Array<{ id: string; companyId: string }>, blockers: [] as Array<{ id: string; companyId: string }>, actions: [] as Array<{ id: string; companyId: string }> };
|
||||
const execute = vi.fn(async (query: any) => {
|
||||
if (dialect.sqlToQuery(query).sql.includes("pg_current_xact_id")) return [{ xid: "42" }];
|
||||
throw new Error("unexpected SQL");
|
||||
});
|
||||
const load = vi.fn((table: unknown, distinct = false) => distinct
|
||||
? table === issueRecoveryActions ? rows.actions : rows.blockers
|
||||
: table === issueWatchdogs ? rows.watchdogs : rows.issues);
|
||||
const query = (distinct = false) => ({ from: (table: unknown) => ({ where: () => ({ orderBy: () => ({ limit: async () => load(table, distinct) }) }) }) });
|
||||
const db: any = installDatabaseWorkSignals({ execute,
|
||||
select: () => query(), selectDistinct: () => query(true),
|
||||
transaction: async (callback: (tx: any) => Promise<unknown>) => callback({ execute, select: db.select, selectDistinct: db.selectDistinct, transaction: db.transaction }),
|
||||
});
|
||||
const state = { draining: false, standby: false, suppressed: false, deadline: null as number | null, fail: false, incomplete: false };
|
||||
const check = vi.fn(async (scope: { onDeferredDeadline(at: number): void }) => {
|
||||
if (state.fail) throw new Error("database unavailable");
|
||||
if (state.deadline !== null && state.deadline > Date.now()) scope.onDeferredDeadline(state.deadline);
|
||||
});
|
||||
const orphanBlockers = vi.fn(async () => {}), recoveryActions = vi.fn(async () => {});
|
||||
const suppression = vi.fn(async () => state.suppressed);
|
||||
const owner = { orphanBlockers, recoveryActions, stranded: check, dependencies: async (scope: any) => { await check(scope); return { enqueueFailed: 0 }; },
|
||||
locks: check, watchdogs: async (scope: any) => { await check(scope); return { incomplete: state.incomplete ? 1 : 0 }; },
|
||||
schedulingSuppressed: suppression };
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !state.standby,
|
||||
canReconcile: () => !state.standby, onError: vi.fn() });
|
||||
const worker = registerIssueRecoveryWork(db, coordinator, owner, () => !state.draining && !state.standby);
|
||||
stops.push(() => coordinator.stop());
|
||||
return { db, rows, state, check, load, coordinator, worker, orphanBlockers, recoveryActions, suppression,
|
||||
notify: () => db.transaction((tx: any) => notifyIssueRecovery(tx, { companyId: "selected" })),
|
||||
settle: () => vi.advanceTimersByTimeAsync(1_000) };
|
||||
}
|
||||
describe("issue recovery obligation lifecycle", () => {
|
||||
it("has no timer, SQL, or owner probe after empty reconstruction", async () => {
|
||||
const f = fixture(); await f.worker.ready;
|
||||
expect(f.coordinator.nextWakeAt()).toBeNull(); const loaded = f.load.mock.calls.length;
|
||||
await vi.advanceTimersByTimeAsync(20 * 60_000);
|
||||
expect(f.load).toHaveBeenCalledTimes(loaded); expect(f.check).not.toHaveBeenCalled();
|
||||
});
|
||||
it("reads scheduling suppression once per pass instead of once per issue", async () => {
|
||||
const f = fixture();
|
||||
f.rows.issues.push(...Array.from({ length: 25 }, (_, index) => ({ id: `issue-${index}`, companyId: "selected" })));
|
||||
await f.worker.ready;
|
||||
expect(f.check).toHaveBeenCalledTimes(75);
|
||||
expect(f.suppression).toHaveBeenCalledTimes(6);
|
||||
expect(f.coordinator.nextWakeAt()).toBeNull();
|
||||
});
|
||||
it("backs off a failed suppression read without probing an empty or failed owner repeatedly", async () => {
|
||||
const f = fixture(); f.suppression.mockRejectedValueOnce(new Error("settings unavailable"));
|
||||
await f.worker.ready; const reads = f.suppression.mock.calls.length;
|
||||
expect(f.coordinator.nextWakeAt()).toBe(1_030_000);
|
||||
await vi.advanceTimersByTimeAsync(29_999); expect(f.suppression).toHaveBeenCalledTimes(reads);
|
||||
await vi.advanceTimersByTimeAsync(1); expect(f.suppression).toHaveBeenCalledTimes(reads + 1);
|
||||
expect(f.coordinator.nextWakeAt()).toBeNull();
|
||||
});
|
||||
it("normalizes a wide scope with cyclic SQL objects before publishing work hints", async () => {
|
||||
const f = fixture(); await f.worker.ready;
|
||||
const wide = { companyId: "selected", endpointId: issueRecoveryActions.id, table: issueRecoveryActions };
|
||||
await f.db.transaction((tx: object) => notifyIssueRecovery(tx, wide)); await f.settle();
|
||||
expect(f.coordinator.nextWakeAt()).toBeNull();
|
||||
const loads = f.load.mock.calls.length;
|
||||
await vi.advanceTimersByTimeAsync(20 * 60_000); expect(f.load).toHaveBeenCalledTimes(loads);
|
||||
});
|
||||
it("runs orphan blocker and action duties without assignment entries", async () => {
|
||||
const f = fixture(); f.rows.blockers.push({ id: "selected", companyId: "selected" });
|
||||
f.rows.actions.push({ id: "selected", companyId: "selected" }); await f.worker.ready;
|
||||
expect(f.check).not.toHaveBeenCalled();
|
||||
expect(f.orphanBlockers).toHaveBeenCalledExactlyOnceWith({ companyId: "selected" });
|
||||
expect(f.recoveryActions).toHaveBeenCalledExactlyOnceWith({ companyId: "selected" });
|
||||
expect(f.coordinator.nextWakeAt()).toBeNull();
|
||||
});
|
||||
it("coalesces repeated company reconstruction without delaying an existing lease deadline", async () => {
|
||||
const f = fixture(); f.rows.issues.push({ id: "issue", companyId: "selected" });
|
||||
f.state.deadline = 1_015_000; await f.worker.ready;
|
||||
await f.notify(); await f.settle(); const loaded = f.load.mock.calls.length, checked = f.check.mock.calls.length;
|
||||
for (let index = 0; index < 100; index++) { await f.notify(); await vi.advanceTimersByTimeAsync(10); }
|
||||
expect(f.load).toHaveBeenCalledTimes(loaded); expect(f.check).toHaveBeenCalledTimes(checked);
|
||||
await vi.advanceTimersByTimeAsync(13_000);
|
||||
expect(f.state.deadline).toBe(1_015_000);
|
||||
expect(f.coordinator.nextWakeAt()).toBe(1_031_000);
|
||||
await vi.advanceTimersByTimeAsync(16_000);
|
||||
expect(f.load.mock.calls.length).toBeGreaterThan(loaded);
|
||||
expect(f.coordinator.nextWakeAt()).toBeNull();
|
||||
});
|
||||
it("retains exact grace and lease deadlines without early owner probes", async () => {
|
||||
const f = fixture(); f.rows.issues.push({ id: "issue", companyId: "selected" });
|
||||
f.rows.watchdogs.push({ id: "watchdog", companyId: "selected" }); f.state.deadline = 1_015_000;
|
||||
await f.worker.ready; expect(f.coordinator.nextWakeAt()).toBe(1_015_000);
|
||||
const checked = f.check.mock.calls.length;
|
||||
await vi.advanceTimersByTimeAsync(14_999); expect(f.check).toHaveBeenCalledTimes(checked);
|
||||
await vi.advanceTimersByTimeAsync(1); expect(f.check.mock.calls.length).toBe(checked * 2);
|
||||
expect(f.coordinator.nextWakeAt()).toBeNull();
|
||||
});
|
||||
it("releases settled producer holds during domain drain while coordinator remains open", async () => {
|
||||
const f = fixture(); await f.worker.ready; f.state.draining = true;
|
||||
f.rows.issues.push({ id: "issue", companyId: "selected" });
|
||||
await f.notify(); await f.settle(); const loaded = f.load.mock.calls.length;
|
||||
await vi.advanceTimersByTimeAsync(90_000);
|
||||
expect(f.load).toHaveBeenCalledTimes(loaded); expect(f.check).not.toHaveBeenCalled();
|
||||
expect(idleWorkSnapshot().active).toBe(0);
|
||||
f.state.draining = false; f.worker.wake(); await f.settle();
|
||||
expect(f.check).toHaveBeenCalled(); expect(f.coordinator.nextWakeAt()).toBeNull();
|
||||
});
|
||||
it("retries concrete failed entries and then disarms without reloading the company", async () => {
|
||||
const f = fixture(); f.rows.issues.push({ id: "issue", companyId: "selected" }); f.state.fail = true;
|
||||
await f.worker.ready; const loaded = f.load.mock.calls.length;
|
||||
expect(f.coordinator.nextWakeAt()).toBe(1_030_000);
|
||||
await vi.advanceTimersByTimeAsync(29_999); const checked = f.check.mock.calls.length;
|
||||
f.state.fail = false; await vi.advanceTimersByTimeAsync(1);
|
||||
expect(f.check.mock.calls.length).toBe(checked * 2); expect(f.load).toHaveBeenCalledTimes(loaded);
|
||||
expect(f.coordinator.nextWakeAt()).toBeNull();
|
||||
});
|
||||
it("keeps an incomplete watchdog subtree as concrete retry debt", async () => {
|
||||
const f = fixture(); f.rows.watchdogs.push({ id: "watchdog", companyId: "selected" }); f.state.incomplete = true;
|
||||
await f.worker.ready; expect(f.coordinator.nextWakeAt()).toBe(1_030_000);
|
||||
const loaded = f.load.mock.calls.length; f.state.incomplete = false;
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
expect(f.check).toHaveBeenCalledTimes(2); expect(f.load).toHaveBeenCalledTimes(loaded);
|
||||
expect(f.coordinator.nextWakeAt()).toBeNull();
|
||||
});
|
||||
it("keeps a concurrent producer dirty until the in-flight owner has completed", async () => {
|
||||
const f = fixture(); f.rows.issues.push({ id: "issue", companyId: "selected" }); await f.worker.ready;
|
||||
let release!: () => void; const held = new Promise<void>(resolve => { release = resolve; });
|
||||
f.check.mockImplementationOnce(() => held);
|
||||
await f.notify(); await f.settle(); await f.notify(); release(); await f.settle(); await vi.advanceTimersByTimeAsync(30_000);
|
||||
expect(f.coordinator.nextWakeAt()).toBeNull();
|
||||
expect(f.check.mock.calls.length).toBeGreaterThan(3);
|
||||
});
|
||||
it("retains startup work on standby and cancels future deadlines on shutdown", async () => {
|
||||
const f = fixture(); f.state.standby = true; f.rows.watchdogs.push({ id: "watchdog", companyId: "selected" });
|
||||
await f.worker.ready; await vi.advanceTimersByTimeAsync(90_000); expect(f.load).not.toHaveBeenCalled();
|
||||
f.state.standby = false; f.state.deadline = Date.now() + 60_000; f.worker.wake(); await f.settle();
|
||||
expect(f.check).toHaveBeenCalledOnce(); await f.worker.stop();
|
||||
await vi.advanceTimersByTimeAsync(120_000); expect(f.check).toHaveBeenCalledOnce();
|
||||
});
|
||||
});
|
||||
@@ -41,7 +41,7 @@ describeEmbeddedPostgres("issueTreeControlService", () => {
|
||||
beforeAll(async () => {
|
||||
if (externalUrl) {
|
||||
await applyPendingMigrations(externalUrl);
|
||||
db = createDb(externalUrl);
|
||||
db = createDb(externalUrl, { maxConnections: 6 });
|
||||
return;
|
||||
}
|
||||
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-issue-tree-control-");
|
||||
|
||||
@@ -10,7 +10,7 @@ afterEach(async () => { await coordinator?.stop(); vi.useRealTimers(); });
|
||||
it("preserves a subsecond replacement deadline when the owner does not require pass spacing", async () => {
|
||||
vi.useFakeTimers();
|
||||
const onError = vi.fn();
|
||||
const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) });
|
||||
const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []), transaction: vi.fn(async () => undefined) });
|
||||
coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => true, onError });
|
||||
const now = Date.now();
|
||||
let dueAt = now + 60_000;
|
||||
@@ -43,7 +43,7 @@ type Entry = { id: string; companyId: string; issueId: string; dueAt: number };
|
||||
function concurrentFixture(entries: Entry[], limit = 4, concurrent = true) {
|
||||
vi.useFakeTimers();
|
||||
const onError = vi.fn();
|
||||
const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) });
|
||||
const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []), transaction: vi.fn(async () => undefined) });
|
||||
let canRun = true;
|
||||
let releaseAll = false;
|
||||
const releases = new Map<string, () => void>();
|
||||
@@ -177,7 +177,7 @@ it("stops starting checks and awaits every admitted check during shutdown", asyn
|
||||
|
||||
it("keeps failures isolated and retries their deadline without repeating successful checks", async () => {
|
||||
vi.useFakeTimers();
|
||||
const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []) });
|
||||
const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []), transaction: vi.fn(async () => undefined) });
|
||||
const onError = vi.fn();
|
||||
coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => true, onError });
|
||||
const entries = [entry("1", "shared"), entry("2", "shared"), entry("3")];
|
||||
@@ -200,3 +200,98 @@ it("keeps failures isolated and retries their deadline without repeating success
|
||||
expect(load).toHaveBeenCalledOnce();
|
||||
expect(coordinator.nextWakeAt()).toBeNull();
|
||||
});
|
||||
|
||||
|
||||
it("joins one beforePass across keyed parallel checks while retaining per-task order", async () => {
|
||||
vi.useFakeTimers();
|
||||
const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []), transaction: vi.fn(async () => undefined) });
|
||||
const onError = vi.fn(), beforePass = vi.fn(async () => {});
|
||||
const started: string[] = [], completed = new Set<string>(), releases = new Map<string, () => void>();
|
||||
let releaseAll = false;
|
||||
const entries = [entry("first", "shared"), entry("second", "shared"), entry("independent", "other")];
|
||||
const load = vi.fn(async () => entries.filter(row => !completed.has(row.id)));
|
||||
coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => true, onError });
|
||||
const worker = registerScopedDeadlineWork(coordinator, {
|
||||
queue: DELIVERY_QUEUES.nativeResponsePresentation, minimumPassIntervalMs: 0,
|
||||
concurrentChecks: { limit: 2, key: (row: Entry) => `${row.companyId}:${row.issueId}` },
|
||||
reconstructionIntervalMs: 30_000, beforePass,
|
||||
decodeScope: hint => hint === "*" ? undefined : hint,
|
||||
matches: (row, companyId?: string) => companyId === undefined || row.companyId === companyId,
|
||||
load,
|
||||
check: async row => {
|
||||
started.push(row.id);
|
||||
if (!releaseAll) await new Promise<void>(resolve => releases.set(row.id, resolve));
|
||||
completed.add(row.id);
|
||||
return null;
|
||||
},
|
||||
}, () => true);
|
||||
try {
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
expect(started).toEqual(["first", "independent"]);
|
||||
expect(beforePass).toHaveBeenCalledOnce();
|
||||
worker.invalidate("company");
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
expect(beforePass).toHaveBeenCalledOnce();
|
||||
releases.get("independent")!();
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
expect(started).not.toContain("second");
|
||||
releases.get("first")!();
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
expect(started).toEqual(["first", "independent", "second"]);
|
||||
expect(beforePass).toHaveBeenCalledOnce();
|
||||
} finally {
|
||||
releaseAll = true; for (const release of releases.values()) release(); await worker.ready;
|
||||
}
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
expect(beforePass).toHaveBeenCalledTimes(2);
|
||||
expect(load).toHaveBeenCalledTimes(2);
|
||||
expect(coordinator.nextWakeAt()).toBeNull();
|
||||
await vi.advanceTimersByTimeAsync(120_000);
|
||||
expect(load).toHaveBeenCalledTimes(2); expect(beforePass).toHaveBeenCalledTimes(2);
|
||||
expect(onError).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("keeps an exact deadline and late reconstruction debt with concurrency enabled across drain", async () => {
|
||||
vi.useFakeTimers();
|
||||
const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []), transaction: vi.fn(async () => undefined) });
|
||||
const onError = vi.fn(), beforePass = vi.fn(async () => {});
|
||||
let canRun = true, completed = false, entered!: () => void, release!: () => void;
|
||||
const loading = new Promise<void>(resolve => { entered = resolve; });
|
||||
const held = new Promise<void>(resolve => { release = resolve; });
|
||||
const now = Date.now(), dueAt = now + 5_000;
|
||||
const candidate = { ...entry("saved", "task"), dueAt };
|
||||
const load = vi.fn(async () => { if (load.mock.calls.length === 2) { entered(); await held; } return completed ? [] : [candidate]; });
|
||||
const check = vi.fn(async () => { completed = true; return null; });
|
||||
coordinator = createDeliveryWorkCoordinator({ owner, canRun: () => canRun, onError });
|
||||
const worker = registerScopedDeadlineWork(coordinator, {
|
||||
queue: DELIVERY_QUEUES.nativeResponsePresentation, minimumPassIntervalMs: 0,
|
||||
concurrentChecks: { limit: 2, key: (row: Entry) => row.issueId },
|
||||
reconstructionIntervalMs: 30_000, beforePass,
|
||||
decodeScope: hint => hint === "*" ? undefined : hint,
|
||||
matches: (row, companyId?: string) => companyId === undefined || row.companyId === companyId,
|
||||
load, check,
|
||||
}, () => canRun);
|
||||
await worker.ready;
|
||||
worker.invalidate("company");
|
||||
try {
|
||||
await vi.advanceTimersByTimeAsync(1); await loading;
|
||||
worker.invalidate("company");
|
||||
} finally { release(); }
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
expect(load).toHaveBeenCalledTimes(2);
|
||||
expect(coordinator.nextWakeAt()).toBe(dueAt);
|
||||
await vi.advanceTimersByTimeAsync(4_999);
|
||||
expect(check).toHaveBeenCalledOnce(); expect(load).toHaveBeenCalledTimes(2);
|
||||
expect(coordinator.nextWakeAt()).toBe(now + 30_001);
|
||||
canRun = false; worker.wake();
|
||||
await vi.advanceTimersByTimeAsync(20_000);
|
||||
expect(load).toHaveBeenCalledTimes(2); expect(check).toHaveBeenCalledOnce();
|
||||
canRun = true; worker.wake();
|
||||
await vi.advanceTimersByTimeAsync(5_001);
|
||||
expect(load).toHaveBeenCalledTimes(3); expect(check).toHaveBeenCalledOnce();
|
||||
expect(beforePass).toHaveBeenCalledTimes(4);
|
||||
expect(coordinator.nextWakeAt()).toBeNull();
|
||||
await vi.advanceTimersByTimeAsync(120_000);
|
||||
expect(load).toHaveBeenCalledTimes(3); expect(check).toHaveBeenCalledOnce();
|
||||
expect(onError).not.toHaveBeenCalled();
|
||||
});
|
||||
@@ -134,6 +134,7 @@ const {
|
||||
})),
|
||||
reconcileHotRestartAdoption: vi.fn(async () => ({ mode: "none" })),
|
||||
reapOrphanedRuns: vi.fn(async () => ({ reaped: 0, runIds: [] })),
|
||||
registerIssueStateRecoveryWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
|
||||
registerRunDispatchWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
|
||||
registerScheduledRetryCancellationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
|
||||
registerStoppedInstructionCollectionWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })),
|
||||
@@ -927,6 +928,7 @@ describe("startServer feedback export wiring", () => {
|
||||
it.each([false, true])("registers retry and queued dispatch only when maintenance is enabled (%s)", async enabled => {
|
||||
loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled }));
|
||||
await startServer(); await settleStartupHousekeeping();
|
||||
expect(heartbeatServiceMock.registerIssueStateRecoveryWork).toHaveBeenCalledTimes(enabled ? 1 : 0);
|
||||
expect(heartbeatServiceMock.registerRunDispatchWork).toHaveBeenCalledTimes(enabled ? 1 : 0);
|
||||
if (enabled) {
|
||||
const [coordinator, admitted] = heartbeatServiceMock.registerRunDispatchWork.mock.calls[0];
|
||||
@@ -1334,11 +1336,11 @@ describe("startServer feedback export wiring", () => {
|
||||
await startServer();
|
||||
await settleStartupHousekeeping();
|
||||
expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1);
|
||||
expect(heartbeatServiceMock.sweepStaleIssueLocks).toHaveBeenCalledTimes(1);
|
||||
expect(heartbeatServiceMock.sweepStaleIssueLocks).not.toHaveBeenCalled();
|
||||
expect(intervalCallback).not.toBeNull();
|
||||
intervalCallback?.();
|
||||
await new Promise<void>((resolve) => setImmediate(resolve));
|
||||
expect(heartbeatServiceMock.sweepStaleIssueLocks).toHaveBeenCalledTimes(2);
|
||||
expect(heartbeatServiceMock.sweepStaleIssueLocks).not.toHaveBeenCalled();
|
||||
expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true, skipNativeRecoveryNotices: true, skipExternalConversationSettlement: true, skipExternalAgentSettlement: true, skipNativeEnvironmentSettlement: true });
|
||||
expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1);
|
||||
expect(heartbeatServiceMock.promoteDueScheduledRetries).not.toHaveBeenCalled();
|
||||
|
||||
@@ -1,8 +1,6 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { classifyTaskWatchdogSubtree, type TaskWatchdogClassifierIssue } from "../services/task-watchdogs.ts";
|
||||
|
||||
import { createIncompleteTaskWatchdogScanReporter } from "../services/task-watchdog-diagnostics.ts";
|
||||
|
||||
const companyId = "company-1";
|
||||
const sourceId = "source-1";
|
||||
const childId = "child-1";
|
||||
@@ -37,39 +35,6 @@ function classify(overrides: Partial<Parameters<typeof classifyTaskWatchdogSubtr
|
||||
}
|
||||
|
||||
describe("task watchdog subtree classifier", () => {
|
||||
it("reports an unchanged incomplete scan once across startup and a day of periodic ticks", () => {
|
||||
const warn = vi.fn();
|
||||
const report = createIncompleteTaskWatchdogScanReporter(warn);
|
||||
report({ incomplete: 2, incompleteIssueIds: [sourceId, childId] }, "startup");
|
||||
for (let tick = 0; tick < 2880; tick += 1) {
|
||||
report({ incomplete: 2, incompleteIssueIds: [childId, sourceId] }, "periodic");
|
||||
}
|
||||
expect(warn).toHaveBeenCalledTimes(1);
|
||||
expect(warn).toHaveBeenCalledWith(
|
||||
{ incomplete: 2, newIncompleteIssueIds: [sourceId, childId] },
|
||||
"startup task-watchdog reconciliation found incomplete subtree scans",
|
||||
);
|
||||
});
|
||||
|
||||
it("reports new incomplete issues and reports a resolved issue if it becomes incomplete again", () => {
|
||||
const warn = vi.fn();
|
||||
const report = createIncompleteTaskWatchdogScanReporter(warn);
|
||||
report({ incomplete: 1, incompleteIssueIds: [sourceId] }, "startup");
|
||||
report({ incomplete: 2, incompleteIssueIds: [sourceId, childId] }, "periodic");
|
||||
expect(warn).toHaveBeenLastCalledWith(
|
||||
{ incomplete: 2, newIncompleteIssueIds: [childId] },
|
||||
"periodic task-watchdog reconciliation found incomplete subtree scans",
|
||||
);
|
||||
report({ incomplete: 0, incompleteIssueIds: [] }, "periodic");
|
||||
expect(warn).toHaveBeenCalledTimes(2);
|
||||
report({ incomplete: 1, incompleteIssueIds: [sourceId] }, "periodic");
|
||||
expect(warn).toHaveBeenCalledTimes(3);
|
||||
expect(warn).toHaveBeenLastCalledWith(
|
||||
{ incomplete: 1, newIncompleteIssueIds: [sourceId] },
|
||||
"periodic task-watchdog reconciliation found incomplete subtree scans",
|
||||
);
|
||||
});
|
||||
|
||||
it("does not produce a stopped fingerprint from an incomplete scan", () => {
|
||||
const result = classify({ subtreeIncomplete: true });
|
||||
expect(result.state).toBe("incomplete");
|
||||
|
||||
+2
-74
@@ -109,7 +109,6 @@ import { createApp } from "./app.js";
|
||||
import { initializeCloudRuntimeConfig } from "./services/cloud-runtime-config.js";
|
||||
import { loadCloudBootstrapConfig, loadConfig, type Config } from "./config.js";
|
||||
import { logger } from "./middleware/logger.js";
|
||||
import { createIncompleteTaskWatchdogScanReporter } from "./services/task-watchdog-diagnostics.js";
|
||||
import { setStartupRecoveryPhase } from "./startup-recovery-state.js";
|
||||
import {
|
||||
StartupRefusalError,
|
||||
@@ -973,9 +972,6 @@ async function startServerWithDatabaseTeardown(
|
||||
}
|
||||
};
|
||||
const pluginWorkerManager = createPluginWorkerManager();
|
||||
const reportIncompleteWatchdogScans = createIncompleteTaskWatchdogScanReporter(
|
||||
(details, message) => logger.warn(details, message),
|
||||
);
|
||||
const heartbeat = config.heartbeatSchedulerEnabled
|
||||
? heartbeatService(db as any, { pluginWorkerManager })
|
||||
: null;
|
||||
@@ -1796,47 +1792,10 @@ async function startServerWithDatabaseTeardown(
|
||||
"startup session-goal recovery resumed durable agent goals",
|
||||
);
|
||||
}
|
||||
const reconciled = await heartbeat.reconcileStrandedAssignedIssues();
|
||||
if (
|
||||
reconciled.assignmentDispatched > 0 ||
|
||||
reconciled.dispatchRequeued > 0 ||
|
||||
reconciled.continuationRequeued > 0 ||
|
||||
reconciled.successfulRunHandoffEscalated > 0 ||
|
||||
reconciled.successfulRunHandoffRetried > 0 ||
|
||||
reconciled.escalated > 0
|
||||
) {
|
||||
logger.warn(
|
||||
{ ...reconciled },
|
||||
"startup heartbeat recovery changed assigned issue state",
|
||||
);
|
||||
}
|
||||
|
||||
const dependencyWakesReconciled = await heartbeat.reconcileResolvedDependencyWakes();
|
||||
if (dependencyWakesReconciled.healed > 0) {
|
||||
logger.warn(
|
||||
{ ...dependencyWakesReconciled },
|
||||
"startup dependency-wake reconciliation restored task execution paths",
|
||||
);
|
||||
}
|
||||
|
||||
const taskWatchdogsReconciled = await heartbeat.reconcileTaskWatchdogs();
|
||||
reportIncompleteWatchdogScans(taskWatchdogsReconciled, "startup");
|
||||
if (taskWatchdogsReconciled.triggered > 0) {
|
||||
logger.warn(
|
||||
{ ...taskWatchdogsReconciled },
|
||||
"startup task-watchdog reconciliation triggered watchdog work",
|
||||
);
|
||||
}
|
||||
|
||||
const scanned = await heartbeat.scanSilentActiveRuns();
|
||||
if (scanned.created > 0 || scanned.escalated > 0) {
|
||||
logger.warn({ ...scanned }, "startup active-run output watchdog created review work");
|
||||
}
|
||||
|
||||
const swept = await heartbeat.sweepStaleIssueLocks();
|
||||
if (swept.cleared > 0) {
|
||||
logger.warn({ ...swept }, "startup stale-lock sweeper cleared issue locks");
|
||||
}
|
||||
})().catch((err) => {
|
||||
logger.error({ err }, "startup heartbeat recovery failed");
|
||||
throw err;
|
||||
@@ -1845,6 +1804,8 @@ async function startServerWithDatabaseTeardown(
|
||||
await startupHeartbeatRecovery;
|
||||
}
|
||||
|
||||
if (config.heartbeatSchedulerEnabled) await heartbeat.registerIssueStateRecoveryWork(deliveryWork,
|
||||
() => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready;
|
||||
if (config.heartbeatSchedulerEnabled) await heartbeat.registerRunDispatchWork(deliveryWork,
|
||||
() => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready;
|
||||
if (config.heartbeatSchedulerEnabled) await heartbeat.registerScheduledRetryCancellationWork(deliveryWork,
|
||||
@@ -2051,33 +2012,6 @@ async function startServerWithDatabaseTeardown(
|
||||
.reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true, skipNativeStatusReassessment: true, skipRetainedNativeSessionCleanup: true, skipNativeWorkspaceFinalization: true, skipNativeResultCommit: true, skipNativeRunProjection: true, skipNativeRecoveryNotices: true, skipExternalConversationSettlement: true, skipExternalAgentSettlement: true, skipNativeEnvironmentSettlement: true })
|
||||
.then(async () => {
|
||||
await heartbeat.resumeQueuedRuns();
|
||||
const reconciled = await heartbeat.reconcileStrandedAssignedIssues({ skipExternalConversationSettlement: true });
|
||||
if (
|
||||
reconciled.assignmentDispatched > 0 ||
|
||||
reconciled.dispatchRequeued > 0 ||
|
||||
reconciled.continuationRequeued > 0 ||
|
||||
reconciled.successfulRunHandoffEscalated > 0 ||
|
||||
reconciled.successfulRunHandoffRetried > 0 ||
|
||||
reconciled.escalated > 0
|
||||
) {
|
||||
logger.warn(
|
||||
{ ...reconciled },
|
||||
"periodic heartbeat recovery changed assigned issue state",
|
||||
);
|
||||
}
|
||||
})
|
||||
.then(async () => {
|
||||
const reconciled = await heartbeat.reconcileResolvedDependencyWakes();
|
||||
if (reconciled.healed > 0) {
|
||||
logger.warn({ ...reconciled }, "periodic dependency-wake reconciliation restored task execution paths");
|
||||
}
|
||||
})
|
||||
.then(async () => {
|
||||
const reconciled = await heartbeat.reconcileTaskWatchdogs();
|
||||
reportIncompleteWatchdogScans(reconciled, "periodic");
|
||||
if (reconciled.triggered > 0) {
|
||||
logger.warn({ ...reconciled }, "periodic task-watchdog reconciliation triggered watchdog work");
|
||||
}
|
||||
})
|
||||
.then(async () => {
|
||||
const scanned = await heartbeat.scanSilentActiveRuns();
|
||||
@@ -2085,12 +2019,6 @@ async function startServerWithDatabaseTeardown(
|
||||
logger.warn({ ...scanned }, "periodic active-run output watchdog created review work");
|
||||
}
|
||||
})
|
||||
.then(async () => {
|
||||
const swept = await heartbeat.sweepStaleIssueLocks();
|
||||
if (swept.cleared > 0) {
|
||||
logger.warn({ ...swept }, "periodic stale-lock sweeper cleared issue locks");
|
||||
}
|
||||
})
|
||||
.catch((err) => {
|
||||
logger.error({ err }, "periodic heartbeat recovery failed");
|
||||
}));
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
|
||||
import { FAST_RESPONSE_AGENT_GUIDANCE, fastResponseHistoryBody } from "@paperclipai/shared";
|
||||
@@ -547,6 +548,7 @@ export async function resumeConversationForReset(db: Db, comment: typeof issueCo
|
||||
eq(issues.id, comment.issueId), eq(issues.companyId, comment.companyId),
|
||||
)).for("update");
|
||||
if (!isConversation(issue) || comment.conversationSessionGeneration != null) return;
|
||||
await notifyIssueRecovery(tx, { companyId: issue.companyId });
|
||||
const released = await tx.update(issueTreeHolds).set({
|
||||
status: "released", releasedAt: new Date(), updatedAt: new Date(),
|
||||
releasedByActorType: "user", releasedByUserId: comment.authorUserId,
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyIssueRecovery } from "../../issue-recovery-notifications.js";
|
||||
import { createProviderEffects, type ProviderEffectTarget } from "../provider-effects.js";
|
||||
|
||||
import type { EndpointRow, RuntimeContext, InboundRuntimeContext } from "../runtime-types.js";
|
||||
@@ -78,6 +79,7 @@ export function createTelegramAttachmentFailure(db: Db, options: TelegramAttachm
|
||||
})(),
|
||||
);
|
||||
const effect = await db.transaction(async (tx) => {
|
||||
await notifyIssueRecovery(tx, { companyId: input.endpoint.companyId });
|
||||
// Replace the optimistic attachment placeholder with an explicit safe
|
||||
// Paperclip-visible failure. The message link stays authoritative for
|
||||
// exact provider redelivery, while no agent wakeup is enqueued for an
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
|
||||
import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
|
||||
import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js";
|
||||
import { notifyInstructionCollection } from "./instruction-collection-notifications.js";
|
||||
@@ -71,6 +72,7 @@ export async function deleteCompany(db: Db, id: string) {
|
||||
const [existing] = await tx.select({ id: companies.id }).from(companies)
|
||||
.where(eq(companies.id, id)).for("no key update");
|
||||
if (!existing) return null;
|
||||
await notifyIssueRecovery(tx, { companyId: id });
|
||||
await notifyChatCompanyConfiguration(tx, id);
|
||||
await chatCompanyDeletion.deleteCompanyData(tx, id);
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync);
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
|
||||
import { createAgentLifecycle } from "./agent-lifecycle.js";
|
||||
import { agentAppearanceSchema, agentHarnessType, paperclipRunnerProfileForHarness, agentRunner, type AgentRunnerChoice } from "@paperclipai/shared";
|
||||
import { resolveNewAgentRunnerForCompany } from "./agent-runner-selection.js";
|
||||
@@ -6465,7 +6466,9 @@ export function companyPortabilityService(db: Db, storage?: StorageService) {
|
||||
}
|
||||
if (relationRows.length > 0) {
|
||||
const relationCompanyId = targetCompany.id;
|
||||
await db
|
||||
await db.transaction(async tx => {
|
||||
await notifyIssueRecovery(tx, { companyId: relationCompanyId });
|
||||
await tx
|
||||
.insert(issueRelations)
|
||||
.values(relationRows.map((row) => ({
|
||||
companyId: relationCompanyId,
|
||||
@@ -6476,6 +6479,7 @@ export function companyPortabilityService(db: Db, storage?: StorageService) {
|
||||
createdByUserId: actorUserId ?? null,
|
||||
})))
|
||||
.onConflictDoNothing();
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -2,6 +2,12 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";
|
||||
|
||||
/** Topics for existing durable queues; register intent before their writes. */
|
||||
export const DELIVERY_QUEUES = {
|
||||
orphanBlockers: "orphan-blockers",
|
||||
activeRecoveryActions: "active-recovery-actions",
|
||||
strandedAssignments: "stranded-assignments",
|
||||
resolvedDependencies: "resolved-dependencies",
|
||||
taskWatchdogs: "task-watchdogs",
|
||||
staleIssueLocks: "stale-issue-locks",
|
||||
scheduledRetryDispatch: "scheduled-retry-dispatch",
|
||||
queuedRunDispatch: "queued-run-dispatch",
|
||||
runEvidenceMigration: "run-evidence-migration",
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { and, asc, desc, eq, sql } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
@@ -275,6 +276,7 @@ export function documentService(db: Db) {
|
||||
.where(eq(issueDocuments.issueId, issue.id));
|
||||
const fallbackKey = nextAvailableDocumentKey(key, issueDocumentKeys.map((row) => row.key));
|
||||
|
||||
await notifyIssueRecovery(tx, { companyId: issue.companyId });
|
||||
const [document] = await tx
|
||||
.insert(documents)
|
||||
.values({
|
||||
@@ -376,6 +378,7 @@ export function documentService(db: Db) {
|
||||
});
|
||||
}
|
||||
|
||||
await notifyIssueRecovery(tx, { companyId: issue.companyId });
|
||||
const nextRevisionNumber = existing.latestRevisionNumber + 1;
|
||||
const [revision] = await tx
|
||||
.insert(documentRevisions)
|
||||
@@ -438,6 +441,7 @@ export function documentService(db: Db) {
|
||||
throw conflict("Document does not exist yet", { key });
|
||||
}
|
||||
|
||||
await notifyIssueRecovery(tx, { companyId: issue.companyId });
|
||||
const [document] = await tx
|
||||
.insert(documents)
|
||||
.values({
|
||||
@@ -592,6 +596,9 @@ export function documentService(db: Db) {
|
||||
});
|
||||
}
|
||||
await db.transaction(async (tx) => {
|
||||
for (const companyId of new Set(rows.map(row => row.companyId))) {
|
||||
await notifyIssueRecovery(tx, { companyId });
|
||||
}
|
||||
await insertRowsInChunks(tx, documents, documentRows);
|
||||
await insertRowsInChunks(tx, documentRevisions, revisionRows);
|
||||
await insertRowsInChunks(tx, issueDocuments, linkRows);
|
||||
@@ -644,6 +651,7 @@ export function documentService(db: Db) {
|
||||
});
|
||||
}
|
||||
|
||||
await notifyIssueRecovery(tx, { companyId: existing.companyId });
|
||||
const now = new Date();
|
||||
const nextRevisionNumber = existing.latestRevisionNumber + 1;
|
||||
const [restoredRevision] = await tx
|
||||
@@ -722,6 +730,7 @@ export function documentService(db: Db) {
|
||||
};
|
||||
}
|
||||
|
||||
await notifyIssueRecovery(tx, { companyId: existing.companyId });
|
||||
const now = new Date();
|
||||
await tx
|
||||
.update(documents)
|
||||
@@ -769,6 +778,7 @@ export function documentService(db: Db) {
|
||||
};
|
||||
}
|
||||
|
||||
await notifyIssueRecovery(tx, { companyId: existing.companyId });
|
||||
const now = new Date();
|
||||
await tx
|
||||
.update(documents)
|
||||
@@ -817,6 +827,7 @@ export function documentService(db: Db) {
|
||||
});
|
||||
}
|
||||
|
||||
await notifyIssueRecovery(tx, { companyId: existing.companyId });
|
||||
await tx.delete(issueDocuments).where(eq(issueDocuments.documentId, existing.id));
|
||||
await tx.delete(documents).where(eq(documents.id, existing.id));
|
||||
|
||||
|
||||
@@ -1,7 +1,9 @@
|
||||
import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
import type { ExternalConversationSettlementScope } from "./slack-conversation-lifecycle.js";
|
||||
|
||||
export function notifyExternalConversationSettlement(tx: object, scope?: ExternalConversationSettlementScope) {
|
||||
export async function notifyExternalConversationSettlement(tx: object, scope?: ExternalConversationSettlementScope) {
|
||||
await notifyIssueRecovery(tx, scope);
|
||||
return notifyDeliveryWork(tx, DELIVERY_QUEUES.externalConversationSettlement,
|
||||
scope ? JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId }) : "*");
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
import { registerIssueRecoveryWork } from "./recovery/work.js";
|
||||
import { registerRunDispatchWork } from "./heartbeat/dispatch-work.js";
|
||||
import { NATIVE_RETAINED_OWNER_WAIT } from "./native-runtime/native-controller-recovery.js";
|
||||
import { externalConversationSettlementWork } from "./external-conversation-settlement-work.js";
|
||||
@@ -1618,15 +1619,15 @@ export function heartbeatService(
|
||||
void reportRunFailure(db, updated, failureReport);
|
||||
}
|
||||
|
||||
async function reconcileStrandedAssignedIssues(opts?: { skipExternalConversationSettlement?: boolean }) {
|
||||
async function reconcileStrandedAssignedIssues(opts?: Parameters<typeof recovery.reconcileStrandedAssignedIssues>[0]) {
|
||||
return recovery.reconcileStrandedAssignedIssues({
|
||||
skipExternalConversationSettlement: opts?.skipExternalConversationSettlement,
|
||||
...opts,
|
||||
issueCreatedAtGte: await getWorktreeExecutionCutoff(),
|
||||
});
|
||||
}
|
||||
|
||||
async function sweepStaleIssueLocks() {
|
||||
return recovery.sweepStaleIssueLocks();
|
||||
async function sweepStaleIssueLocks(opts?: Parameters<typeof recovery.sweepStaleIssueLocks>[0]) {
|
||||
return recovery.sweepStaleIssueLocks(opts);
|
||||
}
|
||||
|
||||
function issueIdFromRunContext(contextSnapshot: unknown) {
|
||||
@@ -1656,10 +1657,7 @@ export function heartbeatService(
|
||||
});
|
||||
}
|
||||
|
||||
async function reconcileTaskWatchdogs(opts?: {
|
||||
companyId?: string | null;
|
||||
runId?: string | null;
|
||||
}) {
|
||||
async function reconcileTaskWatchdogs(opts?: Parameters<typeof taskWatchdogs.reconcileTaskWatchdogs>[0]) {
|
||||
return taskWatchdogs.reconcileTaskWatchdogs({
|
||||
...opts,
|
||||
issueCreatedAtGte: await getWorktreeExecutionCutoff(),
|
||||
@@ -1684,10 +1682,7 @@ export function heartbeatService(
|
||||
return recovery.buildRunOutputSilence(run, now);
|
||||
}
|
||||
|
||||
async function reconcileResolvedDependencyWakes(opts?: {
|
||||
runId?: string | null;
|
||||
companyId?: string | null;
|
||||
}) {
|
||||
async function reconcileResolvedDependencyWakes(opts?: Parameters<typeof recovery.reconcileResolvedDependencyWakeBackstop>[0]) {
|
||||
return recovery.reconcileResolvedDependencyWakeBackstop(opts);
|
||||
}
|
||||
|
||||
@@ -4236,6 +4231,16 @@ export function heartbeatService(
|
||||
computeTaskDrain,
|
||||
applyTaskDrain,
|
||||
|
||||
registerIssueStateRecoveryWork: (coordinator: Parameters<typeof registerIssueRecoveryWork>[1], canRun: () => boolean) =>
|
||||
registerIssueRecoveryWork(db, coordinator, {
|
||||
stranded: scope => reconcileStrandedAssignedIssues({ ...scope, skipExternalConversationSettlement: true, skipRelatedRecovery: true }),
|
||||
orphanBlockers: recovery.reconcileUnassignedBlockingIssues,
|
||||
recoveryActions: recovery.reconcileActiveRecoveryActions,
|
||||
dependencies: reconcileResolvedDependencyWakes,
|
||||
locks: sweepStaleIssueLocks,
|
||||
watchdogs: reconcileTaskWatchdogs,
|
||||
schedulingSuppressed: async () => (await getSchedulingSuppression(true)).suppressed,
|
||||
}, () => canRun() && readTaskDrain(new Date()) === null),
|
||||
registerRunDispatchWork: (coordinator: Parameters<typeof registerRunDispatchWork>[0], canRun: () => boolean) =>
|
||||
registerRunDispatchWork(coordinator, {
|
||||
promoteDueScheduledRetries,
|
||||
|
||||
@@ -624,11 +624,14 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService
|
||||
}
|
||||
}
|
||||
}
|
||||
const chatBecameEnabled = !toExperimentalView(current.experimental).enableAgentChat
|
||||
&& toExperimentalView(nextExperimental).enableAgentChat;
|
||||
const now = new Date();
|
||||
const [updated] = await db.transaction(async tx => {
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.mcpEvents, "*");
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, "*");
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.grokDispatch);
|
||||
if (chatBecameEnabled) await notifyDeliveryWork(tx, DELIVERY_QUEUES.strandedAssignments, "*");
|
||||
if (patch.enableWorktreeRunExecution !== undefined) await notifyScheduledRetryCancellation(tx);
|
||||
if (patch.enableWorktreeRunExecution !== undefined) {
|
||||
await notifyRunDispatch(tx);
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
|
||||
import type { PgInsertValue } from "drizzle-orm/pg-core";
|
||||
import { issueComments, type Db } from "@paperclipai/db";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
@@ -6,6 +7,8 @@ type WriteDb = Db | Parameters<Parameters<Db["transaction"]>[0]>[0];
|
||||
|
||||
export function insertIssueComment(db: WriteDb, input: PgInsertValue<typeof issueComments> & { companyId: string }) {
|
||||
return db.transaction(async tx => {
|
||||
await notifyIssueRecovery(tx, { companyId: input.companyId,
|
||||
issueId: typeof input.issueId === "string" ? input.issueId : undefined });
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, `company:${input.companyId}`);
|
||||
return tx.insert(issueComments).values(input).returning();
|
||||
});
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
|
||||
import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js";
|
||||
import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
|
||||
import { writeExecutionControl } from "./execution-control-notifications.js";
|
||||
@@ -322,6 +323,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
|
||||
): Promise<IssueRecoveryAction> {
|
||||
try {
|
||||
const created = await db.transaction(async (tx) => {
|
||||
await notifyIssueRecovery(tx, { companyId: input.companyId });
|
||||
await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId });
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionDisposition);
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionContinuation);
|
||||
@@ -413,6 +415,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
|
||||
Math.min(nextAttemptCount, effectiveMaxAttempts),
|
||||
);
|
||||
const [exhausted] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], async tx => {
|
||||
await notifyIssueRecovery(tx, { companyId: input.companyId });
|
||||
await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId });
|
||||
if ((!input.preserveExistingOwner && existing.cause !== input.cause)
|
||||
|| (input.evidence?.runId !== undefined && input.evidence.runId !== existing.evidence?.runId)) {
|
||||
@@ -474,6 +477,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
|
||||
return toReadModel(exhausted);
|
||||
}
|
||||
const [updated] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], async tx => {
|
||||
await notifyIssueRecovery(tx, { companyId: input.companyId });
|
||||
await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId });
|
||||
if ((!input.preserveExistingOwner && existing.cause !== input.cause)
|
||||
|| (input.evidence?.runId !== undefined && input.evidence.runId !== existing.evidence?.runId)) {
|
||||
@@ -548,6 +552,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
|
||||
|
||||
try {
|
||||
const [created] = await writeExecutionControl(db, [DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.executionContinuation], async tx => {
|
||||
await notifyIssueRecovery(tx, { companyId: input.companyId });
|
||||
await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId });
|
||||
await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.sourceIssueId });
|
||||
return tx
|
||||
@@ -595,6 +600,7 @@ export function issueRecoveryActionService(db: DbOrTransaction) {
|
||||
const pending = await tx.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions)
|
||||
.where(and(...predicates)).limit(1);
|
||||
if (!pending.length) return [];
|
||||
await notifyIssueRecovery(tx, { companyId: input.companyId });
|
||||
await notifyNativeRecoveryNotices(tx, { companyId: input.companyId, issueId: input.sourceIssueId });
|
||||
await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.companyId, issueId: input.sourceIssueId });
|
||||
return tx
|
||||
|
||||
@@ -0,0 +1,16 @@
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
|
||||
/** An issue hint is reserved for local facts such as comments. Lifecycle,
|
||||
* graph, run and eligibility changes retain company scope because their
|
||||
* consequences can include ancestors and shared execution references. */
|
||||
export type IssueRecoveryScope = { companyId: string; issueId?: string };
|
||||
|
||||
export async function notifyIssueRecovery(transaction: object, scope?: IssueRecoveryScope) {
|
||||
const companyHint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*";
|
||||
const issueHint = scope ? JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId }) : "*";
|
||||
for (const queue of [DELIVERY_QUEUES.orphanBlockers, DELIVERY_QUEUES.activeRecoveryActions, DELIVERY_QUEUES.strandedAssignments, DELIVERY_QUEUES.resolvedDependencies,
|
||||
DELIVERY_QUEUES.taskWatchdogs, DELIVERY_QUEUES.staleIssueLocks]) {
|
||||
await notifyDeliveryWork(transaction, queue,
|
||||
queue === DELIVERY_QUEUES.strandedAssignments || queue === DELIVERY_QUEUES.staleIssueLocks ? issueHint : companyHint);
|
||||
}
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
|
||||
import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
|
||||
@@ -758,6 +759,7 @@ export function issueTreeControlService(db: Db) {
|
||||
const releaseReason = input.reason ?? "Subtree resume applied.";
|
||||
|
||||
const { hold: resumeHold } = await db.transaction(async (tx) => {
|
||||
await notifyIssueRecovery(tx, { companyId });
|
||||
const [createdHold] = await tx
|
||||
.insert(issueTreeHolds)
|
||||
.values({
|
||||
@@ -836,6 +838,7 @@ export function issueTreeControlService(db: Db) {
|
||||
}
|
||||
|
||||
const { hold, members } = await db.transaction(async (tx) => {
|
||||
await notifyIssueRecovery(tx, { companyId });
|
||||
const [createdHold] = await tx
|
||||
.insert(issueTreeHolds)
|
||||
.values({
|
||||
@@ -1177,7 +1180,9 @@ export function issueTreeControlService(db: Db) {
|
||||
throw conflict("Issue tree hold is already released");
|
||||
}
|
||||
|
||||
const [updated] = await db
|
||||
const [updated] = await db.transaction(async tx => {
|
||||
await notifyIssueRecovery(tx, { companyId });
|
||||
return tx
|
||||
.update(issueTreeHolds)
|
||||
.set({
|
||||
status: "released",
|
||||
@@ -1195,6 +1200,7 @@ export function issueTreeControlService(db: Db) {
|
||||
})
|
||||
.where(and(eq(issueTreeHolds.id, holdId), eq(issueTreeHolds.companyId, companyId)))
|
||||
.returning();
|
||||
});
|
||||
|
||||
const members = await db
|
||||
.select()
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
|
||||
import { notifyRunDispatch } from "./run-dispatch-notifications.js";
|
||||
import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js";
|
||||
import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js";
|
||||
@@ -7628,6 +7629,7 @@ export function issueService(db: Db) {
|
||||
await assertNoBlockingCycles(companyId, issueId, deduped, dbOrTx);
|
||||
}
|
||||
|
||||
await notifyIssueRecovery(dbOrTx, { companyId });
|
||||
await dbOrTx
|
||||
.delete(issueRelations)
|
||||
.where(
|
||||
@@ -9676,27 +9678,30 @@ export function issueService(db: Db) {
|
||||
});
|
||||
|
||||
if (blockParentUntilDone) {
|
||||
const existingBlockers = await db
|
||||
.select({ blockerIssueId: issueRelations.issueId })
|
||||
.from(issueRelations)
|
||||
.where(
|
||||
and(
|
||||
eq(issueRelations.companyId, parent.companyId),
|
||||
eq(issueRelations.relatedIssueId, parent.id),
|
||||
eq(issueRelations.type, "blocks"),
|
||||
),
|
||||
await db.transaction(async tx => {
|
||||
const existingBlockers = await tx
|
||||
.select({ blockerIssueId: issueRelations.issueId })
|
||||
.from(issueRelations)
|
||||
.where(
|
||||
and(
|
||||
eq(issueRelations.companyId, parent.companyId),
|
||||
eq(issueRelations.relatedIssueId, parent.id),
|
||||
eq(issueRelations.type, "blocks"),
|
||||
),
|
||||
);
|
||||
await syncBlockedByIssueIds(
|
||||
parent.id,
|
||||
parent.companyId,
|
||||
[
|
||||
...new Set([
|
||||
...existingBlockers.map((row) => row.blockerIssueId),
|
||||
child.id,
|
||||
]),
|
||||
],
|
||||
{ agentId: actorAgentId ?? null, userId: actorUserId ?? null },
|
||||
tx,
|
||||
);
|
||||
await syncBlockedByIssueIds(
|
||||
parent.id,
|
||||
parent.companyId,
|
||||
[
|
||||
...new Set([
|
||||
...existingBlockers.map((row) => row.blockerIssueId),
|
||||
child.id,
|
||||
]),
|
||||
],
|
||||
{ agentId: actorAgentId ?? null, userId: actorUserId ?? null },
|
||||
);
|
||||
});
|
||||
[child] = await withIssueRelationSummaries(
|
||||
parent.companyId,
|
||||
[child],
|
||||
@@ -11495,7 +11500,9 @@ export function issueService(db: Db) {
|
||||
await notifyTerminalWorkspaceWork(tx, `issue:${id}`);
|
||||
}
|
||||
if (patch.status !== undefined || patch.assigneeAgentId !== undefined || patch.assigneeUserId !== undefined
|
||||
|| patch.hiddenAt !== undefined || patch.conversationAgentId !== undefined) {
|
||||
|| patch.hiddenAt !== undefined || patch.conversationAgentId !== undefined || patch.parentId !== undefined
|
||||
|| patch.executionState !== undefined || patch.executionRunId !== undefined || patch.checkoutRunId !== undefined
|
||||
|| blockedByIssueIds !== undefined) {
|
||||
await notifyRunDispatch(tx, { companyId: receiptExisting.companyId });
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
|
||||
}
|
||||
@@ -13816,6 +13823,7 @@ export function issueService(db: Db) {
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!existing) return null;
|
||||
|
||||
await notifyIssueRecovery(tx, { companyId: existing.companyId });
|
||||
await tx.delete(issueAttachments).where(eq(issueAttachments.id, id));
|
||||
await tx.delete(assets).where(eq(assets.id, existing.assetId));
|
||||
return existing;
|
||||
|
||||
@@ -77,14 +77,18 @@ async function renewOnLivenessPool(
|
||||
});
|
||||
}
|
||||
|
||||
export async function hasLiveLegacyController(db: Db, run: Run): Promise<boolean> {
|
||||
if (run.runtimeMode === "native" || !run.controllerBootId) return false;
|
||||
const [owner] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
|
||||
export async function liveLegacyControllerExpiresAt(db: Db, run: Run): Promise<Date | null> {
|
||||
if (run.runtimeMode === "native" || !run.controllerBootId) return null;
|
||||
const [owner] = await db.select({ expiresAt: heartbeatRuns.controllerLeaseExpiresAt }).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId),
|
||||
eq(heartbeatRuns.status, "running"),
|
||||
gt(heartbeatRuns.controllerLeaseExpiresAt, sql`clock_timestamp()`),
|
||||
));
|
||||
return Boolean(owner);
|
||||
return owner?.expiresAt ?? null;
|
||||
}
|
||||
|
||||
export async function hasLiveLegacyController(db: Db, run: Run): Promise<boolean> {
|
||||
return (await liveLegacyControllerExpiresAt(db, run)) !== null;
|
||||
}
|
||||
|
||||
export type RevokedLegacyControllerState = Pick<Run, "executionStage" | "processPid" | "processGroupId">;
|
||||
|
||||
@@ -73,7 +73,7 @@ describe("native same-conversation chat attachment reuse", () => {
|
||||
beforeAll(async () => {
|
||||
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_CHAT_REUSE_TEST_DATABASE_URL");
|
||||
if (!external) temporary = await startEmbeddedPostgresTestDatabase("native-chat-reuse-");
|
||||
db = createDb(external ?? temporary!.connectionString);
|
||||
db = createDb(external ?? temporary!.connectionString, { maxConnections: 6 });
|
||||
const storageRoot = await mkdtemp(
|
||||
path.join(tmpdir(), "paperclip-chat-reuse-"),
|
||||
);
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyIssueRecovery } from "../issue-recovery-notifications.js";
|
||||
import { createHash } from "node:crypto";
|
||||
|
||||
import { and, desc, eq, inArray, isNull, sql } from "drizzle-orm";
|
||||
@@ -1582,7 +1583,9 @@ export async function prepareReusedChatAttachment(input: {
|
||||
if (!workProduct) {
|
||||
throw new Error("paperclip_runner_chat_attachment_origin_not_persisted");
|
||||
}
|
||||
await input.db
|
||||
await input.db.transaction(async tx => {
|
||||
await notifyIssueRecovery(tx, { companyId: input.binding.companyId });
|
||||
await tx
|
||||
.update(issueWorkProducts)
|
||||
.set({
|
||||
title: input.title,
|
||||
@@ -1602,6 +1605,7 @@ export async function prepareReusedChatAttachment(input: {
|
||||
eq(issueWorkProducts.createdByRunId, input.binding.runId),
|
||||
),
|
||||
);
|
||||
});
|
||||
const comment = await issueService(input.db).addComment(
|
||||
input.binding.issueId,
|
||||
`Prepared ${input.title} for this response.`,
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js";
|
||||
import type { PrpStructuredRunResult } from "../../vendor/paperclip-runner/index.js";
|
||||
import { nativeCompletionFeedback } from "./native-completion-feedback.js";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
@@ -61,10 +62,9 @@ describe("native runner file handoff", () => {
|
||||
const runId = "00000000-0000-4000-8000-000000009104";
|
||||
|
||||
beforeAll(async () => {
|
||||
temporary = await startEmbeddedPostgresTestDatabase(
|
||||
"native-runner-file-handoff-",
|
||||
);
|
||||
db = createDb(temporary.connectionString);
|
||||
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_HANDOFF_TEST_DATABASE_URL");
|
||||
if (!external) temporary = await startEmbeddedPostgresTestDatabase("native-runner-file-handoff-");
|
||||
db = createDb(external ?? temporary!.connectionString, { maxConnections: 6 });
|
||||
temporaryRoot = await mkdtemp(
|
||||
path.join(tmpdir(), "paperclip-native-file-handoff-"),
|
||||
);
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyIssueRecovery } from "../issue-recovery-notifications.js";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { execFile } from "node:child_process";
|
||||
import { constants } from "node:fs";
|
||||
@@ -1265,7 +1266,9 @@ export async function prepareNativeRunnerFileHandoff(input: {
|
||||
) {
|
||||
throw new Error("paperclip_runner_file_handoff_origin_not_persisted");
|
||||
}
|
||||
await input.db
|
||||
await input.db.transaction(async tx => {
|
||||
await notifyIssueRecovery(tx, { companyId: input.binding.companyId });
|
||||
await tx
|
||||
.update(issueWorkProducts)
|
||||
.set({ title: verified.title, updatedAt: new Date() })
|
||||
.where(
|
||||
@@ -1276,6 +1279,7 @@ export async function prepareNativeRunnerFileHandoff(input: {
|
||||
eq(issueWorkProducts.createdByRunId, input.binding.runId),
|
||||
),
|
||||
);
|
||||
});
|
||||
const comment = await issueService(input.db).addComment(
|
||||
input.binding.issueId,
|
||||
`Prepared ${verified.title} for this response.`,
|
||||
|
||||
@@ -131,7 +131,7 @@ describe("native workspace finalization recovery", () => {
|
||||
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_WORKSPACE_FINALIZER_TEST_DATABASE_URL");
|
||||
if (!external) temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-workspace-finalizer-");
|
||||
connectionString = external ?? temporary!.connectionString;
|
||||
db = createDb(connectionString);
|
||||
db = createDb(connectionString, { maxConnections: 6 });
|
||||
|
||||
await db.insert(companies).values([
|
||||
{ id: companyId, name: "Native workspace owner", issuePrefix: "NWO" },
|
||||
@@ -452,7 +452,7 @@ describe("native workspace finalization recovery", () => {
|
||||
const seeded = await seedRun({ executionWorkspaceId: randomUUID(), title: "Live copyback" });
|
||||
const live = await withNativeWorkspaceFinalizationOwnership({ db, companyId, runId: seeded.runId }, async (ownership) => {
|
||||
await ownership.assertHeld();
|
||||
expect(await resumeNativeWorkspaceFinalization({ db: createDb(connectionString), runId: seeded.runId })).toBeNull();
|
||||
expect(await resumeNativeWorkspaceFinalization({ db: createDb(connectionString, { maxConnections: 6 }), runId: seeded.runId })).toBeNull();
|
||||
expect(await db.select().from(workspaceOperations).where(eq(workspaceOperations.heartbeatRunId, seeded.runId))).toEqual([]);
|
||||
return "copied";
|
||||
});
|
||||
@@ -476,7 +476,7 @@ describe("native workspace finalization recovery", () => {
|
||||
try {
|
||||
await entered;
|
||||
await db.execute(sql`select pg_terminate_backend(pid) from pg_locks where locktype = 'advisory' and database = (select oid from pg_database where datname = current_database()) and pid <> pg_backend_pid()`);
|
||||
const second = await withNativeWorkspaceFinalizationOwnership({ db: createDb(connectionString), companyId, runId: seeded.runId }, async () => "must not run");
|
||||
const second = await withNativeWorkspaceFinalizationOwnership({ db: createDb(connectionString, { maxConnections: 6 }), companyId, runId: seeded.runId }, async () => "must not run");
|
||||
expect(second).toEqual({ acquired: false, reason: "busy" });
|
||||
const [held] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId));
|
||||
expect(held.runnerProfileJson?.nativeWorkspaceFinalizationOwner).toBeTruthy();
|
||||
@@ -489,7 +489,7 @@ describe("native workspace finalization recovery", () => {
|
||||
|
||||
it("recovers its own joined copyback receipt after the application pool disconnects during cleanup", async () => {
|
||||
const seeded = await seedRun({ executionWorkspaceId: randomUUID(), title: "Reconnect joined copyback" });
|
||||
const interruptedDb = createDb(connectionString);
|
||||
const interruptedDb = createDb(connectionString, { maxConnections: 6 });
|
||||
let joined = false;
|
||||
await expect(withNativeWorkspaceFinalizationOwnership({ db: interruptedDb, companyId, runId: seeded.runId }, async () => {
|
||||
await fs.writeFile(path.join(workspaceRoot, "joined-copyback.txt"), "durable work");
|
||||
@@ -507,7 +507,7 @@ describe("native workspace finalization recovery", () => {
|
||||
.toEqual({ acquired: false, reason: "owner_unverified" });
|
||||
}
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: beforeReconnect.runnerProfileJson }).where(eq(heartbeatRuns.id, seeded.runId));
|
||||
expect(await withNativeWorkspaceFinalizationOwnership({ db: createDb(connectionString), companyId, runId: seeded.runId }, async () => "resumed"))
|
||||
expect(await withNativeWorkspaceFinalizationOwnership({ db: createDb(connectionString, { maxConnections: 6 }), companyId, runId: seeded.runId }, async () => "resumed"))
|
||||
.toEqual({ acquired: true, value: "resumed" });
|
||||
});
|
||||
|
||||
|
||||
@@ -1,8 +1,10 @@
|
||||
import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
|
||||
export type NativeStatusReassessmentScope = { companyId: string; issueId?: string; runId?: string; agentId?: string };
|
||||
|
||||
export function notifyNativeStatusReassessment(tx: object, scope: NativeStatusReassessmentScope) {
|
||||
export async function notifyNativeStatusReassessment(tx: object, scope: NativeStatusReassessmentScope) {
|
||||
await notifyIssueRecovery(tx, scope ? { companyId: scope.companyId } : undefined);
|
||||
return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeStatusReassessment,
|
||||
JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, runId: scope.runId, agentId: scope.agentId }));
|
||||
}
|
||||
@@ -27,7 +27,7 @@ import {
|
||||
decideLegacyContinuation, legacyDispositionEpisode, legacyDispositionFingerprint,
|
||||
LEGACY_DISPOSITION_REPAIR_INSTRUCTION, type LegacyDispositionEpisode,
|
||||
} from "./legacy-continuation.js";
|
||||
import { CONTROLLER_LEASE_LOST_FAILURE_CODE, hasLiveLegacyController } from "../legacy-controller-lease.js";
|
||||
import { CONTROLLER_LEASE_LOST_FAILURE_CODE, hasLiveLegacyController, liveLegacyControllerExpiresAt } from "../legacy-controller-lease.js";
|
||||
import { isCompletedOnboardingHandoffWake } from "../chat-completion-delivery.js";
|
||||
import { instanceSettingsService } from "../instance-settings.js";
|
||||
import { isWaitingConversation, settleConversationTurnForRun, deliverConversationComments } from "../agent-conversations.js";
|
||||
@@ -245,6 +245,7 @@ type ResolvedDependencyWakeBackstopOptions = {
|
||||
runId?: string | null;
|
||||
companyId?: string | null;
|
||||
blockerIssueId?: string | null;
|
||||
issueId?: string;
|
||||
source?: ResolvedDependencyWakeBackstopSource;
|
||||
};
|
||||
|
||||
@@ -2202,7 +2203,7 @@ export function recoveryService(
|
||||
return Boolean(budgetBlock);
|
||||
}
|
||||
|
||||
async function reconcileUnassignedBlockingIssues() {
|
||||
async function reconcileUnassignedBlockingIssues(opts?: { companyId?: string }) {
|
||||
const candidates = await db
|
||||
.select({
|
||||
id: issues.id,
|
||||
@@ -2216,6 +2217,8 @@ export function recoveryService(
|
||||
.where(
|
||||
and(
|
||||
eq(issueRelations.type, "blocks"),
|
||||
eq(issueRelations.companyId, issues.companyId),
|
||||
opts?.companyId ? eq(issues.companyId, opts.companyId) : undefined,
|
||||
inArray(issues.status, ["todo", "blocked"]),
|
||||
isNull(issues.assigneeAgentId),
|
||||
isNull(issues.assigneeUserId),
|
||||
@@ -3500,7 +3503,7 @@ export function recoveryService(
|
||||
return Boolean(run || wake);
|
||||
}
|
||||
|
||||
async function reconcileActiveRecoveryActions() {
|
||||
async function reconcileActiveRecoveryActions(opts?: { companyId?: string }) {
|
||||
const rows = await db
|
||||
.select({
|
||||
action: issueRecoveryActions,
|
||||
@@ -3514,7 +3517,8 @@ export function recoveryService(
|
||||
eq(issues.companyId, issueRecoveryActions.companyId),
|
||||
),
|
||||
)
|
||||
.where(inArray(issueRecoveryActions.status, ["active", "escalated"]));
|
||||
.where(and(inArray(issueRecoveryActions.status, ["active", "escalated"]),
|
||||
opts?.companyId ? eq(issueRecoveryActions.companyId, opts.companyId) : undefined));
|
||||
|
||||
const result = {
|
||||
requeued: 0,
|
||||
@@ -4526,14 +4530,20 @@ export function recoveryService(
|
||||
}
|
||||
|
||||
async function reconcileStrandedAssignedIssues(opts?: {
|
||||
companyId?: string;
|
||||
issueId?: string;
|
||||
onDeferredDeadline?: (at: number) => void;
|
||||
issueCreatedAtGte?: Date | null;
|
||||
skipExternalConversationSettlement?: boolean;
|
||||
skipRelatedRecovery?: boolean;
|
||||
}) {
|
||||
const candidates = await db
|
||||
.select(recoveryIssueFields)
|
||||
.from(issues)
|
||||
.where(
|
||||
and(
|
||||
opts?.companyId ? eq(issues.companyId, opts.companyId) : undefined,
|
||||
opts?.issueId ? eq(issues.id, opts.issueId) : undefined,
|
||||
isNull(issues.assigneeUserId),
|
||||
inArray(issues.status, ["todo", "in_progress", "in_review"]),
|
||||
or(
|
||||
@@ -5714,6 +5724,7 @@ export function recoveryService(
|
||||
classification.baseBackoffMs *
|
||||
Math.pow(2, Math.max(0, consecutive - 1));
|
||||
if (elapsed < requiredDelay) {
|
||||
opts?.onDeferredDeadline?.(latestFinishedAt.getTime() + requiredDelay);
|
||||
result.skipped += 1;
|
||||
continue;
|
||||
}
|
||||
@@ -5773,16 +5784,18 @@ export function recoveryService(
|
||||
}
|
||||
}
|
||||
|
||||
const orphanBlockerRecovery = await reconcileUnassignedBlockingIssues();
|
||||
result.orphanBlockersAssigned = orphanBlockerRecovery.assigned;
|
||||
result.skipped += orphanBlockerRecovery.skipped;
|
||||
result.issueIds.push(...orphanBlockerRecovery.issueIds);
|
||||
if (!opts?.skipRelatedRecovery) {
|
||||
const orphanBlockerRecovery = await reconcileUnassignedBlockingIssues(opts);
|
||||
result.orphanBlockersAssigned = orphanBlockerRecovery.assigned;
|
||||
result.skipped += orphanBlockerRecovery.skipped;
|
||||
result.issueIds.push(...orphanBlockerRecovery.issueIds);
|
||||
|
||||
const activeRecovery = await reconcileActiveRecoveryActions();
|
||||
result.continuationRequeued += activeRecovery.requeued;
|
||||
result.escalated += activeRecovery.escalated;
|
||||
result.skipped += activeRecovery.skipped;
|
||||
result.issueIds.push(...activeRecovery.issueIds);
|
||||
const activeRecovery = await reconcileActiveRecoveryActions(opts);
|
||||
result.continuationRequeued += activeRecovery.requeued;
|
||||
result.escalated += activeRecovery.escalated;
|
||||
result.skipped += activeRecovery.skipped;
|
||||
result.issueIds.push(...activeRecovery.issueIds);
|
||||
}
|
||||
result.issueIds = [...new Set(result.issueIds)];
|
||||
|
||||
return result;
|
||||
@@ -5815,7 +5828,7 @@ export function recoveryService(
|
||||
source === "workspace.finalize"
|
||||
? "workspace_finalize_reconciliation"
|
||||
: "issue_graph_liveness_reconciliation";
|
||||
const useCursor = !opts?.blockerIssueId;
|
||||
const useCursor = !opts?.blockerIssueId && !opts?.issueId;
|
||||
|
||||
const queryCandidates = (afterIssueId: string | null) => {
|
||||
const filters = [
|
||||
@@ -5825,6 +5838,7 @@ export function recoveryService(
|
||||
sql`${issues.assigneeAgentId} is not null`,
|
||||
];
|
||||
if (opts?.companyId) filters.push(eq(issues.companyId, opts.companyId));
|
||||
if (opts?.issueId) filters.push(eq(issues.id, opts.issueId));
|
||||
if (afterIssueId) filters.push(gt(issues.id, afterIssueId));
|
||||
|
||||
if (opts?.blockerIssueId) {
|
||||
@@ -6115,6 +6129,7 @@ export function recoveryService(
|
||||
// fallback below. Without it, a terminal issue named in the run context
|
||||
// snapshot would still terminalize the shared run and defeat the guard.
|
||||
runReferencedByActiveIssue?: boolean;
|
||||
onDeferredDeadline?: (at: number) => void;
|
||||
},
|
||||
): Promise<{ terminalized: boolean; status: string }> {
|
||||
// Act only on a run in "running" status. A "queued" run has no process yet,
|
||||
@@ -6130,8 +6145,11 @@ export function recoveryService(
|
||||
// Another controller may own a sandbox run whose PID has no meaning on
|
||||
// this host. Its live lease owns both execution and finalization, even if
|
||||
// the issue is already terminal or this process has no in-memory handle.
|
||||
if (await hasLiveLegacyController(db, run))
|
||||
const controllerDeadline = await liveLegacyControllerExpiresAt(db, run);
|
||||
if (controllerDeadline) {
|
||||
options?.onDeferredDeadline?.(controllerDeadline.getTime());
|
||||
return { terminalized: false, status: run.status };
|
||||
}
|
||||
|
||||
const pid = run.processPid ?? null;
|
||||
const processGroupId = run.processGroupId ?? null;
|
||||
@@ -6352,7 +6370,7 @@ export function recoveryService(
|
||||
// still claims to be live but can no longer reach a terminal status on its
|
||||
// own, so a stuck "running" run can no longer block the sweep. Idempotent and
|
||||
// safe: clears at most one row's worth of lock columns per candidate.
|
||||
async function sweepStaleIssueLocks() {
|
||||
async function sweepStaleIssueLocks(opts?: { companyId?: string; issueId?: string; onDeferredDeadline?: (at: number) => void }) {
|
||||
const result = {
|
||||
cleared: 0,
|
||||
issueIds: [] as string[],
|
||||
@@ -6369,7 +6387,9 @@ export function recoveryService(
|
||||
})
|
||||
.from(issues)
|
||||
.where(
|
||||
sql`(${issues.checkoutRunId} is not null or ${issues.executionRunId} is not null)`,
|
||||
and(sql`(${issues.checkoutRunId} is not null or ${issues.executionRunId} is not null)`,
|
||||
opts?.companyId ? eq(issues.companyId, opts.companyId) : undefined,
|
||||
opts?.issueId ? eq(issues.id, opts.issueId) : undefined),
|
||||
);
|
||||
|
||||
const referencedRunIds = [
|
||||
@@ -6384,7 +6404,8 @@ export function recoveryService(
|
||||
? await db
|
||||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(inArray(heartbeatRuns.id, referencedRunIds))
|
||||
.where(and(inArray(heartbeatRuns.id, referencedRunIds),
|
||||
opts?.companyId ? eq(heartbeatRuns.companyId, opts.companyId) : undefined))
|
||||
: [];
|
||||
const runStatusById = new Map<string, string>();
|
||||
for (const row of runRows) runStatusById.set(row.id, row.status);
|
||||
@@ -6395,8 +6416,15 @@ export function recoveryService(
|
||||
// the same run id in a stale lock column. The terminal reference alone must
|
||||
// not terminalize a run that an active issue still owns, so exclude these
|
||||
// runs from the issue-terminal authority below.
|
||||
const ownershipReferences = opts?.issueId && referencedRunIds.length > 0
|
||||
? await db.select({ companyId: issues.companyId, status: issues.status,
|
||||
checkoutRunId: issues.checkoutRunId, executionRunId: issues.executionRunId }).from(issues).where(and(
|
||||
opts.companyId ? eq(issues.companyId, opts.companyId) : undefined,
|
||||
or(inArray(issues.checkoutRunId, referencedRunIds), inArray(issues.executionRunId, referencedRunIds)),
|
||||
))
|
||||
: candidates;
|
||||
const runIdsReferencedByActiveIssue = new Set<string>();
|
||||
for (const issue of candidates) {
|
||||
for (const issue of ownershipReferences) {
|
||||
if (issue.status === "done" || issue.status === "cancelled") continue;
|
||||
for (const runId of [issue.checkoutRunId, issue.executionRunId]) {
|
||||
if (runId) runIdsReferencedByActiveIssue.add(runId);
|
||||
@@ -6437,6 +6465,7 @@ export function recoveryService(
|
||||
referencingIssueTerminalStatus:
|
||||
issueTerminalStatusByRunId.get(row.id) ?? null,
|
||||
runReferencedByActiveIssue: runIdsReferencedByActiveIssue.has(row.id),
|
||||
onDeferredDeadline: opts?.onDeferredDeadline,
|
||||
});
|
||||
runStatusById.set(row.id, outcome.status);
|
||||
if (outcome.terminalized) result.terminalizedRunIds.push(row.id);
|
||||
@@ -6540,6 +6569,8 @@ export function recoveryService(
|
||||
recordWatchdogDecision,
|
||||
scanSilentActiveRuns,
|
||||
reconcileStrandedAssignedIssues,
|
||||
reconcileUnassignedBlockingIssues,
|
||||
reconcileActiveRecoveryActions,
|
||||
reconcileLegacyContinuation,
|
||||
legacyRepairDispatchBlock,
|
||||
sweepStaleIssueLocks,
|
||||
|
||||
@@ -0,0 +1,113 @@
|
||||
import { registerTaskWatchdogWork } from "../task-watchdog-work.js";
|
||||
import { and, asc, eq, gt, inArray, isNotNull, isNull, or, sql } from "drizzle-orm";
|
||||
import { issueRecoveryActions, issueRelations, issues, type Db } from "@paperclipai/db";
|
||||
import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
|
||||
import { DELIVERY_QUEUES, type DeliveryQueue } from "../delivery-work-notifications.js";
|
||||
import type { IssueRecoveryScope } from "../issue-recovery-notifications.js";
|
||||
import { registerScopedDeadlineWork } from "../scoped-deadline-work.js";
|
||||
|
||||
type Entry = { id: string; companyId: string; dueAt: number };
|
||||
type Scope = { companyId: string; issueId: string; onDeferredDeadline: (at: number) => void };
|
||||
const PAGE_SIZE = 100;
|
||||
|
||||
export function registerIssueRecoveryWork(
|
||||
db: Db,
|
||||
coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
|
||||
owner: {
|
||||
stranded(scope: Scope): Promise<unknown>;
|
||||
orphanBlockers(scope: { companyId: string }): Promise<unknown>;
|
||||
recoveryActions(scope: { companyId: string }): Promise<unknown>;
|
||||
dependencies(scope: Scope): Promise<{ enqueueFailed: number }>;
|
||||
locks(scope: Scope): Promise<unknown>;
|
||||
watchdogs(scope: { companyId: string; watchdogId: string; onDeferredDeadline: (at: number) => void }): Promise<{ incomplete: number }>;
|
||||
schedulingSuppressed(): Promise<boolean>;
|
||||
},
|
||||
canRun: () => boolean,
|
||||
) {
|
||||
function register(queue: DeliveryQueue, statuses: string[] | null, check: (scope: Scope) => Promise<unknown>) {
|
||||
const cursors = new Map<string, string>();
|
||||
let enabled = false;
|
||||
return registerScopedDeadlineWork<Entry, IssueRecoveryScope>(coordinator, {
|
||||
queue,
|
||||
reconstructionIntervalMs: 30_000,
|
||||
beforePass: async () => { enabled = !(await owner.schedulingSuppressed()); },
|
||||
decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as IssueRecoveryScope,
|
||||
matches: (entry, scope) => !scope || (entry.companyId === scope.companyId && (!scope.issueId || entry.id === scope.issueId)),
|
||||
load: async scope => {
|
||||
const key = scope ? JSON.stringify([scope.companyId, scope.issueId]) : "*";
|
||||
if (!enabled) { cursors.delete(key); return []; }
|
||||
const cursor = cursors.get(key);
|
||||
const rows = await db.select({ id: issues.id, companyId: issues.companyId }).from(issues).where(and(
|
||||
scope ? eq(issues.companyId, scope.companyId) : undefined,
|
||||
scope?.issueId ? eq(issues.id, scope.issueId) : undefined,
|
||||
cursor ? gt(issues.id, cursor) : undefined,
|
||||
statuses ? inArray(issues.status, statuses) : or(isNotNull(issues.checkoutRunId), isNotNull(issues.executionRunId)),
|
||||
statuses ? isNull(issues.hiddenAt) : undefined,
|
||||
statuses ? isNull(issues.assigneeUserId) : undefined,
|
||||
statuses ? or(isNotNull(issues.assigneeAgentId), eq(issues.status, "in_review")) : undefined,
|
||||
)).orderBy(asc(issues.id)).limit(PAGE_SIZE);
|
||||
const complete = rows.length < PAGE_SIZE;
|
||||
if (complete) cursors.delete(key); else cursors.set(key, rows.at(-1)!.id);
|
||||
return { rows: rows.map(row => ({ ...row, dueAt: Date.now() })), complete };
|
||||
},
|
||||
check: async entry => {
|
||||
if (!enabled) return null;
|
||||
let next: number | null = null;
|
||||
await check({ companyId: entry.companyId, issueId: entry.id,
|
||||
onDeferredDeadline: at => { next = next === null ? at : Math.min(next, at); } });
|
||||
return next === null ? null : { ...entry, dueAt: next };
|
||||
},
|
||||
}, canRun);
|
||||
}
|
||||
function registerCompany(queue: DeliveryQueue, kind: "blockers" | "actions", check: (scope: { companyId: string }) => Promise<unknown>) {
|
||||
const cursors = new Map<string, string>();
|
||||
let enabled = false;
|
||||
return registerScopedDeadlineWork<Entry, IssueRecoveryScope>(coordinator, {
|
||||
queue,
|
||||
reconstructionIntervalMs: 30_000,
|
||||
beforePass: async () => { enabled = !(await owner.schedulingSuppressed()); },
|
||||
decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as IssueRecoveryScope,
|
||||
matches: (entry, scope) => !scope || entry.companyId === scope.companyId,
|
||||
load: async scope => {
|
||||
const key = scope?.companyId ?? "*";
|
||||
if (!enabled) { cursors.delete(key); return []; }
|
||||
const cursor = cursors.get(key);
|
||||
const rows = kind === "actions"
|
||||
? await db.selectDistinct({ id: issueRecoveryActions.companyId, companyId: issueRecoveryActions.companyId })
|
||||
.from(issueRecoveryActions).where(and(inArray(issueRecoveryActions.status, ["active", "escalated"]),
|
||||
scope ? eq(issueRecoveryActions.companyId, scope.companyId) : undefined,
|
||||
cursor ? gt(issueRecoveryActions.companyId, cursor) : undefined))
|
||||
.orderBy(asc(issueRecoveryActions.companyId)).limit(PAGE_SIZE)
|
||||
: await db.selectDistinct({ id: issues.companyId, companyId: issues.companyId }).from(issues).where(and(
|
||||
inArray(issues.status, ["todo", "blocked"]), isNull(issues.assigneeAgentId), isNull(issues.assigneeUserId),
|
||||
isNotNull(issues.createdByAgentId),
|
||||
scope ? eq(issues.companyId, scope.companyId) : undefined,
|
||||
cursor ? gt(issues.companyId, cursor) : undefined,
|
||||
sql`exists (select 1 from ${issueRelations} r inner join ${issues} dependent
|
||||
on dependent.id = r.related_issue_id and dependent.company_id = r.company_id
|
||||
where r.company_id = ${issues.companyId} and r.issue_id = ${issues.id}
|
||||
and r.type = 'blocks' and dependent.status not in ('done', 'cancelled'))`,
|
||||
)).orderBy(asc(issues.companyId)).limit(PAGE_SIZE);
|
||||
const complete = rows.length < PAGE_SIZE;
|
||||
if (complete) cursors.delete(key); else cursors.set(key, rows.at(-1)!.id);
|
||||
return { rows: rows.map(row => ({ ...row, dueAt: Date.now() })), complete };
|
||||
},
|
||||
check: async entry => { if (enabled) await check({ companyId: entry.companyId }); return null; },
|
||||
}, canRun);
|
||||
}
|
||||
const assignments = register(DELIVERY_QUEUES.strandedAssignments, ["todo", "in_progress", "in_review"], owner.stranded);
|
||||
const orphanBlockers = registerCompany(DELIVERY_QUEUES.orphanBlockers, "blockers", owner.orphanBlockers);
|
||||
const recoveryActions = registerCompany(DELIVERY_QUEUES.activeRecoveryActions, "actions", owner.recoveryActions);
|
||||
const dependencies = register(DELIVERY_QUEUES.resolvedDependencies, ["blocked"], async scope => {
|
||||
const result = await owner.dependencies(scope);
|
||||
if (result.enqueueFailed > 0) throw new Error("Resolved dependency wake remains undelivered");
|
||||
});
|
||||
const locks = register(DELIVERY_QUEUES.staleIssueLocks, null, owner.locks);
|
||||
const watchdogs = registerTaskWatchdogWork(db, coordinator, {
|
||||
reconcile: owner.watchdogs, enabled: async () => !(await owner.schedulingSuppressed()),
|
||||
}, canRun);
|
||||
const workers = [assignments, orphanBlockers, recoveryActions, dependencies, locks, watchdogs];
|
||||
return { ready: Promise.all(workers.map(worker => worker.ready)),
|
||||
wake() { for (const worker of workers) worker.invalidate(); },
|
||||
async stop() { await Promise.all(workers.map(worker => worker.stop())); } };
|
||||
}
|
||||
@@ -1,8 +1,10 @@
|
||||
import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
|
||||
export type RunDispatchScope = { companyId: string };
|
||||
|
||||
export async function notifyRunDispatch(transaction: object, scope?: RunDispatchScope) {
|
||||
await notifyIssueRecovery(transaction, scope ? { companyId: scope.companyId } : undefined);
|
||||
const hint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*";
|
||||
await notifyDeliveryWork(transaction, DELIVERY_QUEUES.queuedRunDispatch, hint);
|
||||
}
|
||||
|
||||
@@ -13,6 +13,8 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
|
||||
batchSize?: number;
|
||||
minimumPassIntervalMs?: number;
|
||||
concurrentChecks?: { limit: number; key(entry: Entry): string };
|
||||
reconstructionIntervalMs?: number;
|
||||
beforePass?(): Promise<void>;
|
||||
decodeScope(hint: string): Scope | undefined;
|
||||
matches(entry: Entry, scope?: Scope): boolean;
|
||||
load(scope?: Scope, signal?: AbortSignal): Promise<Entry[] | { rows: Entry[] | null; complete: boolean }>;
|
||||
@@ -29,6 +31,7 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
|
||||
const entries = new Map<string, Entry>();
|
||||
const failures = new Map<string, { count: number; retryAt: number }>();
|
||||
const scopeFailures = new Map<string, { count: number; retryAt: number }>();
|
||||
const reconstructAfter = new Map<string, number>();
|
||||
let earliest = 0;
|
||||
let lastCompany: string | null = null;
|
||||
function failed(map: typeof failures, id: string) {
|
||||
@@ -38,13 +41,14 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
|
||||
function dirtyScopes() {
|
||||
return [...dirty].filter(hint => hint !== "*" || !scopeFailures.has(hint))
|
||||
.filter(hint => !scans.has(hint) || scopeFailures.has(hint))
|
||||
.filter(hint => (reconstructAfter.get(hint) ?? 0) <= Date.now() || scopeFailures.has(hint))
|
||||
.map(hint => { try { return options.decodeScope(hint); } catch { return undefined; } });
|
||||
}
|
||||
function next() {
|
||||
if (!canRun()) return dirty.size > 0 || entries.size > 0 ? Date.now() + 30_000 : null;
|
||||
let nextAt: number | null = null;
|
||||
const take = (at: number) => { nextAt = nextAt === null ? at : Math.min(nextAt, at); };
|
||||
for (const hint of dirty) take(Math.max(earliest, scopeFailures.get(hint)?.retryAt ?? 0));
|
||||
for (const hint of dirty) take(Math.max(earliest, reconstructAfter.get(hint) ?? 0, scopeFailures.get(hint)?.retryAt ?? 0));
|
||||
const unavailable = dirtyScopes();
|
||||
for (const entry of entries.values()) if (!unavailable.some(scope => options.matches(entry, scope)))
|
||||
take(Math.max(earliest, entry.dueAt, failures.get(entry.id)?.retryAt ?? 0));
|
||||
@@ -58,9 +62,18 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
|
||||
} },
|
||||
run: async signal => {
|
||||
if (!canRun() || Date.now() < earliest) return;
|
||||
if ((options.reconstructionIntervalMs ?? 0) > 0) for (const [hint, until] of reconstructAfter) {
|
||||
if (until <= Date.now()) reconstructAfter.delete(hint);
|
||||
}
|
||||
const ready = [...dirty].filter(hint => Math.max(reconstructAfter.get(hint) ?? 0, scopeFailures.get(hint)?.retryAt ?? 0) <= Date.now());
|
||||
if (ready.length === 0 && ![...entries.values()].some(entry => Math.max(entry.dueAt, failures.get(entry.id)?.retryAt ?? 0) <= Date.now())) return;
|
||||
earliest = Date.now() + (options.minimumPassIntervalMs ?? 1000);
|
||||
try { await options.beforePass?.(); } catch (error) {
|
||||
for (const hint of ready) failed(scopeFailures, hint);
|
||||
for (const entry of entries.values()) if (entry.dueAt <= Date.now()) failed(failures, entry.id);
|
||||
throw error;
|
||||
}
|
||||
const errors: unknown[] = [];
|
||||
const ready = [...dirty].filter(hint => (scopeFailures.get(hint)?.retryAt ?? 0) <= Date.now());
|
||||
const requested = ready.includes("*")
|
||||
? ["*", ...ready.filter(hint => hint !== "*")].slice(0, 25)
|
||||
: ready.slice(0, 25);
|
||||
@@ -85,6 +98,9 @@ export function registerScopedDeadlineWork<Entry extends ScheduledEntry, Scope>(
|
||||
entries.delete(entry.id); failures.delete(entry.id);
|
||||
}
|
||||
scans.delete(hint);
|
||||
if ((options.reconstructionIntervalMs ?? 0) > 0) {
|
||||
reconstructAfter.set(hint, Date.now() + options.reconstructionIntervalMs!);
|
||||
}
|
||||
if ((versions.get(hint) ?? 0) !== scan.version) dirty.add(hint); else versions.delete(hint);
|
||||
if (!scope) scopeFailures.clear(); else scopeFailures.delete(hint);
|
||||
} catch (error) {
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { heartbeatRuns, issueComments, issueRecoveryActions, type Db } from "@paperclipai/db";
|
||||
import { writeExecutionControl } from "./execution-control-notifications.js";
|
||||
@@ -11,6 +12,7 @@ export async function adoptSystemRecovery(db: Db, source: WorkspaceRecoverySourc
|
||||
.where(and(eq(issueRecoveryActions.companyId, source.companyId), eq(issueRecoveryActions.id, source.actionId),
|
||||
eq(issueRecoveryActions.sourceIssueId, source.issueId))).for("update");
|
||||
if (!existing || existing.evidence.systemRecovery) return;
|
||||
await notifyIssueRecovery(tx, { companyId: source.companyId });
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.workspaceRecovery);
|
||||
const state = createSystemRecovery(source.runId);
|
||||
// Atomic, once per source. An exhausted previous execution budget is history;
|
||||
|
||||
@@ -1,16 +0,0 @@
|
||||
export function createIncompleteTaskWatchdogScanReporter(
|
||||
warn: (details: { incomplete: number; newIncompleteIssueIds: string[] }, message: string) => void,
|
||||
) {
|
||||
let previousIncompleteIssueIds = new Set<string>();
|
||||
return (result: { incomplete: number; incompleteIssueIds: string[] }, phase: "startup" | "periodic") => {
|
||||
const currentIssueIds = new Set(result.incompleteIssueIds);
|
||||
const newIncompleteIssueIds = [...currentIssueIds].filter((id) => !previousIncompleteIssueIds.has(id));
|
||||
previousIncompleteIssueIds = currentIssueIds;
|
||||
if (newIncompleteIssueIds.length === 0) return;
|
||||
warn(
|
||||
{ incomplete: result.incomplete, newIncompleteIssueIds },
|
||||
`${phase} task-watchdog reconciliation found incomplete subtree scans`,
|
||||
);
|
||||
};
|
||||
}
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
import { and, asc, eq, gt } from "drizzle-orm";
|
||||
import { issueWatchdogs, type Db } from "@paperclipai/db";
|
||||
import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
|
||||
import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
|
||||
import type { IssueRecoveryScope } from "./issue-recovery-notifications.js";
|
||||
import { registerScopedDeadlineWork } from "./scoped-deadline-work.js";
|
||||
|
||||
type Entry = { id: string; companyId: string; dueAt: number };
|
||||
|
||||
export function registerTaskWatchdogWork(
|
||||
db: Db,
|
||||
coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
|
||||
owner: {
|
||||
enabled(): Promise<boolean>;
|
||||
reconcile(scope: { companyId: string; watchdogId: string; onDeferredDeadline: (at: number) => void }): Promise<{ incomplete: number }>;
|
||||
},
|
||||
canRun: () => boolean,
|
||||
) {
|
||||
const cursors = new Map<string, string>();
|
||||
let enabled = false;
|
||||
return registerScopedDeadlineWork<Entry, IssueRecoveryScope>(coordinator, {
|
||||
queue: DELIVERY_QUEUES.taskWatchdogs,
|
||||
reconstructionIntervalMs: 30_000,
|
||||
beforePass: async () => { enabled = await owner.enabled(); },
|
||||
decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as IssueRecoveryScope,
|
||||
matches: (entry, scope) => !scope || entry.companyId === scope.companyId,
|
||||
load: async scope => {
|
||||
const key = scope?.companyId ?? "*";
|
||||
if (!enabled) { cursors.delete(key); return []; }
|
||||
const cursor = cursors.get(key);
|
||||
const rows = await db.select({ id: issueWatchdogs.id, companyId: issueWatchdogs.companyId })
|
||||
.from(issueWatchdogs).where(and(eq(issueWatchdogs.status, "active"),
|
||||
scope ? eq(issueWatchdogs.companyId, scope.companyId) : undefined,
|
||||
cursor ? gt(issueWatchdogs.id, cursor) : undefined))
|
||||
.orderBy(asc(issueWatchdogs.id)).limit(100);
|
||||
const complete = rows.length < 100;
|
||||
if (complete) cursors.delete(key); else cursors.set(key, rows.at(-1)!.id);
|
||||
return { rows: rows.map(row => ({ ...row, dueAt: Date.now() })), complete };
|
||||
},
|
||||
check: async entry => {
|
||||
if (!enabled) return null;
|
||||
let next: number | null = null;
|
||||
const result = await owner.reconcile({ companyId: entry.companyId, watchdogId: entry.id,
|
||||
onDeferredDeadline: at => { next = next === null ? at : Math.min(next, at); } });
|
||||
if (result.incomplete > 0) throw new Error(`Task watchdog ${entry.id} in company ${entry.companyId} has an incomplete subtree`);
|
||||
return next === null ? null : { ...entry, dueAt: next };
|
||||
},
|
||||
}, canRun);
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyIssueRecovery } from "./issue-recovery-notifications.js";
|
||||
import { createHash } from "node:crypto";
|
||||
import { and, asc, eq, gte, inArray, isNull, or, sql } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
@@ -831,7 +832,7 @@ async function updateIssueWatchdogRow(
|
||||
return updated;
|
||||
}
|
||||
|
||||
export async function upsertIssueWatchdogForIssue(
|
||||
async function upsertIssueWatchdogInTransaction(
|
||||
dbOrTx: any,
|
||||
companyId: string,
|
||||
issueId: string,
|
||||
@@ -847,12 +848,13 @@ export async function upsertIssueWatchdogForIssue(
|
||||
.where(and(eq(issueWatchdogs.companyId, companyId), eq(issueWatchdogs.issueId, issueId)))
|
||||
.then((rows: IssueWatchdogRow[]) => rows[0] ?? null);
|
||||
|
||||
await notifyIssueRecovery(dbOrTx, { companyId });
|
||||
if (existing) {
|
||||
const updated = await updateIssueWatchdogRow(dbOrTx, existing, input, now);
|
||||
return { watchdog: toIssueWatchdog(updated), created: false };
|
||||
}
|
||||
|
||||
const insertResult: { row: IssueWatchdogRow; created: boolean } = await dbOrTx
|
||||
const insertResult: { row: IssueWatchdogRow; created: boolean } = await dbOrTx.transaction(async (tx: Db) => tx
|
||||
.insert(issueWatchdogs)
|
||||
.values({
|
||||
companyId,
|
||||
@@ -870,7 +872,7 @@ export async function upsertIssueWatchdogForIssue(
|
||||
updatedAt: now,
|
||||
})
|
||||
.returning()
|
||||
.then((rows: IssueWatchdogRow[]) => ({ row: rows[0], created: true }))
|
||||
.then((rows: IssueWatchdogRow[]) => ({ row: rows[0], created: true })))
|
||||
.catch(async (error: unknown) => {
|
||||
if (!isIssueWatchdogUniqueConflict(error)) throw error;
|
||||
const winner = await dbOrTx
|
||||
@@ -885,6 +887,15 @@ export async function upsertIssueWatchdogForIssue(
|
||||
return { watchdog: toIssueWatchdog(insertResult.row), created: insertResult.created };
|
||||
}
|
||||
|
||||
export function upsertIssueWatchdogForIssue(
|
||||
dbOrTx: Db | Parameters<Parameters<Db["transaction"]>[0]>[0],
|
||||
companyId: string,
|
||||
issueId: string,
|
||||
input: IssueWatchdogUpsertInput,
|
||||
): Promise<{ watchdog: IssueWatchdog; created: boolean }> {
|
||||
return dbOrTx.transaction(tx => upsertIssueWatchdogInTransaction(tx, companyId, issueId, input));
|
||||
}
|
||||
|
||||
export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {}) {
|
||||
const issuesSvc = issueService(db);
|
||||
|
||||
@@ -1478,7 +1489,7 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {})
|
||||
return created;
|
||||
}
|
||||
|
||||
async function evaluateWatchdog(row: IssueWatchdogRow, opts: { runId?: string | null } = {}) {
|
||||
async function evaluateWatchdog(row: IssueWatchdogRow, opts: { runId?: string | null; onDeferredDeadline?: (at: number) => void } = {}) {
|
||||
const watchdog = await markTerminalWatchdogIssueReviewed(row, opts);
|
||||
const sourceIssue = await db
|
||||
.select()
|
||||
@@ -1491,6 +1502,13 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {})
|
||||
|
||||
const input = await collectClassifierInput(watchdog.companyId, watchdog);
|
||||
const classification = classifyTaskWatchdogSubtree(input);
|
||||
if (classification.state === "pending_first_run") {
|
||||
const pending = new Set(classification.pendingIssueIds);
|
||||
for (const issue of input.issues) if (pending.has(issue.id)) {
|
||||
const createdAt = toEpochMs(issue.createdAt);
|
||||
if (createdAt !== null) opts.onDeferredDeadline?.(createdAt + TASK_WATCHDOG_FIRST_RUN_GRACE_MS);
|
||||
}
|
||||
}
|
||||
if (classification.state !== "stopped") {
|
||||
return { state: classification.state, reason: classification.reason, classification };
|
||||
}
|
||||
@@ -1610,13 +1628,14 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {})
|
||||
};
|
||||
}
|
||||
|
||||
async function listActiveWatchdogsForCompany(companyId?: string | null) {
|
||||
async function listActiveWatchdogsForCompany(companyId?: string | null, watchdogId?: string) {
|
||||
return db
|
||||
.select()
|
||||
.from(issueWatchdogs)
|
||||
.where(and(
|
||||
eq(issueWatchdogs.status, "active"),
|
||||
...(companyId ? [eq(issueWatchdogs.companyId, companyId)] : []),
|
||||
watchdogId ? eq(issueWatchdogs.id, watchdogId) : undefined,
|
||||
));
|
||||
}
|
||||
|
||||
@@ -1746,14 +1765,16 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {})
|
||||
issueId: string,
|
||||
actor: ActorFields = {},
|
||||
): Promise<IssueWatchdog | null> => {
|
||||
await assertWatchedIssue(db, companyId, issueId);
|
||||
const existing = await db
|
||||
return db.transaction(async tx => {
|
||||
await assertWatchedIssue(tx, companyId, issueId);
|
||||
const existing = await tx
|
||||
.select()
|
||||
.from(issueWatchdogs)
|
||||
.where(and(eq(issueWatchdogs.companyId, companyId), eq(issueWatchdogs.issueId, issueId)))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!existing || existing.status === "disabled") return null;
|
||||
const [updated] = await db
|
||||
await notifyIssueRecovery(tx, { companyId });
|
||||
const [updated] = await tx
|
||||
.update(issueWatchdogs)
|
||||
.set({
|
||||
status: "disabled",
|
||||
@@ -1765,14 +1786,17 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {})
|
||||
.where(eq(issueWatchdogs.id, existing.id))
|
||||
.returning();
|
||||
return toIssueWatchdog(updated);
|
||||
});
|
||||
},
|
||||
|
||||
reconcileTaskWatchdogs: async (opts: {
|
||||
companyId?: string | null;
|
||||
runId?: string | null;
|
||||
watchdogId?: string;
|
||||
onDeferredDeadline?: (at: number) => void;
|
||||
issueCreatedAtGte?: Date | null;
|
||||
} = {}) => {
|
||||
let rows = await listActiveWatchdogsForCompany(opts.companyId ?? null);
|
||||
let rows = await listActiveWatchdogsForCompany(opts.companyId ?? null, opts.watchdogId);
|
||||
if (opts.issueCreatedAtGte) {
|
||||
const watchdogIssueIds = [...new Set(rows.map((row) => row.issueId))];
|
||||
const eligibleIssueIds = new Set(
|
||||
@@ -1802,7 +1826,7 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {})
|
||||
};
|
||||
for (const row of rows) {
|
||||
result.checked += 1;
|
||||
const evaluated = await evaluateWatchdog(row, { runId: opts.runId ?? null });
|
||||
const evaluated = await evaluateWatchdog(row, { runId: opts.runId ?? null, onDeferredDeadline: opts.onDeferredDeadline });
|
||||
if (evaluated.state === "triggered") {
|
||||
result.triggered += 1;
|
||||
result.watchdogIssueIds.push(evaluated.watchdogIssueId);
|
||||
|
||||
Reference in new issue
Block a user