mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-02 16:58:52 +01:00
Merge branch 'codex/sleep-stack-03' into codex/sleep-stack-05
* codex/sleep-stack-03: fix(ui): invalidate pending startup work on sign-out fix(server): park persisted deferred admission refusals test(server): preserve attention feed beyond blocker root budget
This commit is contained in:
commit
4a2155d9ef
11 files changed
+312
-26
No files matched your search
@@ -318,6 +318,25 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000;
|
||||
}
|
||||
});
|
||||
|
||||
it("keeps deferred roots visible in the ordinary attention feed above the root cap", async () => {
|
||||
await db.execute(sql`insert into issues (id, company_id, title, status, updated_at)
|
||||
select gen_random_uuid(), ${companyId}::uuid, 'overflow-root-' || n, 'blocked', ${new Date().toISOString()}::timestamptz
|
||||
from generate_series(1, ${BLOCKER_ATTENTION_MAX_NODES + 1}) n`);
|
||||
const roots = await db.select().from(issues).where(eq(issues.companyId, companyId));
|
||||
const batch = await issueService(db).listBlockerAttentionBatch(companyId, roots);
|
||||
expect(batch.deferredRootIds.size).toBeGreaterThan(0);
|
||||
expect(batch.nodeCount).toBeLessThanOrEqual(BLOCKER_ATTENTION_MAX_NODES);
|
||||
for (const rootId of batch.deferredRootIds) expect(batch.attention.get(rootId)).toMatchObject({
|
||||
state: "needs_attention", reason: "attention_required", terminalBlockerIssueId: rootId,
|
||||
});
|
||||
const listed = await issueService(db).list(companyId, { status: "blocked", includeBlockedBy: true });
|
||||
expect(listed).toHaveLength(roots.length);
|
||||
for (const root of listed) expect(root.blockerAttention).toMatchObject({ state: "needs_attention", terminalBlockerIssueId: root.id });
|
||||
const feed = await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true });
|
||||
const visible = new Set(feed.items.filter(item => item.sourceKind === "blocker_attention").map(item => item.subject.id));
|
||||
expect(visible).toEqual(new Set(roots.map(root => root.id)));
|
||||
});
|
||||
|
||||
it("bounds every graph query when independent root trees contain over 65535 unfinished nodes", async () => {
|
||||
await db.execute(sql`insert into issues (id, company_id, title, status, updated_at)
|
||||
select gen_random_uuid(), ${companyId}::uuid, 'parameter-root-' || n, 'blocked', ${old().toISOString()}::timestamptz
|
||||
|
||||
@@ -1,7 +1,11 @@
|
||||
import { conflict, HttpError } from "../errors.js";
|
||||
import { setImmediate as immediate } from "node:timers/promises";
|
||||
import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js";
|
||||
import { instanceSettingsService } from "../services/instance-settings.js";
|
||||
import { budgetService } from "../services/budgets.js";
|
||||
import { configureAgentLifecycle, createAgentLifecycle } from "../services/agent-lifecycle.js";
|
||||
import { accessService } from "../services/access.js";
|
||||
import { writeDeferredExecutionWait } from "../services/deferred-execution-wait.js";
|
||||
import { PersistedExecutionWaitRefusal, writeDeferredExecutionWait } from "../services/deferred-execution-wait.js";
|
||||
import { createQueuedCommentIssueLockWriter } from "../modules/wake-queue/adapters/queued-comment-postgres.js";
|
||||
import { createHeartbeatQueue } from "../services/heartbeat/queue.js";
|
||||
import { createHeartbeatQueueTestDependencies } from "./helpers/heartbeat-queue.js";
|
||||
@@ -9,7 +13,7 @@ import { issueTreeControlService } from "../services/issue-tree-control.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, agentWakeupRequests, authUsers, chatActions, chatEndpoints, toolApplications, toolConnections, companies, companyMemberships, createDb, reconcileDatabaseWork, heartbeatRuns, issueComments, issues } from "@paperclipai/db";
|
||||
import { agents, agentWakeupRequests, authUsers, chatActions, chatEndpoints, toolApplications, toolConnections, companies, companyMemberships, costEvents, createDb, reconcileDatabaseWork, heartbeatRuns, issueComments, issues } from "@paperclipai/db";
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
|
||||
@@ -28,18 +32,23 @@ describe("deferred receipt durable dispatch", () => {
|
||||
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator>;
|
||||
let drain = false;
|
||||
const errors: unknown[] = [];
|
||||
const errorTimes: number[] = [];
|
||||
const statements: string[] = [];
|
||||
const passes = new Set<Promise<unknown>>();
|
||||
const lifecycleWorkers = new Set<ReturnType<typeof configureAgentLifecycle>>();
|
||||
beforeAll(async () => {
|
||||
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_DEFERRED_RECEIPT_TEST_DATABASE_URL");
|
||||
if (!external) temporary = await startEmbeddedPostgresTestDatabase("deferred-receipts-");
|
||||
db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
|
||||
db.$client.options.onnotice = () => {};
|
||||
db.$client.options.debug = (_connection, query) => { statements.push(query); };
|
||||
await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`)));
|
||||
});
|
||||
beforeEach(async () => {
|
||||
await db.execute(sql`truncate companies cascade`); drain = false; errors.length = 0;
|
||||
await db.execute(sql`truncate companies cascade`); drain = false; errors.length = 0; errorTimes.length = 0;
|
||||
heartbeat = heartbeatService(db);
|
||||
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !drain, canReconcile: () => true,
|
||||
onError: error => errors.push(error) });
|
||||
onError: error => { errors.push(error); errorTimes.push(Date.now()); } });
|
||||
const register = coordinator.register.bind(coordinator);
|
||||
coordinator.register = (queue, task) => register(queue, { ...task, run: signal => {
|
||||
const pass = Promise.resolve(task.run(signal)); passes.add(pass);
|
||||
@@ -48,6 +57,7 @@ describe("deferred receipt durable dispatch", () => {
|
||||
});
|
||||
afterEach(async () => {
|
||||
await coordinator.stop(); await heartbeat.drainActiveRunExecutions();
|
||||
await Promise.all([...lifecycleWorkers].map(worker => worker.stop())); lifecycleWorkers.clear();
|
||||
vi.useRealTimers(); vi.restoreAllMocks(); expect(errors).toEqual([]);
|
||||
});
|
||||
afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
|
||||
@@ -72,7 +82,9 @@ describe("deferred receipt durable dispatch", () => {
|
||||
const reads = vi.spyOn(db, "select"), executes = vi.spyOn(db, "execute");
|
||||
const selected = reads.mock.calls.length, executed = executes.mock.calls.length;
|
||||
if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
|
||||
await vi.advanceTimersByTimeAsync(20 * 60_000);
|
||||
const count = statements.length;
|
||||
await vi.advanceTimersByTimeAsync(25 * 60_000);
|
||||
expect(statements).toHaveLength(count);
|
||||
expect(reads).toHaveBeenCalledTimes(selected); expect(executes).toHaveBeenCalledTimes(executed);
|
||||
}
|
||||
async function finish(issueId: string) {
|
||||
@@ -130,6 +142,156 @@ describe("deferred receipt durable dispatch", () => {
|
||||
expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${receipt.requestId}`)).toHaveLength(1);
|
||||
await quiet();
|
||||
});
|
||||
async function blockedRetry(gate: "budget" | "paused agent", clocked = false) {
|
||||
const f = await fixture(), lifecycle = createAgentLifecycle(db), budgets = budgetService(db);
|
||||
// Host lifecycle completion is controlled; all state, admission and the
|
||||
// eventual process adapter execution use their production implementations.
|
||||
const lifecycleWorker = configureAgentLifecycle(db, { requiredPluginIds: async () => [],
|
||||
runHost: async () => "complete", runPlugin: async () => "complete" });
|
||||
lifecycleWorkers.add(lifecycleWorker);
|
||||
const policy = { scopeType: "agent" as const, scopeId: f.agent.id, amount: 100,
|
||||
metric: "billed_cents" as const, windowKind: "lifetime" as const, hardStopEnabled: true, notifyEnabled: false };
|
||||
if (gate === "budget") {
|
||||
await db.insert(costEvents).values({ companyId: f.companyId, agentId: f.agent.id,
|
||||
provider: "test", model: "receipt-proof", costCents: 200, occurredAt: new Date() });
|
||||
await budgets.upsertPolicy(f.companyId, policy, f.actorId);
|
||||
} else await lifecycle.pauseAgent(f.agent.id);
|
||||
await lifecycleWorker.process(f.agent.id);
|
||||
if (clocked) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
|
||||
await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready;
|
||||
await heartbeat.registerRunDispatchWork(coordinator, () => !drain).ready;
|
||||
await heartbeat.registerScheduledRetryCancellationWork(coordinator, () => !drain).ready;
|
||||
const request = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issue.id, actorId: f.actorId,
|
||||
observedRunId: f.source.id, idempotencyKey: randomUUID() });
|
||||
expect(request.accepted).toBe(true);
|
||||
const read = async () => (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!)))[0];
|
||||
await vi.waitFor(async () => expect((await read()).payload?.executionWait).toMatchObject({
|
||||
reason: gate === "budget" ? "budget.blocked" : "agent.not_invokable",
|
||||
}), { timeout: 10_000 });
|
||||
const refusals: unknown[] = [];
|
||||
await heartbeat.resumeExecutionWaitComments({ companyId: f.companyId, requestId: request.requestId!, onError: error => refusals.push(error) });
|
||||
expect(refusals).toHaveLength(1);
|
||||
expect(refusals[0]).toBeInstanceOf(HttpError);
|
||||
expect(refusals[0]).toMatchObject({ status: 409,
|
||||
details: gate === "budget" ? { scopeType: "agent", scopeId: f.agent.id } : { status: "paused", reason: "paused" } });
|
||||
return { f, lifecycle, budgets, policy, lifecycleWorker, request, read };
|
||||
}
|
||||
|
||||
it.each(["budget", "paused agent"] as const)("parks real task Retry behind unchanged %s without recurring SQL", async gate => {
|
||||
const { request, read } = await blockedRetry(gate, true);
|
||||
// The clock was installed before worker registration and every timer.
|
||||
// No execution exists while we observe actual driver SQL across25min.
|
||||
for (let n = 0; n < 3; n++) {
|
||||
await vi.advanceTimersByTimeAsync(31_000); await Promise.allSettled([...passes]); await immediate();
|
||||
}
|
||||
expect(errors).toEqual([]);
|
||||
await quiet();
|
||||
expect(await read()).toMatchObject({ status: "deferred_issue_execution", runId: null });
|
||||
expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${request.requestId}`)).toHaveLength(0);
|
||||
});
|
||||
|
||||
it.each(["budget", "paused agent"] as const)("releases real %s Retry into exactly one saved process execution", async gate => {
|
||||
const { f, lifecycle, budgets, policy, lifecycleWorker, request, read } = await blockedRetry(gate);
|
||||
// This independent warm-worker fixture stays on the real PostgreSQL clock
|
||||
// through gate release, controller authority and actual child-process exit.
|
||||
if (gate === "budget") await budgets.upsertPolicy(f.companyId, { ...policy, amount: 1000 }, f.actorId);
|
||||
else await lifecycle.resumeAgent(f.agent.id);
|
||||
await lifecycleWorker.process(f.agent.id);
|
||||
let runId: string | null = null;
|
||||
await vi.waitFor(async () => {
|
||||
const wake = await read(); expect(wake.runId).toBeTruthy(); runId = wake.runId;
|
||||
}, { timeout: 40_000 });
|
||||
await vi.waitFor(async () => {
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId!));
|
||||
expect(run.status, JSON.stringify({ error: run.error, code: run.errorCode })).toBe("succeeded");
|
||||
expect(run.stdoutExcerpt).toContain("receipt-proof");
|
||||
}, { timeout: 15_000 });
|
||||
expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.contextSnapshot}->>'taskRetryRequestId' = ${request.requestId}`)).toHaveLength(1);
|
||||
await finish(f.issue.id); await quiet();
|
||||
});
|
||||
|
||||
it.each(["missing", "cancelled"] as const)("preserves ordinary409 when a %s receipt cannot save the gate", async status => {
|
||||
const f = await fixture(), requestId = randomUUID();
|
||||
if (status === "cancelled") await db.insert(agentWakeupRequests).values({ id: requestId, companyId: f.companyId,
|
||||
agentId: f.agent.id, source: "on_demand", status, payload: { issueId: f.issue.id } });
|
||||
const deps = createHeartbeatQueueTestDependencies(db), queue = createHeartbeatQueue(db, deps);
|
||||
deps.getAgent.mockResolvedValue(f.agent);
|
||||
deps.instanceSettings.getExperimental.mockImplementation(() => instanceSettingsService(db).getExperimental());
|
||||
for (const gate of ["budget", "agent"] as const) {
|
||||
deps.budgets.getInvocationBlock.mockResolvedValue(gate === "budget"
|
||||
? { scopeType: "agent", scopeId: f.agent.id, scopeName: f.agent.name, reason: "Budget blocked" } : null);
|
||||
deps.getAgentInvokability.mockResolvedValue({ invokable: false, message: "Paused", reason: "paused", invalidOrgChain: false });
|
||||
const error = await queue.enqueueWakeup(f.agent.id, { source: "on_demand", reason: "task_retry",
|
||||
requestedByActorType: "user", requestedByActorId: f.actorId, payload: { issueId: f.issue.id },
|
||||
contextSnapshot: { issueId: f.issue.id, responsibleUserId: f.actorId } }, requestId).catch(cause => cause);
|
||||
expect(error).toBeInstanceOf(HttpError);
|
||||
expect(error).not.toBeInstanceOf(PersistedExecutionWaitRefusal);
|
||||
expect(error).toMatchObject({ status: 409, message: gate === "budget" ? "Budget blocked" : "Paused" });
|
||||
}
|
||||
const rows = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, requestId));
|
||||
expect(rows).toHaveLength(status === "missing" ? 0 : 1);
|
||||
if (status === "cancelled") expect(rows[0]).toMatchObject({ status: "cancelled", runId: null, payload: { issueId: f.issue.id } });
|
||||
});
|
||||
|
||||
it("retains failure backoff for a real wait-persistence database error", async () => {
|
||||
const f = await fixture();
|
||||
await createAgentLifecycle(db).pauseAgent(f.agent.id);
|
||||
const request = await requestTaskRetry({ db, companyId: f.companyId, issueId: f.issue.id, actorId: f.actorId,
|
||||
observedRunId: f.source.id, idempotencyKey: randomUUID() });
|
||||
await db.execute(sql`create function c1_refusal_fault() returns trigger language plpgsql as $$
|
||||
begin if new.reason = 'task_retry' and new.payload ? 'executionWait' then
|
||||
raise exception 'C1 injected wait persistence failure'; end if; return new; end $$`);
|
||||
await db.execute(sql`create trigger c1_refusal_fault before update on agent_wakeup_requests
|
||||
for each row execute function c1_refusal_fault()`);
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
|
||||
try {
|
||||
await heartbeat.registerDeferredReceiptWork(coordinator, () => !drain).ready;
|
||||
await vi.waitFor(() => expect(errors).toHaveLength(1));
|
||||
const failure = errors[0] as AggregateError;
|
||||
expect(failure).toBeInstanceOf(AggregateError);
|
||||
expect(String(failure.errors[0]?.errors[0]?.cause?.message)).toContain("C1 injected wait persistence failure");
|
||||
const due = errorTimes[0] + 30_000;
|
||||
await Promise.allSettled([...passes]); await immediate();
|
||||
await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBe(due));
|
||||
expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!)))[0].payload?.executionWait).toBeUndefined();
|
||||
await vi.advanceTimersByTimeAsync(29_999);
|
||||
expect(errors).toHaveLength(1);
|
||||
await db.execute(sql`drop trigger c1_refusal_fault on agent_wakeup_requests`);
|
||||
await db.execute(sql`drop function c1_refusal_fault()`);
|
||||
errors.length = 0;
|
||||
for (let n = 0; n < 3; n++) {
|
||||
await vi.advanceTimersByTimeAsync(31_000); await Promise.allSettled([...passes]); await immediate();
|
||||
}
|
||||
expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, request.requestId!)))[0].payload?.executionWait)
|
||||
.toMatchObject({ reason: "agent.not_invokable" });
|
||||
await quiet();
|
||||
} finally {
|
||||
await db.execute(sql`drop trigger if exists c1_refusal_fault on agent_wakeup_requests`);
|
||||
await db.execute(sql`drop function if exists c1_refusal_fault()`);
|
||||
}
|
||||
});
|
||||
|
||||
it("does not park an unrelated HTTP409 as a persisted refusal", async () => {
|
||||
const f = await fixture();
|
||||
const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id,
|
||||
source: "on_demand", reason: "task_retry", status: "deferred_issue_execution" }).returning();
|
||||
const failure = conflict("Unexpected admission conflict");
|
||||
let fail = true;
|
||||
const resume = vi.fn(async (scope: { onError?: (error: unknown) => void }) => { if (fail) scope.onError?.(failure); });
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
|
||||
await registerDeferredReceiptWork(db, coordinator, { resume, settleTaskChat: async () => {}, dispatchNative: async () => {},
|
||||
schedulingSuppressed: async () => false }, () => !drain).ready;
|
||||
expect(errors).toHaveLength(1);
|
||||
expect((errors[0] as AggregateError).errors[0].errors).toEqual([failure]);
|
||||
expect(coordinator.nextWakeAt()).toBe(Date.now() + 30_000);
|
||||
expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)))[0].status).toBe("deferred_issue_execution");
|
||||
errors.length = 0; fail = false;
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
await vi.waitFor(() => expect(resume).toHaveBeenCalledTimes(2));
|
||||
await quiet();
|
||||
});
|
||||
|
||||
it.each(["native claim", "execution wait"])("reconstructs an exact fresh %s deadline without probing early", async kind => {
|
||||
const f = await fixture(), updatedAt = new Date(), native = kind === "native claim";
|
||||
const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agent.id,
|
||||
|
||||
@@ -1,24 +1,33 @@
|
||||
import { HttpError } from "../errors.js";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { agentWakeupRequests, type Db } from "@paperclipai/db";
|
||||
import { notifyDeferredExecutionWork } from "./deferred-execution-notifications.js";
|
||||
|
||||
/** Admission has saved this refusal on the caller's durable deferred receipt.
|
||||
* HTTP callers still receive the existing 409; receipt recovery waits for its
|
||||
* owning gate producer instead of treating the refusal as infrastructure debt. */
|
||||
export class PersistedExecutionWaitRefusal extends HttpError {
|
||||
constructor(message: string, details?: unknown) { super(409, message, details); }
|
||||
}
|
||||
|
||||
export async function writeDeferredExecutionWait(db: Db, scope: { companyId: string; requestId: string },
|
||||
condition: Record<string, unknown>, options: { replace?: boolean } = {}) {
|
||||
return db.transaction(async tx => {
|
||||
const [current] = await tx.select({ payload: agentWakeupRequests.payload }).from(agentWakeupRequests)
|
||||
.where(and(eq(agentWakeupRequests.companyId, scope.companyId), eq(agentWakeupRequests.id, scope.requestId),
|
||||
eq(agentWakeupRequests.status, "deferred_issue_execution"))).for("update");
|
||||
if (!current) return;
|
||||
if (!current) return false;
|
||||
const previous = current.payload?.executionWait;
|
||||
const previousReason = previous && typeof previous === "object" && !Array.isArray(previous)
|
||||
? (previous as Record<string, unknown>).reason : undefined;
|
||||
if (!previous || previousReason !== condition.reason) await notifyDeferredExecutionWork(tx, scope);
|
||||
await tx.update(agentWakeupRequests).set({
|
||||
const [saved] = await tx.update(agentWakeupRequests).set({
|
||||
payload: sql`jsonb_set(coalesce(${agentWakeupRequests.payload}, '{}'::jsonb), '{executionWait}',
|
||||
${options.replace ? sql`'{}'::jsonb` : sql`coalesce(${agentWakeupRequests.payload}->'executionWait', '{}'::jsonb)`}
|
||||
|| ${JSON.stringify(condition)}::jsonb)`,
|
||||
updatedAt: new Date(),
|
||||
}).where(and(eq(agentWakeupRequests.companyId, scope.companyId), eq(agentWakeupRequests.id, scope.requestId),
|
||||
eq(agentWakeupRequests.status, "deferred_issue_execution")));
|
||||
eq(agentWakeupRequests.status, "deferred_issue_execution"))).returning({ id: agentWakeupRequests.id });
|
||||
return Boolean(saved);
|
||||
});
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
import { PersistedExecutionWaitRefusal } from "../deferred-execution-wait.js";
|
||||
import { and, asc, eq, gt, inArray, isNull, or, sql } from "drizzle-orm";
|
||||
import { agentWakeupRequests, type Db } from "@paperclipai/db";
|
||||
import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js";
|
||||
@@ -105,7 +106,11 @@ export function registerDeferredReceiptWork(
|
||||
onError: error => { errors.push(error); },
|
||||
});
|
||||
else if (kind === "task_chat") await owner.settleTaskChat(scope);
|
||||
else await owner.resume({ ...scope, onError: error => { errors.push(error); } });
|
||||
else {
|
||||
const failed = (error: unknown) => { if (!(error instanceof PersistedExecutionWaitRefusal)) errors.push(error); };
|
||||
try { await owner.resume({ ...scope, onError: failed }); }
|
||||
catch (error) { failed(error); }
|
||||
}
|
||||
if (errors.length) throw new AggregateError(errors, `Deferred receipt ${entry.id} could not settle`);
|
||||
return next === null ? null : { ...entry, dueAt: next };
|
||||
},
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { notifyRoutineExecutionStatus } from "../routine-execution-status-notifications.js";
|
||||
import { notifyStatusCardIssueChange } from "../status-card-notifications.js";
|
||||
import { writeDeferredExecutionWait } from "../deferred-execution-wait.js";
|
||||
import { PersistedExecutionWaitRefusal, writeDeferredExecutionWait } from "../deferred-execution-wait.js";
|
||||
import { notifyDeferredExecutionWork } from "../deferred-execution-notifications.js";
|
||||
import type { DeferredReceiptRecoveryOptions } from "../deferred-execution-notifications.js";
|
||||
import { notifyRunDispatch } from "../run-dispatch-notifications.js";
|
||||
@@ -2522,17 +2522,17 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
|
||||
skipReason: string,
|
||||
patch: Partial<typeof agentWakeupRequests.$inferInsert> = {},
|
||||
waitCondition?: Record<string, unknown>,
|
||||
) => {
|
||||
): Promise<{ created: boolean; waitSaved?: boolean }> => {
|
||||
// A temporary refusal must leave the handoff retryable, without burning
|
||||
// its one durable receipt ID on a skipped wake.
|
||||
if (caseRequest) return { created: false };
|
||||
if (executionWaitRequestId) {
|
||||
await writeDeferredExecutionWait(db, { companyId: agent.companyId, requestId: executionWaitRequestId }, {
|
||||
const waitSaved = await writeDeferredExecutionWait(db, { companyId: agent.companyId, requestId: executionWaitRequestId }, {
|
||||
reason: skipReason, message: patch.error ?? (skipReason === "issue_tree_hold_active"
|
||||
? "This task is paused. Resume it to send your saved message."
|
||||
: "Waiting for task execution to be enabled. Your message is saved."),
|
||||
});
|
||||
return { created: false };
|
||||
return { created: false, waitSaved };
|
||||
}
|
||||
const request = {
|
||||
...durableReceiptFields,
|
||||
@@ -2804,31 +2804,30 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
|
||||
},
|
||||
);
|
||||
if (budgetBlock) {
|
||||
await writeSkippedRequest("budget.blocked", { error: budgetBlock.reason }, {
|
||||
const wait = await writeSkippedRequest("budget.blocked", { error: budgetBlock.reason }, {
|
||||
scopeType: budgetBlock.scopeType, scopeId: budgetBlock.scopeId,
|
||||
});
|
||||
if (caseRequest) return null;
|
||||
throw conflict(budgetBlock.reason, {
|
||||
scopeType: budgetBlock.scopeType,
|
||||
scopeId: budgetBlock.scopeId,
|
||||
});
|
||||
const details = { scopeType: budgetBlock.scopeType, scopeId: budgetBlock.scopeId };
|
||||
throw wait.waitSaved ? new PersistedExecutionWaitRefusal(budgetBlock.reason, details)
|
||||
: conflict(budgetBlock.reason, details);
|
||||
}
|
||||
|
||||
// Setup delays execution, but must not discard work accepted by a caller.
|
||||
const invokability = await getAgentInvokability(isAgentAwaitingSetup(agent) ? { ...agent, status: "idle" } : agent);
|
||||
if (!invokability.invokable) {
|
||||
if (caseRequest) return null;
|
||||
let waitSaved = false;
|
||||
if (opts.requestedByActorType !== "user" || executionWaitRequestId) {
|
||||
await writeSkippedRequest("agent.not_invokable", {
|
||||
const wait = await writeSkippedRequest("agent.not_invokable", {
|
||||
error: invokability.message,
|
||||
}, { status: agent.status, reason: invokability.reason });
|
||||
waitSaved = wait.waitSaved === true;
|
||||
}
|
||||
throw conflict(invokability.message, {
|
||||
status: agent.status,
|
||||
reason: invokability.reason,
|
||||
invalidOrgChain: invokability.invalidOrgChain,
|
||||
...invokability.details,
|
||||
});
|
||||
const details = { status: agent.status, reason: invokability.reason,
|
||||
invalidOrgChain: invokability.invalidOrgChain, ...invokability.details };
|
||||
throw waitSaved ? new PersistedExecutionWaitRefusal(invokability.message, details)
|
||||
: conflict(invokability.message, details);
|
||||
}
|
||||
|
||||
const policy = parseHeartbeatPolicy(agent);
|
||||
|
||||
Reference in new issue
Block a user