mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-08 01:57:31 +01:00
fix(server): wake failed retry notices from committed changes (#17649)
## Thinking Path > - Paperclip keeps exact failed chat retries in durable action and wake receipts. > - Deferred and declined retries owe safe notices to their original conversation. > - Periodic publication maintenance repeatedly inspects those receipts when idle. > - Committing receipt, authority, lifecycle and lease changes already identifies when eligibility can change. > - A scoped owner reconstructs durable obligations, drains bounded candidates, retains contention, then becomes quiet. ## Linked Issues or Issue Description This existing failed-retry notice slice follows #17647. It removes the failed-retry notice duty from periodic publication maintenance while preserving the explicit manual API. Existing action, receipt, publication and source rows remain authoritative; no new table or scheduler is added. **Problem:** Removing the scan alone loses notices after real admission, cancellation, early refusal, requester or agent restoration, and relevant lease settlement. Millisecond cursor truncation can also repeat denied history. **Expected behavior:** Committed supported writes wake the scoped notice owner, which uses current authorization and precise cursors. Real retries survive failure, contention, restart and drain; settled work has no recurring SQL. ## What Changed - Extract failed-retry projection under the existing authorization owner. Preserve current receipt/source authority and provider transport authorization. - Register bounded scoped work using the existing delivery coordinator/scheduler. Preserve exact PostgreSQL cursor precision, contention and a wake arriving during reconciliation. - Extend the shared transaction-aware notification boundary to failed-retry notices. Cover admission, conditional denial, early refusal, requester changes, and agent lifecycle transitions to an invokable status. Two shared receipt hooks and their budget fixture correction already landed in the preceding inbound slice; preserve them once. - Notify when relevant native lease cleanup completes or a reusable lease is retired, preserving claim fencing and caller-owned transactions. - Avoid the discovery scan identified in #17647: known task scopes require zero discovery SQL; company/startup discovery uses a single DISTINCT projection from the same candidate predicate. Bounded current authorization still runs afterward. - Remove only this duty from periodic publication maintenance, retain manual behavior and startup reconstruction, and add real producer/lifecycle/query-budget regressions with an audit. ## Verification Published head: `fa95a1f2fb9d546aa62dbd3b43aa714f5c88bd35`, targeting master after #17647 merged at `1aa3a21ac41f602977638cf0be875b01a243e69f`. - **42 distinct current combined-master cases pass:** 21 retry-notice/native-lease/restored-input producer cases (1,651 skipped), plus all 21 app/startup and task-drain admission-release cases. Server typechecking, build and module boundaries pass. The final follow-up adds acceptance documentation only; all 17 owned server source/test hashes are unchanged. - The prior **404-case** matrix passes at prepared revision `c2c53b839497be6f5e28aa2cd5ba50cfd1d76d4a`: 259 across six complete consumer/lifecycle/environment suites, 20 notice/lease cases, 53 ordinary-notice and 72 Retry/budget cases. Those chat selections skip 1,651/1,618/1,599 cases. A subsequent 21-case combined producer check passes after the parent restoration repair. These earlier runs retain their revision scope. - The normal two-parent integration `43d0f84ea56845b8fc4946a7533ecbf7f07b679a` preserves the complete 19-path owned binary patch and all 23 unrelated master paths/deletions. The semantic merge base is the verified landed parent head `41f7e2c9e351efa6cfa5ef699c0e69df7425012b`; squash ancestry otherwise causes historical conflicts. Both parent commits and the exact resolved tree are verified; no rebase or force push occurs. - Three scope-discovery regressions fail before the fix. The first SQL projection fails with PostgreSQL 42702 (ambiguous company column), then passes after using a distinct company alias. Known task hints have zero discovery SQL; company/startup discovery has one query returning only distinct task scopes. Existing bounded projection/current authorization remain intact. - Current producer tests cover commit/rollback, parked authority restoration, real agent resume, early refusal, requester changes, native lease cleanup/replacement, drain/restart/shutdown, precise cursors and contention. Parent shared admission/denial/restored-input notifications are preserved once. No polling backstop or storage is added. - [The owner audit](https://github.com/paperclipai/paperclip/blob/fa95a1f2fb9d546aa62dbd3b43aa714f5c88bd35/doc/plans/2026-10-25-failed-retry-notices-sleep-audit.md#final-landed-parent-acceptance--october-28) records current commands, fixtures, failure history and limits. Fresh exact-head remote checks and Greptile 5/5 are pending; old-head reviews do not count. The owner audit preserves prior failures and their original revision scope. The old remote head's Runner receipt timeout, cancelled gate, and review error are not current passes; fresh exact-head checks and Greptile 5/5 are required. No full local repository suite, live provider tenant, external wake alarm, physical suspension or savings claim is made. ### Final remote gates Exact published head `fa95a1f2fb9d546aa62dbd3b43aa714f5c88bd35` has Greptile 5/5 with no actionable findings, successful current main CI and ready-triggered security scan. The final dependency/quality review workflow `40747725096` passed after waiting for a GitHub-hosted runner. No Ponytail verdict is present. Two automatic main CI requests and two identical-head quality-review requests were created one second apart. The older main run was automatically cancelled; only the proven duplicate, unstarted quality request was cancelled manually. Their same-head, same-workflow replacements both finish successfully. The merge guard verifies these replacements, all current checks and current required-status integration rules before excluding the superseded cancellation contexts. No actual failing review or current check is waived and no protection bypass is used. ## Risks - Process-local signals cover supported writers using this boundary, not independent SQL or other replicas. - Lease completion now reads/locks the exact row before updating. Reusable lease retirement wakes company scope because its previous issue can differ. Claims and transaction ownership remain enforced. - Broad discovery can inspect history inside PostgreSQL; DISTINCT reduces returned rows and round trips, without claiming constant database cost. Task hints are wake requests, so bounded reconciliation handles already exhausted scopes. - Native filesystem activation belongs to execution authorization and is not an eligibility input for this notice projection. Other notice/maintenance duties and Cloud sleep gates remain separate. ## Model Used OpenAI Codex, based on GPT-6, with reasoning, local code inspection, shell execution, and PostgreSQL testing. The exact serving model identifier and context-window size are not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
c4b6168e08
commit
f4f02880ca
19 files changed
+1321
-187
No files matched your search
@@ -91,7 +91,8 @@ describe("createChatReconciliationCoordinator", () => {
|
||||
const publicationMaintenance = chat.slice(chat.indexOf("async function processPublicationMaintenance("),
|
||||
chat.indexOf("async function schedulePendingPublications("));
|
||||
expect(publicationMaintenance).not.toContain("await enqueueInboundWakeupPublications(");
|
||||
expect(publicationMaintenance).toContain("await enqueueFailedChatRetryPublications(");
|
||||
expect(publicationMaintenance).not.toContain("await enqueueFailedChatRetryPublications(");
|
||||
expect(source).toContain("chatFailedRetryNoticeWork(chatChannels).register(deliveryWork,");
|
||||
});
|
||||
|
||||
it("separates live voice recovery from the remaining periodic maintenance", () => {
|
||||
|
||||
@@ -1,3 +1,6 @@
|
||||
import { environmentService } from "../services/environments.js";
|
||||
import { createAdmissionTransactionScope, createWakeAdmissionWriter } from "../modules/wake-queue/adapters/postgres.js";
|
||||
import { chatFailedRetryNoticeWork } from "../services/chat-failed-retry-notice-work.js";
|
||||
import { issueTreeControlService } from "../services/issue-tree-control.js";
|
||||
import { chatInboundWakeNoticeWork } from "../services/chat-inbound-wake-notice-work.js";
|
||||
import { nativeChatProgressWork } from "../services/chat-native-progress-work.js";
|
||||
@@ -14,7 +17,7 @@ import { chatRunSourceMessageId } from "../services/chat-channels/run-origin.js"
|
||||
import * as durableChatWakeup from "../services/durable-chat-wakeup.js";
|
||||
import { conflict } from "../errors.js";
|
||||
import * as githubTaskPreparation from "../services/chat-channels/github/task-preparation.js";
|
||||
import { configureAgentLifecycle } from "../services/agent-lifecycle.js";
|
||||
import { configureAgentLifecycle, createAgentLifecycle } from "../services/agent-lifecycle.js";
|
||||
import { budgetPolicies } from "@paperclipai/db";
|
||||
import { budgetService } from "../services/budgets.js";
|
||||
import { idleWorkSnapshot } from "../services/task-admission.js";
|
||||
@@ -28,7 +31,7 @@ import { createDatabaseResponseCapture, installDatabaseResponseCapture } from "@
|
||||
import { githubResponseAuthorityFields } from "../services/chat-channels/github/response-authority-read.js";
|
||||
import * as activityLogModule from "../services/activity-log.js";
|
||||
import * as connectionCredentialBindings from "../services/connection-credential-bindings.js";
|
||||
import { chatSlackManagerGrants, chatSlackRegistrations, toolOauthStates, environments } from "@paperclipai/db";
|
||||
import { chatSlackManagerGrants, chatSlackRegistrations, toolOauthStates } from "@paperclipai/db";
|
||||
import type { PaperclipCloudConnector, SealedConnectorCredentials } from "../services/paperclip-cloud-connector.js";
|
||||
import { buildSlackAppManifest, SLACK_MANAGER_SCOPES } from "@paperclipai/shared";
|
||||
import { registerChatActionWork, dueChatAction } from "../services/chat-action-work.js";
|
||||
@@ -143,6 +146,7 @@ import {
|
||||
nativeRunResults,
|
||||
nativeRunFinalizations,
|
||||
environmentLeases,
|
||||
environments,
|
||||
principalPermissionGrants,
|
||||
projects,
|
||||
projectWorkspaces,
|
||||
@@ -1098,17 +1102,18 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
secretsTmpDir,
|
||||
"master.key",
|
||||
);
|
||||
if (externalTestDatabaseUrl) {
|
||||
db = createDb(externalTestDatabaseUrl);
|
||||
} else {
|
||||
tempDb = await startEmbeddedPostgresTestDatabase(
|
||||
"paperclip-chat-channels-",
|
||||
);
|
||||
db = createDb(tempDb.connectionString);
|
||||
if (!externalTestDatabaseUrl) {
|
||||
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-chat-channels-");
|
||||
}
|
||||
// Scheduler tests advance virtual time while PostgreSQL uses real I/O.
|
||||
// Do not expire or time out its connections on those virtual clock jumps.
|
||||
db = createDb(externalTestDatabaseUrl ?? tempDb!.connectionString, {
|
||||
idleTimeoutSeconds: 0, connectTimeoutSeconds: 0, maxLifetimeSeconds: 0,
|
||||
});
|
||||
}, 90_000);
|
||||
|
||||
afterAll(async () => {
|
||||
await db?.$client.end();
|
||||
await tempDb?.cleanup();
|
||||
if (previousKeyFile === undefined)
|
||||
delete process.env.PAPERCLIP_SECRETS_MASTER_KEY_FILE;
|
||||
@@ -61698,45 +61703,6 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
);
|
||||
});
|
||||
|
||||
async function deferredFailedRetryNoticeFixture(wakeOnDemand = true) {
|
||||
const context = await failedChatRetryFixture("telegram", undefined, true);
|
||||
const companyId = context.fixture.companyId, issueId = context.issue.id, leaseId = randomUUID();
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issueId,
|
||||
nativePhase: "terminal_failure", runnerProfileJson: {},
|
||||
resultJson: { workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: {
|
||||
schema: "paperclip.workspace-restore-recovery.v1", runId: context.runId, leaseIds: [leaseId],
|
||||
} },
|
||||
}).where(eq(heartbeatRuns.id, context.runId));
|
||||
await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId: context.runId, phase: "terminal_failure" });
|
||||
await db.insert(environmentLeases).values({ id: leaseId, companyId, issueId,
|
||||
heartbeatRunId: context.runId, provider: "daytona", providerLeaseId: randomUUID(),
|
||||
status: "pending_cleanup", cleanupStatus: "failed", leasePolicy: "retain_on_failure",
|
||||
metadata: { remoteCwd: "/home/daytona/paperclip-workspace", pendingCleanupInFlight: false },
|
||||
});
|
||||
await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId,
|
||||
kind: "active_run_watchdog", status: "resolved", ownerType: "system",
|
||||
cause: "native_continuation_requires_reconciliation", fingerprint: randomUUID(),
|
||||
nextAction: "Verify and preserve the retained source before continuing.",
|
||||
evidence: { runId: context.runId, automaticRecovery: { replay: "blocked" },
|
||||
workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [leaseId] },
|
||||
historicalWorkspaceSource: { schema: "paperclip.historical-workspace-source.v1", runId: context.runId, leaseIds: [leaseId] },
|
||||
systemRecovery: { ...createSystemRecovery(context.runId), state: "exhausted", attempts: 2, retriesUsed: 1,
|
||||
attemptId: randomUUID(), firstFailureAt: Date.now() - 1000, deadlineAt: Date.now() + 60_000,
|
||||
preservationOperationIds: [randomUUID()] },
|
||||
},
|
||||
});
|
||||
const retry = await requestTaskRetry({ db, companyId, issueId, actorId: "owner-user",
|
||||
actorSource: "cloud_tenant", observedRunId: context.runId, idempotencyKey: randomUUID(),
|
||||
prepareChatRetry: context.service.prepareFailedChatRunRetry });
|
||||
await db.update(agents).set({ adapterType: "paperclip_runner", status: "running",
|
||||
runtimeConfig: { heartbeat: { wakeOnDemand, maxConcurrentRuns: 1 } },
|
||||
}).where(eq(agents.id, context.fixture.assignedAgentId));
|
||||
await db.insert(heartbeatRuns).values({ companyId, agentId: context.fixture.assignedAgentId,
|
||||
status: "running", startedAt: new Date() });
|
||||
context.wakeup.mockImplementation(heartbeatService(db).wakeup);
|
||||
return { ...context, retry, admit: () => context.service.processFailedChatRunRetry(retry.chatActionId!) };
|
||||
}
|
||||
|
||||
it("registers ordinary-notice work through the retry admission transaction", async () => {
|
||||
const context = await deferredFailedRetryNoticeFixture();
|
||||
const events: string[] = [];
|
||||
@@ -64040,6 +64006,8 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
"integrity",
|
||||
"cleanup",
|
||||
"active_environment",
|
||||
"notice_cleanup",
|
||||
"notice_replacement",
|
||||
"live_pid",
|
||||
"missing_input",
|
||||
"late_provider_event",
|
||||
@@ -64073,6 +64041,9 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
: providerAccount;
|
||||
const previousStateDirectory = process.env.PAPERCLIP_RUNNER_STATE_DIR;
|
||||
let stateDirectory: string | null = null;
|
||||
let noticeCoordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
|
||||
let noticeActionId: string | undefined;
|
||||
let noticeEnvironmentId: string | undefined;
|
||||
try {
|
||||
const nativeSessionId = randomUUID();
|
||||
const runnerInstanceId = randomUUID();
|
||||
@@ -64281,7 +64252,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
);
|
||||
// A null account without existing provider-session evidence remains
|
||||
// conservatively ineligible for the exhausted-run recovery lane.
|
||||
if (["bootstrap", "checkpoint", "distinct_account"].includes(kind)) {
|
||||
if (["bootstrap", "checkpoint", "distinct_account", "notice_cleanup", "notice_replacement"].includes(kind)) {
|
||||
const staged = await attempt;
|
||||
await expect(
|
||||
context.service.processFailedChatRunRetry(staged.actionId),
|
||||
@@ -64291,12 +64262,101 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
retryOfRunId: context.runId,
|
||||
wakeCommentIds: [context.action.payload.commentId],
|
||||
});
|
||||
if (kind === "notice_cleanup" || kind === "notice_replacement") {
|
||||
noticeActionId = staged.actionId;
|
||||
await db.update(agentWakeupRequests).set({ status: "deferred_issue_execution" })
|
||||
.where(eq(agentWakeupRequests.id, staged.actionId));
|
||||
const cleanupAttemptId = randomUUID();
|
||||
if (kind === "notice_replacement") {
|
||||
const [environment] = await db.insert(environments).values({ name: randomUUID(), driver: "ssh" }).returning();
|
||||
noticeEnvironmentId = environment.id;
|
||||
}
|
||||
const [lease] = await db.insert(environmentLeases).values({
|
||||
companyId: context.fixture.companyId, issueId: context.issue.id,
|
||||
heartbeatRunId: context.runId, status: "pending_cleanup", cleanupStatus: "failed",
|
||||
metadata: { pendingCleanupAttemptId: cleanupAttemptId, agentId: context.fixture.assignedAgentId },
|
||||
...(kind === "notice_replacement" ? { environmentId: noticeEnvironmentId,
|
||||
status: "retained", leasePolicy: "reuse_by_environment", providerLeaseId: randomUUID() } : {}),
|
||||
}).returning();
|
||||
const errors: unknown[] = [];
|
||||
noticeCoordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true,
|
||||
canReconcile: () => true, onError: error => errors.push(error) });
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
|
||||
await chatFailedRetryNoticeWork(context.service).register(noticeCoordinator, () => true).ready;
|
||||
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000);
|
||||
expect(noticeCoordinator!.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); });
|
||||
expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0);
|
||||
const replace = (owner: Db, replacesReusableLeaseId = lease.id) => environmentService(owner).acquireLease({
|
||||
companyId: context.fixture.companyId, environmentId: noticeEnvironmentId!, issueId: context.issue.id,
|
||||
leasePolicy: "reuse_by_environment", providerLeaseId: lease.providerLeaseId,
|
||||
metadata: { agentId: context.fixture.assignedAgentId }, replacesReusableLeaseId,
|
||||
});
|
||||
if (kind === "notice_replacement") {
|
||||
await expect(replace(db, randomUUID())).rejects.toThrow("ownership changed");
|
||||
const rollback = new Error("rollback reusable lease handoff");
|
||||
await expect(db.transaction(async tx => {
|
||||
await replace(tx as unknown as Db);
|
||||
throw rollback;
|
||||
})).rejects.toBe(rollback);
|
||||
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000);
|
||||
expect(noticeCoordinator!.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); });
|
||||
expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0);
|
||||
await db.transaction(tx => replace(tx as unknown as Db));
|
||||
} else {
|
||||
expect(await environmentService(db).releaseLease(lease.id, "released", {
|
||||
cleanupStatus: "success", expectedPendingCleanupAttemptId: randomUUID(),
|
||||
})).toBeNull();
|
||||
const rollback = new Error("rollback native lease settlement");
|
||||
await expect(db.transaction(async tx => {
|
||||
await environmentService(tx as unknown as Db).releaseLease(lease.id, "released", {
|
||||
cleanupStatus: "success", expectedPendingCleanupAttemptId: cleanupAttemptId,
|
||||
});
|
||||
throw rollback;
|
||||
})).rejects.toBe(rollback);
|
||||
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000);
|
||||
expect(noticeCoordinator!.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); });
|
||||
expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0);
|
||||
await db.transaction(async tx => {
|
||||
await environmentService(tx as unknown as Db).releaseLease(lease.id, "released", {
|
||||
cleanupStatus: "success", expectedPendingCleanupAttemptId: cleanupAttemptId,
|
||||
});
|
||||
});
|
||||
}
|
||||
await vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1);
|
||||
expect(noticeCoordinator!.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0);
|
||||
}, { timeout: 5000 });
|
||||
const selection = vi.spyOn(context.service, "loadFailedRetryNoticeScopes");
|
||||
try {
|
||||
if (kind === "notice_replacement") await expect(replace(db)).rejects.toThrow("ownership changed");
|
||||
else await environmentService(db).releaseLease(lease.id, "released", { cleanupStatus: "success" });
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
expect(selection).not.toHaveBeenCalled();
|
||||
} finally { selection.mockRestore(); }
|
||||
const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction");
|
||||
try {
|
||||
await vi.advanceTimersByTimeAsync(120_000);
|
||||
expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled();
|
||||
} finally { select.mockRestore(); transaction.mockRestore(); }
|
||||
expect(errors).toEqual([]);
|
||||
}
|
||||
} else
|
||||
await expect(attempt).rejects.toMatchObject({
|
||||
details: { code: "chat_failed_run_retry_not_authorized" },
|
||||
});
|
||||
} finally {
|
||||
await noticeCoordinator?.stop();
|
||||
if (kind === "notice_cleanup" || kind === "notice_replacement") vi.useRealTimers();
|
||||
await context.service.shutdown();
|
||||
if (noticeEnvironmentId) {
|
||||
await db.delete(environmentLeases).where(eq(environmentLeases.environmentId, noticeEnvironmentId));
|
||||
await db.delete(environments).where(eq(environments.id, noticeEnvironmentId));
|
||||
}
|
||||
if (noticeActionId) {
|
||||
await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, noticeActionId));
|
||||
await db.delete(chatActions).where(eq(chatActions.id, noticeActionId));
|
||||
}
|
||||
if (stateDirectory) {
|
||||
if (previousStateDirectory === undefined)
|
||||
delete process.env.PAPERCLIP_RUNNER_STATE_DIR;
|
||||
@@ -64781,6 +64841,447 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
}
|
||||
});
|
||||
|
||||
async function deferredFailedRetryNoticeFixture(wakeOnDemand = true) {
|
||||
const context = await failedChatRetryFixture("telegram", undefined, true);
|
||||
const companyId = context.fixture.companyId, issueId = context.issue.id, leaseId = randomUUID();
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issueId,
|
||||
nativePhase: "terminal_failure", runnerProfileJson: {},
|
||||
resultJson: { workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: {
|
||||
schema: "paperclip.workspace-restore-recovery.v1", runId: context.runId, leaseIds: [leaseId],
|
||||
} },
|
||||
}).where(eq(heartbeatRuns.id, context.runId));
|
||||
await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId: context.runId, phase: "terminal_failure" });
|
||||
await db.insert(environmentLeases).values({ id: leaseId, companyId, issueId,
|
||||
heartbeatRunId: context.runId, provider: "daytona", providerLeaseId: randomUUID(),
|
||||
status: "pending_cleanup", cleanupStatus: "failed", leasePolicy: "retain_on_failure",
|
||||
metadata: { remoteCwd: "/home/daytona/paperclip-workspace", pendingCleanupInFlight: false },
|
||||
});
|
||||
await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId,
|
||||
kind: "active_run_watchdog", status: "resolved", ownerType: "system",
|
||||
cause: "native_continuation_requires_reconciliation", fingerprint: randomUUID(),
|
||||
nextAction: "Verify and preserve the retained source before continuing.",
|
||||
evidence: { runId: context.runId, automaticRecovery: { replay: "blocked" },
|
||||
workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [leaseId] },
|
||||
historicalWorkspaceSource: { schema: "paperclip.historical-workspace-source.v1", runId: context.runId, leaseIds: [leaseId] },
|
||||
systemRecovery: { ...createSystemRecovery(context.runId), state: "exhausted", attempts: 2, retriesUsed: 1,
|
||||
attemptId: randomUUID(), firstFailureAt: Date.now() - 1000, deadlineAt: Date.now() + 60_000,
|
||||
preservationOperationIds: [randomUUID()] },
|
||||
},
|
||||
});
|
||||
const retry = await requestTaskRetry({ db, companyId, issueId, actorId: "owner-user",
|
||||
actorSource: "cloud_tenant", observedRunId: context.runId, idempotencyKey: randomUUID(),
|
||||
prepareChatRetry: context.service.prepareFailedChatRunRetry });
|
||||
await db.update(agents).set({ adapterType: "paperclip_runner", status: "running",
|
||||
runtimeConfig: { heartbeat: { wakeOnDemand, maxConcurrentRuns: 1 } },
|
||||
}).where(eq(agents.id, context.fixture.assignedAgentId));
|
||||
await db.insert(heartbeatRuns).values({ companyId, agentId: context.fixture.assignedAgentId,
|
||||
status: "running", startedAt: new Date() });
|
||||
context.wakeup.mockImplementation(heartbeatService(db).wakeup);
|
||||
return { ...context, retry, admit: () => context.service.processFailedChatRunRetry(retry.chatActionId!) };
|
||||
}
|
||||
|
||||
it.each(["normal", "drain", "failed selection", "stopped", "disabled"] as const)("reactivates failed retry notices from real admission after idle through %s", async mode => {
|
||||
const context = await deferredFailedRetryNoticeFixture(mode !== "disabled");
|
||||
let canRun = true;
|
||||
const errors: unknown[] = [];
|
||||
let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun,
|
||||
canReconcile: () => true, onError: error => errors.push(error) });
|
||||
const selection = vi.spyOn(context.service, "loadFailedRetryNoticeScopes");
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
|
||||
const settle = async () => vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(10);
|
||||
expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull();
|
||||
}, { timeout: 10_000 });
|
||||
const quiet = async () => {
|
||||
const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction");
|
||||
try {
|
||||
await vi.advanceTimersByTimeAsync(120_000);
|
||||
expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled();
|
||||
} finally { select.mockRestore(); transaction.mockRestore(); }
|
||||
};
|
||||
try {
|
||||
await chatFailedRetryNoticeWork(context.service).register(coordinator, () => canRun).ready;
|
||||
await settle(); await quiet();
|
||||
if (mode === "drain") canRun = false;
|
||||
if (mode === "stopped") await coordinator.stop();
|
||||
if (mode === "failed selection") selection.mockRejectedValueOnce(new Error("temporary retry selection failure"));
|
||||
expect(await context.admit()).toMatchObject({ status: mode === "disabled" ? "failed" : "deferred" });
|
||||
if (mode === "failed selection") {
|
||||
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10); expect(errors).toHaveLength(1); });
|
||||
canRun = false;
|
||||
}
|
||||
if (["drain", "failed selection", "stopped"].includes(mode)) {
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0));
|
||||
expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0);
|
||||
await quiet(); canRun = true;
|
||||
if (mode === "stopped") {
|
||||
await context.service.processPublicationMaintenance();
|
||||
expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0);
|
||||
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun,
|
||||
canReconcile: () => true, onError: error => errors.push(error) });
|
||||
await chatFailedRetryNoticeWork(context.service).register(coordinator, () => canRun).ready;
|
||||
}
|
||||
}
|
||||
await vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([expect.objectContaining({
|
||||
payload: mode === "disabled"
|
||||
? { text: "This retry was not started. Open the task in Paperclip for details.", progressState: "failed" }
|
||||
: { text: "Your retry is queued.", progressState: "queued" },
|
||||
})]);
|
||||
expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0);
|
||||
}, { timeout: 10_000 });
|
||||
await quiet();
|
||||
expect(errors).toEqual(mode === "failed selection" ? [expect.objectContaining({ errors: [expect.objectContaining({ message: "temporary retry selection failure" })] })] : []);
|
||||
} finally {
|
||||
await coordinator.stop(); selection.mockRestore(); vi.useRealTimers(); await context.service.shutdown();
|
||||
await db.delete(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, context.fixture.companyId),
|
||||
inArray(agentWakeupRequests.id, [context.retry.requestId!, context.retry.chatActionId!]), isNull(agentWakeupRequests.runId)));
|
||||
}
|
||||
});
|
||||
|
||||
it.each(["idle", "bounded pass", "requester merge", "agent resume"] as const)("reactivates denied retry notices when authority is restored during %s", async mode => {
|
||||
const context = await deferredFailedRetryNoticeFixture();
|
||||
const rejectedIds = mode === "bounded pass" ? Array.from({ length: 100 }, () => randomUUID()) : [];
|
||||
const errors: unknown[] = [];
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true,
|
||||
canReconcile: () => true, onError: error => errors.push(error) });
|
||||
let held = false, release!: () => void;
|
||||
const pending = new Promise<void>(resolve => { release = resolve; });
|
||||
let ready: Promise<unknown> | undefined;
|
||||
let lifecycle: ReturnType<typeof configureAgentLifecycle> | undefined;
|
||||
const reconcile = context.service.reconcileFailedRetryNotices.bind(context.service);
|
||||
const pass = vi.spyOn(context.service, "reconcileFailedRetryNotices");
|
||||
try {
|
||||
expect(await context.admit()).toMatchObject({ status: "deferred" });
|
||||
const [originalReceipt] = await readAgentWakeupRequestsById(context.retry.chatActionId!);
|
||||
if (mode === "agent resume") {
|
||||
lifecycle = configureAgentLifecycle(db, { requiredPluginIds: async () => [],
|
||||
runHost: async () => "complete", runPlugin: async () => "complete" });
|
||||
await createAgentLifecycle(db).pauseAgent(context.fixture.assignedAgentId);
|
||||
await lifecycle.process(context.fixture.assignedAgentId);
|
||||
expect((await db.select().from(agents).where(eq(agents.id, context.fixture.assignedAgentId)))[0])
|
||||
.toMatchObject({ status: "paused", lifecycleState: "paused" });
|
||||
} else if (mode === "requester merge") {
|
||||
expect(originalReceipt.requestedByActorType).toBe("user");
|
||||
await db.update(agentWakeupRequests).set({ requestedByActorId: "another-requester" })
|
||||
.where(eq(agentWakeupRequests.id, originalReceipt.id));
|
||||
} else {
|
||||
await db.update(issues).set({ visibility: "private", privacyRootIssueId: context.issue.id,
|
||||
createdByUserId: null, responsibleUserId: null, assigneeUserId: null,
|
||||
}).where(eq(issues.id, context.issue.id));
|
||||
}
|
||||
if (rejectedIds.length) {
|
||||
const [action] = await db.select().from(chatActions).where(eq(chatActions.id, context.retry.chatActionId!));
|
||||
const [receipt] = await readAgentWakeupRequestsById(action.id);
|
||||
await db.insert(chatActions).values(rejectedIds.map(id => ({ ...action, id,
|
||||
providerActionId: `retry-cursor:${id}`, createdAt: new Date(action.createdAt.getTime() + 1000),
|
||||
payload: { ...action.payload, sourceScopeSha256: "invalid-source" },
|
||||
})));
|
||||
await db.insert(agentWakeupRequests).values(rejectedIds.map(id => ({ ...receipt, id,
|
||||
idempotencyKey: `retry-cursor:${id}`,
|
||||
})));
|
||||
pass.mockImplementationOnce(async input => {
|
||||
const result = await reconcile(input);
|
||||
expect(result).toMatchObject({ inserted: 0, nextCursor: expect.any(Object) });
|
||||
held = true; await pending; return result;
|
||||
});
|
||||
}
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
|
||||
ready = chatFailedRetryNoticeWork(context.service).register(coordinator, () => true).ready;
|
||||
if (mode === "bounded pass") await vi.waitFor(() => expect(held).toBe(true), { timeout: 10_000 });
|
||||
else {
|
||||
await ready;
|
||||
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(10);
|
||||
expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0); });
|
||||
const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction");
|
||||
try {
|
||||
await vi.advanceTimersByTimeAsync(120_000);
|
||||
expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled();
|
||||
} finally { select.mockRestore(); transaction.mockRestore(); }
|
||||
}
|
||||
expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(0);
|
||||
if (mode === "agent resume") {
|
||||
await createAgentLifecycle(db).resumeAgent(context.fixture.assignedAgentId);
|
||||
await lifecycle!.process(context.fixture.assignedAgentId);
|
||||
expect((await db.select().from(agents).where(eq(agents.id, context.fixture.assignedAgentId)))[0])
|
||||
.toMatchObject({ status: "idle", lifecycleState: "ready" });
|
||||
} else if (mode === "requester merge") {
|
||||
await db.transaction(async tx => {
|
||||
await tx.execute(sql`select id from issues where id = ${context.issue.id} for update`);
|
||||
await createWakeAdmissionWriter().mergeIntoExistingDeferredWake(
|
||||
createAdmissionTransactionScope(context.fixture.companyId, tx as unknown as Db), {
|
||||
companyId: context.fixture.companyId, existingDeferredWakeId: originalReceipt.id,
|
||||
mergedPayload: originalReceipt.payload ?? {}, nextCoalescedCount: 1,
|
||||
manualUserWakeActorId: originalReceipt.requestedByActorId!,
|
||||
});
|
||||
});
|
||||
} else await issueService(db).update(context.issue.id, { visibility: "open" });
|
||||
release(); await ready;
|
||||
await vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(10);
|
||||
expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1);
|
||||
expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0);
|
||||
}, { timeout: 10_000 });
|
||||
const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction");
|
||||
try {
|
||||
await vi.advanceTimersByTimeAsync(120_000);
|
||||
expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled();
|
||||
} finally { select.mockRestore(); transaction.mockRestore(); }
|
||||
expect(errors).toEqual([]);
|
||||
} finally {
|
||||
release(); await ready; await lifecycle?.stop(); await coordinator.stop(); pass.mockRestore(); vi.useRealTimers();
|
||||
await context.service.shutdown();
|
||||
await db.delete(agentWakeupRequests).where(inArray(agentWakeupRequests.id,
|
||||
[...rejectedIds, context.retry.requestId!, context.retry.chatActionId!]));
|
||||
if (rejectedIds.length) await db.delete(chatActions).where(inArray(chatActions.id, rejectedIds));
|
||||
}
|
||||
});
|
||||
|
||||
it("wakes a not-started retry notice after resumed admission is denied", async () => {
|
||||
const context = await deferredFailedRetryNoticeFixture();
|
||||
const errors: unknown[] = [];
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true,
|
||||
canReconcile: () => true, onError: error => errors.push(error) });
|
||||
try {
|
||||
expect(await context.admit()).toMatchObject({ status: "deferred" });
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
|
||||
await chatFailedRetryNoticeWork(context.service).register(coordinator, () => true).ready;
|
||||
await vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1);
|
||||
expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0);
|
||||
});
|
||||
await reconcileTaskChatRetryReceipts(db);
|
||||
context.wakeup.mockRejectedValueOnce(Object.assign(new Error("denied at resume"), {
|
||||
details: { code: "chat_failed_run_retry_not_authorized" },
|
||||
}));
|
||||
expect(await context.admit()).toMatchObject({ status: "failed" });
|
||||
expect((await readAgentWakeupRequestsById(context.retry.chatActionId!))[0].status).toBe("failed");
|
||||
await vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(2);
|
||||
expect(coordinator.nextWakeAt()).toBeNull(); expect(idleWorkSnapshot().active).toBe(0);
|
||||
}, { timeout: 5000 });
|
||||
const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction");
|
||||
try {
|
||||
await vi.advanceTimersByTimeAsync(120_000);
|
||||
expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled();
|
||||
} finally { select.mockRestore(); transaction.mockRestore(); }
|
||||
expect(errors).toEqual([]);
|
||||
} finally {
|
||||
await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown();
|
||||
await db.delete(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, context.fixture.companyId),
|
||||
inArray(agentWakeupRequests.id, [context.retry.requestId!, context.retry.chatActionId!]), isNull(agentWakeupRequests.runId)));
|
||||
}
|
||||
});
|
||||
|
||||
it.each(["agent", "tree"] as const)("wakes a failed retry notice only after %s cancellation commits", async kind => {
|
||||
const context = await failedChatRetryFixture("telegram");
|
||||
const errors: unknown[] = [];
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true,
|
||||
canReconcile: () => true, onError: error => errors.push(error) });
|
||||
let stagedActionId: string | undefined;
|
||||
const settle = async () => vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(1000);
|
||||
expect(idleWorkSnapshot().active).toBe(0); expect(coordinator.nextWakeAt()).toBeNull();
|
||||
}, { timeout: 10_000 });
|
||||
const cancel = (database: typeof db) => kind === "agent"
|
||||
? heartbeatService(database).cancelInvocationsForAgents([context.fixture.assignedAgentId], "Cancelled retry fixture")
|
||||
: issueTreeControlService(database).cancelUnclaimedWakeupsForTree(context.fixture.companyId, context.issue.id, "Cancelled retry fixture");
|
||||
try {
|
||||
const staged = await db.transaction(tx => context.service.prepareFailedChatRunRetry(tx, {
|
||||
companyId: context.fixture.companyId, issueId: context.issue.id,
|
||||
agentId: context.fixture.assignedAgentId, failedRunId: context.runId, initiatedByUserId: "owner-user",
|
||||
}));
|
||||
stagedActionId = staged.actionId;
|
||||
await context.service.processFailedChatRunRetry(staged.actionId);
|
||||
await db.update(agentWakeupRequests).set({ status: "deferred_issue_execution" })
|
||||
.where(eq(agentWakeupRequests.id, staged.actionId));
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
|
||||
await chatFailedRetryNoticeWork(context.service).register(coordinator, () => true).ready;
|
||||
await settle();
|
||||
expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1);
|
||||
const rollback = new Error("rollback retry cancellation");
|
||||
await expect(db.transaction(async tx => { await cancel(tx as unknown as typeof db); throw rollback; })).rejects.toBe(rollback);
|
||||
await settle();
|
||||
expect((await readAgentWakeupRequestsById(staged.actionId))[0].status).toBe("deferred_issue_execution");
|
||||
expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toHaveLength(1);
|
||||
await db.transaction(async tx => { await cancel(tx as unknown as typeof db); });
|
||||
await settle();
|
||||
expect((await readAgentWakeupRequestsById(staged.actionId))[0].status).toBe("cancelled");
|
||||
expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual(expect.arrayContaining([expect.objectContaining({
|
||||
payload: { text: "This retry was not started. Open the task in Paperclip for details.", progressState: "failed" },
|
||||
})]));
|
||||
const select = vi.spyOn(db, "select"), transaction = vi.spyOn(db, "transaction");
|
||||
try {
|
||||
await vi.advanceTimersByTimeAsync(120_000);
|
||||
expect(select).not.toHaveBeenCalled(); expect(transaction).not.toHaveBeenCalled();
|
||||
} finally { select.mockRestore(); transaction.mockRestore(); }
|
||||
expect(errors).toEqual([]);
|
||||
} finally {
|
||||
await coordinator.stop(); vi.useRealTimers(); await context.service.shutdown();
|
||||
if (stagedActionId) {
|
||||
await db.delete(chatActions).where(eq(chatActions.id, stagedActionId));
|
||||
await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, stagedActionId));
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
it.each(["manual", "scoped"] as const)("advances failed retry notices past denied history with microsecond timestamps (%s)", async mode => {
|
||||
const context = await failedChatRetryFixture("telegram");
|
||||
const rejectedIds = Array.from({ length: 50 }, () => randomUUID());
|
||||
let stagedActionId: string | undefined;
|
||||
try {
|
||||
const staged = await db.transaction(tx => context.service.prepareFailedChatRunRetry(tx, {
|
||||
companyId: context.fixture.companyId, issueId: context.issue.id,
|
||||
agentId: context.fixture.assignedAgentId, failedRunId: context.runId, initiatedByUserId: "owner-user",
|
||||
}));
|
||||
stagedActionId = staged.actionId;
|
||||
await context.service.processFailedChatRunRetry(staged.actionId);
|
||||
const [receipt] = await db.update(agentWakeupRequests).set({ status: "deferred_issue_execution" })
|
||||
.where(eq(agentWakeupRequests.id, staged.actionId)).returning();
|
||||
const [action] = await db.select().from(chatActions).where(eq(chatActions.id, staged.actionId));
|
||||
const before = new Date(action.createdAt.getTime() - 1000).toISOString();
|
||||
await db.insert(chatActions).values(rejectedIds.map(id => ({
|
||||
...action, id, providerActionId: `failed-run-retry:${id}`,
|
||||
payload: { ...action.payload, failedRunId: randomUUID() },
|
||||
createdAt: sql`${before}::timestamptz + interval '0.000123 seconds'`,
|
||||
})));
|
||||
await db.insert(agentWakeupRequests).values(rejectedIds.map(id => ({
|
||||
...receipt, id, idempotencyKey: `retry-notice-cursor:${id}`,
|
||||
})));
|
||||
if (mode === "manual") {
|
||||
expect(await context.service.enqueueInboundWakeupPublications(50)).toBe(0);
|
||||
expect(await context.service.enqueueInboundWakeupPublications(50)).toBe(1);
|
||||
} else {
|
||||
const scope = { companyId: context.fixture.companyId, issueId: context.issue.id };
|
||||
const transaction = vi.spyOn(db, "transaction");
|
||||
try {
|
||||
const first = await context.service.reconcileFailedRetryNotices({ scope, limit: 50 });
|
||||
expect(first).toMatchObject({ inserted: 0, contended: false, nextCursor: expect.any(Object) });
|
||||
expect(transaction).toHaveBeenCalledTimes(50); transaction.mockClear();
|
||||
expect(await context.service.reconcileFailedRetryNotices({ scope, limit: 50, cursor: first.nextCursor }))
|
||||
.toEqual({ inserted: 1, contended: false, nextCursor: null });
|
||||
expect(transaction).toHaveBeenCalledTimes(1);
|
||||
} finally { transaction.mockRestore(); }
|
||||
}
|
||||
expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([expect.objectContaining({
|
||||
idempotencyKey: `wake:${staged.actionId}:queued:${context.endpoint.id}:${context.conversation.id}`,
|
||||
payload: { text: "Your retry is queued.", progressState: "queued" },
|
||||
})]);
|
||||
} finally {
|
||||
await context.service.shutdown();
|
||||
const actionIds = stagedActionId ? [...rejectedIds, stagedActionId] : rejectedIds;
|
||||
await db.delete(chatActions).where(inArray(chatActions.id, actionIds));
|
||||
await db.delete(agentWakeupRequests).where(inArray(agentWakeupRequests.id, actionIds));
|
||||
}
|
||||
});
|
||||
|
||||
it.each(["task", "company", "startup"] as const)("bounds failed retry notice scope discovery with retained candidates (%s)", async mode => {
|
||||
const context = await failedChatRetryFixture("telegram");
|
||||
const other = await failedChatRetryFixture("telegram");
|
||||
const copiedIds = Array.from({ length: 250 }, () => randomUUID());
|
||||
const stagedIds: string[] = [];
|
||||
try {
|
||||
for (const target of [context, other]) {
|
||||
const staged = await db.transaction(tx => target.service.prepareFailedChatRunRetry(tx, {
|
||||
companyId: target.fixture.companyId, issueId: target.issue.id,
|
||||
agentId: target.fixture.assignedAgentId, failedRunId: target.runId, initiatedByUserId: "owner-user",
|
||||
}));
|
||||
await target.service.processFailedChatRunRetry(staged.actionId);
|
||||
await db.update(agentWakeupRequests).set({ status: "deferred_issue_execution" })
|
||||
.where(eq(agentWakeupRequests.id, staged.actionId));
|
||||
stagedIds.push(staged.actionId);
|
||||
}
|
||||
const [action] = await db.select().from(chatActions).where(eq(chatActions.id, stagedIds[0]));
|
||||
const [receipt] = await readAgentWakeupRequestsById(stagedIds[0]);
|
||||
await db.insert(chatActions).values(copiedIds.map(id => ({
|
||||
...action, id, providerActionId: `failed_run_retry:${id}`,
|
||||
})));
|
||||
await db.insert(agentWakeupRequests).values(copiedIds.map(id => ({
|
||||
...receipt, id, idempotencyKey: `retry-discovery-budget:${id}`,
|
||||
})));
|
||||
const scope = { companyId: context.fixture.companyId, issueId: context.issue.id };
|
||||
const events: Array<Record<string, unknown>> = [], now = Date.now();
|
||||
const capture = createDatabaseResponseCapture({ start: now - 1, end: now + 59_000 }, event => events.push(event));
|
||||
installDatabaseResponseCapture(db, { chatActions, agentWakeupRequests }, capture);
|
||||
let scopes: Array<{ companyId: string; issueId: string }>;
|
||||
try {
|
||||
scopes = await context.service.loadFailedRetryNoticeScopes(
|
||||
mode === "task" ? scope : mode === "company" ? { companyId: scope.companyId } : undefined,
|
||||
);
|
||||
} finally { capture.flush(); }
|
||||
if (mode === "startup") {
|
||||
expect(scopes).toEqual(expect.arrayContaining([scope, {
|
||||
companyId: other.fixture.companyId, issueId: other.issue.id,
|
||||
}]));
|
||||
} else {
|
||||
expect(scopes).toEqual([scope]);
|
||||
}
|
||||
expect(events.at(-1)).toMatchObject({ startedQueries: mode === "task" ? 0 : 1,
|
||||
completedQueries: mode === "task" ? 0 : 1, inFlightAtEnd: 0 });
|
||||
expect(events.filter(event => event.event === "paperclip.db_response_capture.entry.v1")
|
||||
.reduce((count, event) => count + Number(event.rows), 0)).toBe(mode === "task" ? 0 : scopes.length);
|
||||
} finally {
|
||||
await context.service.shutdown(); await other.service.shutdown();
|
||||
await db.delete(chatActions).where(inArray(chatActions.id, copiedIds));
|
||||
await db.delete(agentWakeupRequests).where(inArray(agentWakeupRequests.id, copiedIds));
|
||||
}
|
||||
});
|
||||
|
||||
it("scopes failed retry notices and retains receipt contention without retrying settled history", async () => {
|
||||
const context = await failedChatRetryFixture("telegram");
|
||||
const other = await failedChatRetryFixture("telegram");
|
||||
let release = () => {};
|
||||
let holder: Promise<unknown> | undefined;
|
||||
try {
|
||||
const staged = [];
|
||||
for (const target of [context, other]) {
|
||||
const action = await db.transaction(tx => target.service.prepareFailedChatRunRetry(tx, {
|
||||
companyId: target.fixture.companyId, issueId: target.issue.id,
|
||||
agentId: target.fixture.assignedAgentId, failedRunId: target.runId, initiatedByUserId: "owner-user",
|
||||
}));
|
||||
await target.service.processFailedChatRunRetry(action.actionId);
|
||||
await db.update(agentWakeupRequests).set({ status: "deferred_issue_execution" })
|
||||
.where(eq(agentWakeupRequests.id, action.actionId));
|
||||
staged.push(action);
|
||||
}
|
||||
const scope = { companyId: context.fixture.companyId, issueId: context.issue.id };
|
||||
expect(await context.service.loadFailedRetryNoticeScopes({ companyId: scope.companyId })).toEqual([scope]);
|
||||
let locked!: () => void;
|
||||
const ready = new Promise<void>(resolve => { locked = resolve; });
|
||||
const hold = new Promise<void>(resolve => { release = resolve; });
|
||||
holder = db.transaction(async tx => {
|
||||
await tx.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, staged[0].actionId)).for("update");
|
||||
locked(); await hold;
|
||||
});
|
||||
await ready;
|
||||
expect(await context.service.reconcileFailedRetryNotices({ scope }))
|
||||
.toEqual({ inserted: 0, contended: true, nextCursor: null });
|
||||
expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual([]);
|
||||
release(); await holder;
|
||||
expect(await context.service.reconcileFailedRetryNotices({ scope }))
|
||||
.toEqual({ inserted: 1, contended: false, nextCursor: null });
|
||||
expect(await context.service.loadFailedRetryNoticeScopes({ companyId: scope.companyId })).toEqual([]);
|
||||
expect(await readChatPublicationsByEndpointId(other.endpoint.id)).toEqual([]);
|
||||
await db.update(agentWakeupRequests).set({ status: "cancelled" })
|
||||
.where(eq(agentWakeupRequests.id, staged[0].actionId));
|
||||
expect(await context.service.reconcileFailedRetryNotices({ scope }))
|
||||
.toEqual({ inserted: 1, contended: false, nextCursor: null });
|
||||
expect(await readChatPublicationsByEndpointId(context.endpoint.id)).toEqual(expect.arrayContaining([expect.objectContaining({
|
||||
payload: { text: "This retry was not started. Open the task in Paperclip for details.", progressState: "failed" },
|
||||
})]));
|
||||
expect(await context.service.reconcileFailedRetryNotices({ scope }))
|
||||
.toEqual({ inserted: 0, contended: false, nextCursor: null });
|
||||
expect(await context.service.loadFailedRetryNoticeScopes({ companyId: scope.companyId })).toEqual([]);
|
||||
await db.update(agentWakeupRequests).set({ runId: other.runId })
|
||||
.where(eq(agentWakeupRequests.id, staged[1].actionId));
|
||||
expect(await other.service.loadFailedRetryNoticeScopes({ companyId: other.fixture.companyId })).toEqual([]);
|
||||
} finally { release(); await holder; await context.service.shutdown(); await other.service.shutdown(); }
|
||||
});
|
||||
|
||||
it.each(["slack", "telegram"] as const)(
|
||||
"keeps a %s retry's queue, working, and selected final on its own provider identity",
|
||||
async (provider) => {
|
||||
|
||||
@@ -11,6 +11,7 @@ import {
|
||||
companySecretBindings,
|
||||
companySecrets,
|
||||
createDb,
|
||||
closeRegisteredClients,
|
||||
environmentCustomImageSetupSessions,
|
||||
environmentLeases,
|
||||
environments,
|
||||
@@ -26,7 +27,11 @@ import {
|
||||
} from "./helpers/embedded-postgres.js";
|
||||
import { environmentService } from "../services/environments.ts";
|
||||
|
||||
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
|
||||
const externalTestDatabaseUrl = process.env.PAPERCLIP_TEST_DATABASE_URL;
|
||||
if (externalTestDatabaseUrl && !new URL(externalTestDatabaseUrl).pathname.endsWith("_test")) {
|
||||
throw new Error("Environment service tests require an isolated database ending in _test");
|
||||
}
|
||||
const embeddedPostgresSupport = externalTestDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
|
||||
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
|
||||
|
||||
if (!embeddedPostgresSupport.supported) {
|
||||
@@ -41,9 +46,14 @@ describeEmbeddedPostgres("environmentService leases", () => {
|
||||
let svc!: ReturnType<typeof environmentService>;
|
||||
|
||||
beforeAll(async () => {
|
||||
const started = await startEmbeddedPostgresTestDatabase("environment-service");
|
||||
stopDb = started.stop;
|
||||
db = createDb(started.connectionString);
|
||||
if (externalTestDatabaseUrl) {
|
||||
db = createDb(externalTestDatabaseUrl);
|
||||
stopDb = () => closeRegisteredClients(externalTestDatabaseUrl);
|
||||
} else {
|
||||
const started = await startEmbeddedPostgresTestDatabase("environment-service");
|
||||
stopDb = started.stop;
|
||||
db = createDb(started.connectionString);
|
||||
}
|
||||
svc = environmentService(db);
|
||||
});
|
||||
|
||||
|
||||
@@ -197,6 +197,7 @@ import { chatChannelService } from "./services/chat-channels.js";
|
||||
import { deliverNativeQuestionResponse } from "./services/native-runtime/native-question-bridge.js";
|
||||
import { nativeChatProgressWork } from "./services/chat-native-progress-work.js";
|
||||
import { chatInboundWakeNoticeWork } from "./services/chat-inbound-wake-notice-work.js";
|
||||
import { chatFailedRetryNoticeWork } from "./services/chat-failed-retry-notice-work.js";
|
||||
import { chatRunMilestoneWork } from "./services/chat-run-milestone-work.js";
|
||||
import {
|
||||
createCoalescedAsyncTrigger,
|
||||
@@ -1290,6 +1291,7 @@ export async function createApp(
|
||||
chatChannels.registerCommandWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||
nativeChatProgressWork(db).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||
chatInboundWakeNoticeWork(chatChannels).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||
chatFailedRetryNoticeWork(chatChannels).register(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive());
|
||||
chatRunMilestoneWork(db, opts.authPublicBaseUrl).register(
|
||||
deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive(),
|
||||
);
|
||||
|
||||
@@ -1,3 +1,6 @@
|
||||
import { subscribeDatabaseWork } from "@paperclipai/db";
|
||||
import { DELIVERY_QUEUES } from "../../../services/delivery-work-notifications.js";
|
||||
import { resolveExternalTestDatabaseUrl } from "../../../__tests__/helpers/external-test-database.js";
|
||||
import { createAgentLifecycle as createLifecycleCommands } from "../index.js";
|
||||
import { createAgentLifecycleEffects } from "../../../services/agent-lifecycle.js";
|
||||
import { deleteCompany } from "../../../services/company-deletion.js";
|
||||
@@ -26,14 +29,13 @@ import { transition } from "../domain/policy.js";
|
||||
import type { LifecycleAgent } from "../application/ports.js";
|
||||
import { budgetService } from "../../../services/budgets.js";
|
||||
import { agentConfigurationRevision } from "../../../lib/agent-configuration-revision.js";
|
||||
import { resolveExternalTestDatabaseUrl } from "../../../__tests__/helpers/external-test-database.js";
|
||||
|
||||
const createLifecycleStore = (db: Db) => createStore(db, createAgentLifecycleEffects());
|
||||
|
||||
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_AGENT_LIFECYCLE_TEST_DATABASE_URL");
|
||||
const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
|
||||
(support.supported ? describe : describe.skip)("agent lifecycle commands", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let db: Db;
|
||||
let companyId: string;
|
||||
const workers: ReturnType<typeof configureAgentLifecycle>[] = [];
|
||||
@@ -637,6 +639,48 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
|
||||
}
|
||||
});
|
||||
|
||||
it("settles notice reactivation only with the outer lifecycle transaction", async () => {
|
||||
const work = worker();
|
||||
const agent = await hire();
|
||||
await work.process(agent.id);
|
||||
await createAgentLifecycle(db).pauseAgent(agent.id);
|
||||
await work.process(agent.id);
|
||||
await work.stop();
|
||||
await createAgentLifecycle(db).resumeAgent(agent.id);
|
||||
const store = createLifecycleStore(db);
|
||||
const owner = "resume-notice-test";
|
||||
const claimed = await store.claim(agent.id, owner, new Date());
|
||||
expect(claimed).not.toBeNull();
|
||||
await store.setRequiredPlugins(claimed!, owner, []);
|
||||
await store.completeHost(claimed!, owner);
|
||||
const events: string[] = [];
|
||||
const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.chatFailedRetryNotices, (event, scopes) => {
|
||||
if ([...scopes ?? []].some(scope => JSON.parse(scope).companyId === companyId)) events.push(event);
|
||||
});
|
||||
try {
|
||||
await expect(db.transaction(async tx => {
|
||||
expect(await createLifecycleStore(tx as unknown as Db).change(agent.id, "complete", { owner }))
|
||||
.toMatchObject({ status: "idle", lifecycleState: "ready" });
|
||||
expect(events).toEqual(["intent"]);
|
||||
expect(await current(agent.id)).toMatchObject({ status: "paused", lifecycleState: "resuming" });
|
||||
throw new Error("rollback resumed agent");
|
||||
})).rejects.toThrow("rollback resumed agent");
|
||||
expect(events).toEqual(["intent", "settled"]);
|
||||
expect(await current(agent.id)).toMatchObject({ status: "paused", lifecycleState: "resuming" });
|
||||
events.length = 0;
|
||||
await db.transaction(async tx => {
|
||||
await createLifecycleStore(tx as unknown as Db).change(agent.id, "complete", { owner });
|
||||
expect(events).toEqual(["intent"]);
|
||||
expect((await current(agent.id)).status).toBe("paused");
|
||||
});
|
||||
expect(events).toEqual(["intent", "settled"]);
|
||||
expect(await current(agent.id)).toMatchObject({ status: "idle", lifecycleState: "ready" });
|
||||
events.length = 0;
|
||||
await store.change(agent.id, "reconcile");
|
||||
expect(events).toEqual([]);
|
||||
} finally { unsubscribe(); }
|
||||
});
|
||||
|
||||
it("rejects transaction injection and generic state writes", async () => {
|
||||
const agent = await hire("pending_approval");
|
||||
await expect(db.transaction(async tx => createAgentLifecycle(tx as unknown as Db))).rejects.toThrow("root database");
|
||||
|
||||
@@ -8,6 +8,7 @@ import type { ActivityPublication } from "../../../types/activity-publication.js
|
||||
export interface LifecycleEffects {
|
||||
setupRequiresPreparation(db: Db, companyId: string, environmentId?: string | null): Promise<boolean>;
|
||||
notifyWork(db: Db): Promise<void>;
|
||||
beforeStatusChange(db: Db, companyId: string, agentId: string, before: string, after: string): Promise<void>;
|
||||
reportFailure(context: LifecycleFailure, error: unknown): void;
|
||||
deleteDependencies(db: Db, companyId: string, agentId: string): Promise<void>;
|
||||
prepareConfiguration(db: Db, current: AgentConfigurationRecord, data: AgentConfigurationPatch, options?: UpdateAgentOptions): Promise<AgentConfigurationPatch>;
|
||||
|
||||
@@ -80,6 +80,7 @@ async function changeLockedAgent(tx: Db, effects: LifecycleEffects, agent: Lifec
|
||||
};
|
||||
if (next === "ready") operation.resumeState = "ready";
|
||||
const status = compatibilityStatus(next, agent.status);
|
||||
if (status !== agent.status) await effects.beforeStatusChange(tx, agent.companyId, agent.id, agent.status, status);
|
||||
if (isLifecycleWorkPending(next)) await effects.notifyWork(tx);
|
||||
const [updated] = await tx.update(agents).set({
|
||||
lifecycleHolds: holds, lifecycleState: next, lifecycleVersion: agent.lifecycleVersion + 1, lifecycleError: null,
|
||||
|
||||
@@ -996,6 +996,7 @@ export function createWakeAdmissionWriter(): WakeAdmissionWriter {
|
||||
|
||||
async mergeIntoExistingDeferredWake(scope, input) {
|
||||
const tx = requireAdmissionTx(scope, input.companyId);
|
||||
if (input.manualUserWakeActorId) await notifyChatInboundWakeNotices(tx, { companyId: input.companyId });
|
||||
const rows = await tx
|
||||
.update(agentWakeupRequests)
|
||||
.set({
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import { isAgentStatusInvokable } from "@paperclipai/shared";
|
||||
import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js";
|
||||
import { logger } from "../middleware/logger.js";
|
||||
import { companies, type Db } from "@paperclipai/db";
|
||||
import { eq } from "drizzle-orm";
|
||||
@@ -33,6 +35,11 @@ export function createAgentLifecycleEffects(_hooks: BudgetServiceHooks = {}): li
|
||||
return true;
|
||||
},
|
||||
notifyWork: tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle),
|
||||
async beforeStatusChange(tx, companyId, _agentId, before, after) {
|
||||
if (!isAgentStatusInvokable(before) && isAgentStatusInvokable(after)) {
|
||||
await notifyChatInboundWakeNotices(tx, { companyId });
|
||||
}
|
||||
},
|
||||
reportFailure(context, error) {
|
||||
const code = lifecycle.lifecycleFailureCode(error);
|
||||
logger.warn({ ...context, failureCode: code ?? "unclassified" }, "Agent lifecycle operation failed");
|
||||
|
||||
@@ -1236,6 +1236,8 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
processFailedChatRunRetry,
|
||||
processFailedChatRunRetries,
|
||||
enqueueFailedChatRetryPublications,
|
||||
loadFailedRetryNoticeScopes,
|
||||
reconcileFailedRetryNotices,
|
||||
pendingFailedRetries,
|
||||
} = createChatFailedRunRetry(db, {
|
||||
authorizeInboundWakeup,
|
||||
@@ -3348,9 +3350,6 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
|
||||
async function processPublicationMaintenance(limit = 25) {
|
||||
scheduleGitHubMaintenance();
|
||||
// Normal inbound notices are commit-driven. Failed-run retries retain their
|
||||
// separate maintenance owner until its own producer audit is complete.
|
||||
await enqueueFailedChatRetryPublications(50);
|
||||
await reconcileTerminalConfirmationActions(limit);
|
||||
await processPendingInteractionWakeups(limit);
|
||||
}
|
||||
@@ -3504,6 +3503,8 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
const { prepareTeamsBoardSend } = createTeamsBoardSendPreparation(db, { endpointRecord, runtimeFor });
|
||||
const { getPublicationBatchStatus, publishComment, publishBoardMessage } = createChatPublicationRequests(db, { endpointRecord, stageTeamsBoardFileIntent, processPendingSlackBoardMessages, processPendingPublications, prepareTeamsBoardSend });
|
||||
return {
|
||||
loadFailedRetryNoticeScopes,
|
||||
reconcileFailedRetryNotices,
|
||||
registerTeamsAttemptRecoveryWork,
|
||||
registerTeamsConsentExpiryWork,
|
||||
registerCommandWork,
|
||||
|
||||
@@ -0,0 +1,107 @@
|
||||
import { and, asc, eq, inArray, isNull, sql } from "drizzle-orm";
|
||||
import { agentWakeupRequests, chatActions, chatPublications, type Db } from "@paperclipai/db";
|
||||
import type { DbOrTransaction } from "./runtime-types.js";
|
||||
import { notifyChatPublicationWork } from "../chat-work-notifications.js";
|
||||
import { projectSafeChatPublication } from "../chat-publication-projection.js";
|
||||
import { inboundWakePublicationKey } from "../chat-inbound-wakeup-publications.js";
|
||||
import { NativeChatReviewPresentationContentionError } from "../native-runtime/native-chat-review-presentation.js";
|
||||
import { isExternalChatWaitAuthorizationContention } from "../native-runtime/chat-attachment-reuse.js";
|
||||
|
||||
export type FailedRetryNoticeScope = { companyId: string; issueId?: string };
|
||||
export type FailedRetryNoticeCursor = { createdAt: string; id: string };
|
||||
type Options = {
|
||||
authorize: (tx: DbOrTransaction, action: typeof chatActions.$inferSelect) => Promise<{
|
||||
endpointId: string; conversationId: string; issueId: string; commentIds: string[];
|
||||
}>;
|
||||
isExternalActionAuthorizationChange: (error: unknown) => boolean;
|
||||
isShuttingDown: () => boolean;
|
||||
};
|
||||
|
||||
export function createFailedRetryNotices(db: Db, options: Options) {
|
||||
let manualCursor: FailedRetryNoticeCursor | null = null;
|
||||
function candidateQuery(cursor: FailedRetryNoticeCursor | null, scope?: FailedRetryNoticeScope) {
|
||||
return db.select({ action: chatActions, receipt: agentWakeupRequests,
|
||||
cursorCreatedAt: sql<string>`${chatActions.createdAt}::text`,
|
||||
companyId: sql<string>`${chatActions.companyId}`.as("retry_notice_company_id"),
|
||||
issueId: sql<string>`${chatActions.payload}->>'issueId'`.as("retry_notice_issue_id"),
|
||||
}).from(chatActions).innerJoin(agentWakeupRequests, and(
|
||||
eq(agentWakeupRequests.id, chatActions.id), eq(agentWakeupRequests.companyId, chatActions.companyId),
|
||||
)).where(and(
|
||||
eq(chatActions.kind, "failed_run_retry"), isNull(agentWakeupRequests.runId),
|
||||
scope ? eq(chatActions.companyId, scope.companyId) : undefined,
|
||||
scope?.issueId ? sql`${chatActions.payload}->>'issueId' = ${scope.issueId}` : undefined,
|
||||
sql`${chatActions.payload}->>'issueId' ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$'`,
|
||||
cursor ? sql`(${chatActions.createdAt}, ${chatActions.id}) > (${cursor.createdAt}::timestamptz, ${cursor.id}::uuid)` : undefined,
|
||||
inArray(agentWakeupRequests.status, ["deferred_issue_execution", "cancelled", "failed", "skipped"]),
|
||||
sql`not exists (select 1 from chat_publications retry_notice where retry_notice.company_id = ${chatActions.companyId}
|
||||
and retry_notice.idempotency_key = 'wake:' || ${chatActions.id}::text || ':' ||
|
||||
case when ${agentWakeupRequests.status} = 'deferred_issue_execution' then 'queued' else 'not_started' end || ':' ||
|
||||
${chatActions.endpointId}::text || ':' || ${chatActions.conversationId}::text)`,
|
||||
));
|
||||
}
|
||||
async function candidates(limit: number, cursor: FailedRetryNoticeCursor | null, scope?: FailedRetryNoticeScope) {
|
||||
return candidateQuery(cursor, scope).orderBy(asc(chatActions.createdAt), asc(chatActions.id)).limit(limit);
|
||||
}
|
||||
type Candidates = Awaited<ReturnType<typeof candidates>>;
|
||||
function continuation(rows: Candidates, limit: number): FailedRetryNoticeCursor | null {
|
||||
const last = rows.at(-1);
|
||||
return rows.length === limit && last ? { createdAt: last.cursorCreatedAt, id: last.action.id } : null;
|
||||
}
|
||||
async function project(rows: Candidates, signal?: AbortSignal) {
|
||||
let inserted = 0, contended = false;
|
||||
for (const { action, receipt } of rows) {
|
||||
signal?.throwIfAborted();
|
||||
if (options.isShuttingDown()) throw new Error("Failed retry notice reconciliation stopped");
|
||||
try {
|
||||
inserted += await db.transaction(async tx => {
|
||||
const source = await options.authorize(tx, action);
|
||||
const [current] = await tx.select().from(agentWakeupRequests)
|
||||
.where(eq(agentWakeupRequests.id, receipt.id)).for("update", { noWait: true });
|
||||
if (!current || current.runId || current.status !== receipt.status) return 0;
|
||||
const state = current.status === "deferred_issue_execution" ? "queued" : "not_started";
|
||||
await notifyChatPublicationWork(tx);
|
||||
return (await tx.insert(chatPublications).values({
|
||||
companyId: action.companyId, endpointId: source.endpointId, conversationId: source.conversationId,
|
||||
issueId: source.issueId, commentId: source.commentIds.at(-1)!,
|
||||
idempotencyKey: inboundWakePublicationKey(action.id, state, source.endpointId, source.conversationId),
|
||||
payload: projectSafeChatPublication({ classification: "external", source: "safe_milestone",
|
||||
text: state === "queued" ? "Your retry is queued."
|
||||
: "This retry was not started. Open the task in Paperclip for details.",
|
||||
progressState: state === "queued" ? "queued" : "failed",
|
||||
}), state: "pending",
|
||||
}).onConflictDoNothing().returning({ id: chatPublications.id })).length;
|
||||
});
|
||||
} catch (error) {
|
||||
if (isExternalChatWaitAuthorizationContention(error)
|
||||
|| error instanceof NativeChatReviewPresentationContentionError) contended = true;
|
||||
else if (options.isExternalActionAuthorizationChange(error)
|
||||
|| (error && typeof error === "object" && "status" in error && error.status === 409)) continue;
|
||||
else throw error;
|
||||
}
|
||||
}
|
||||
return { inserted, contended };
|
||||
}
|
||||
async function enqueueFailedChatRetryPublications(limit: number) {
|
||||
if (options.isShuttingDown()) return 0;
|
||||
const size = Math.max(1, Math.min(50, limit));
|
||||
const rows = await candidates(size, manualCursor);
|
||||
manualCursor = continuation(rows, size);
|
||||
return (await project(rows)).inserted;
|
||||
}
|
||||
async function loadFailedRetryNoticeScopes(scope?: FailedRetryNoticeScope, signal?: AbortSignal) {
|
||||
signal?.throwIfAborted();
|
||||
if (scope?.issueId) return [{ companyId: scope.companyId, issueId: scope.issueId }];
|
||||
const candidates = candidateQuery(null, scope).as("failed_retry_notice_candidates");
|
||||
return db.selectDistinct({ companyId: candidates.companyId, issueId: candidates.issueId })
|
||||
.from(candidates);
|
||||
}
|
||||
async function reconcileFailedRetryNotices(input: {
|
||||
scope: FailedRetryNoticeScope; limit?: number; cursor?: FailedRetryNoticeCursor | null; signal?: AbortSignal;
|
||||
}) {
|
||||
input.signal?.throwIfAborted();
|
||||
const limit = Math.max(1, Math.min(input.limit ?? 100, 200));
|
||||
const rows = await candidates(limit, input.cursor ?? null, input.scope);
|
||||
return { ...await project(rows, input.signal), nextCursor: continuation(rows, limit) };
|
||||
}
|
||||
return { enqueueFailedChatRetryPublications, loadFailedRetryNoticeScopes, reconcileFailedRetryNotices };
|
||||
}
|
||||
@@ -1,4 +1,5 @@
|
||||
import { notifyChatInboundWakeNotices } from "../chat-inbound-wake-notice-notifications.js";
|
||||
import { createFailedRetryNotices } from "./failed-retry-notices.js";
|
||||
import { safeChatWakeAuthorityReason, type ChatWakeAuthorityReason } from "./inbound/wakeup-authorization-error.js";
|
||||
import { adapterExecutionControls } from "../adapter-execution-control.js";
|
||||
import { admitExplicitNativeContinuation } from "../explicit-native-continuation.js";
|
||||
@@ -87,13 +88,8 @@ import {
|
||||
import {
|
||||
NativeChatReviewPresentationContentionError,
|
||||
} from "../native-runtime/native-chat-review-presentation.js";
|
||||
import { isExternalChatWaitAuthorizationContention } from "../native-runtime/chat-attachment-reuse.js";
|
||||
import { CHAT_CONVERSATION_ID_KEY } from "../../modules/run-dispatch/index.js";
|
||||
import { projectSafeChatPublication } from "../chat-publication-projection.js";
|
||||
|
||||
import {
|
||||
inboundWakePublicationKey,
|
||||
} from "../chat-inbound-wakeup-publications.js";
|
||||
|
||||
type DbOrTransaction = Db | Parameters<Parameters<Db["transaction"]>[0]>[0];
|
||||
type DeliveryRow = typeof chatDeliveries.$inferSelect;
|
||||
@@ -1980,117 +1976,13 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
|
||||
return candidates.length;
|
||||
}
|
||||
|
||||
let failedRetryNoticeCursor: { createdAt: Date; id: string } | null = null;
|
||||
async function enqueueFailedChatRetryPublications(
|
||||
limit: number,
|
||||
): Promise<number> {
|
||||
const cursor = failedRetryNoticeCursor;
|
||||
const batchSize = Math.max(1, Math.min(50, limit));
|
||||
const candidates = await db
|
||||
.select({ action: chatActions, receipt: agentWakeupRequests })
|
||||
.from(chatActions)
|
||||
.innerJoin(
|
||||
agentWakeupRequests,
|
||||
and(
|
||||
eq(agentWakeupRequests.id, chatActions.id),
|
||||
eq(agentWakeupRequests.companyId, chatActions.companyId),
|
||||
),
|
||||
)
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.kind, "failed_run_retry"),
|
||||
isNull(agentWakeupRequests.runId),
|
||||
cursor
|
||||
? sql`(${chatActions.createdAt}, ${chatActions.id}) > (${cursor.createdAt.toISOString()}::timestamptz, ${cursor.id}::uuid)`
|
||||
: undefined,
|
||||
inArray(agentWakeupRequests.status, [
|
||||
"deferred_issue_execution",
|
||||
"cancelled",
|
||||
"failed",
|
||||
"skipped",
|
||||
]),
|
||||
sql`not exists (select 1 from chat_publications retry_notice where retry_notice.company_id = ${chatActions.companyId}
|
||||
and retry_notice.idempotency_key = 'wake:' || ${chatActions.id}::text || ':' ||
|
||||
case when ${agentWakeupRequests.status} = 'deferred_issue_execution' then 'queued' else 'not_started' end || ':' ||
|
||||
${chatActions.endpointId}::text || ':' || ${chatActions.conversationId}::text)`,
|
||||
),
|
||||
)
|
||||
.orderBy(asc(chatActions.createdAt), asc(chatActions.id))
|
||||
.limit(batchSize);
|
||||
const last = candidates.at(-1)?.action;
|
||||
failedRetryNoticeCursor =
|
||||
candidates.length >= batchSize && last
|
||||
? { createdAt: last.createdAt, id: last.id }
|
||||
: null;
|
||||
let inserted = 0;
|
||||
for (const { action, receipt } of candidates) {
|
||||
try {
|
||||
inserted += await db.transaction(async (tx) => {
|
||||
const source = await authorizeFailedChatRunRetry(tx, {
|
||||
companyId: action.companyId,
|
||||
issueId: String(action.payload.issueId),
|
||||
agentId: String(action.payload.agentId),
|
||||
retryActionId: action.id,
|
||||
publication: true,
|
||||
});
|
||||
const [current] = await tx
|
||||
.select()
|
||||
.from(agentWakeupRequests)
|
||||
.where(eq(agentWakeupRequests.id, receipt.id))
|
||||
.for("update", { noWait: true });
|
||||
if (!current || current.runId || current.status !== receipt.status)
|
||||
return 0;
|
||||
const state =
|
||||
current.status === "deferred_issue_execution"
|
||||
? "queued"
|
||||
: "not_started";
|
||||
await notifyChatPublicationWork(tx);
|
||||
return (
|
||||
await tx
|
||||
.insert(chatPublications)
|
||||
.values({
|
||||
companyId: action.companyId,
|
||||
endpointId: source.endpointId,
|
||||
conversationId: source.conversationId,
|
||||
issueId: source.issueId,
|
||||
commentId: source.commentIds.at(-1)!,
|
||||
idempotencyKey: inboundWakePublicationKey(
|
||||
action.id,
|
||||
state,
|
||||
source.endpointId,
|
||||
source.conversationId,
|
||||
),
|
||||
payload: projectSafeChatPublication({
|
||||
classification: "external",
|
||||
source: "safe_milestone",
|
||||
text:
|
||||
state === "queued"
|
||||
? "Your retry is queued."
|
||||
: "This retry was not started. Open the task in Paperclip for details.",
|
||||
progressState: state === "queued" ? "queued" : "failed",
|
||||
}),
|
||||
state: "pending",
|
||||
})
|
||||
.onConflictDoNothing()
|
||||
.returning({ id: chatPublications.id })
|
||||
).length;
|
||||
});
|
||||
} catch (error) {
|
||||
if (
|
||||
isExternalChatWaitAuthorizationContention(error) ||
|
||||
error instanceof NativeChatReviewPresentationContentionError ||
|
||||
isExternalActionAuthorizationChange(error) ||
|
||||
(error &&
|
||||
typeof error === "object" &&
|
||||
"status" in error &&
|
||||
error.status === 409)
|
||||
)
|
||||
continue;
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
return inserted;
|
||||
}
|
||||
const retryNotices = createFailedRetryNotices(db, {
|
||||
authorize: (tx, action) => authorizeFailedChatRunRetry(tx, {
|
||||
companyId: action.companyId, issueId: String(action.payload.issueId),
|
||||
agentId: String(action.payload.agentId), retryActionId: action.id, publication: true,
|
||||
}),
|
||||
isExternalActionAuthorizationChange, isShuttingDown,
|
||||
});
|
||||
|
||||
return {
|
||||
retryCommentIds,
|
||||
@@ -2103,7 +1995,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
|
||||
unregisterFailedRetryAuthority,
|
||||
processFailedChatRunRetry,
|
||||
processFailedChatRunRetries,
|
||||
enqueueFailedChatRetryPublications,
|
||||
...retryNotices,
|
||||
pendingFailedRetries: () => [...failedRetryTasks.values()],
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,39 @@
|
||||
import type { ChatChannelService } from "./chat-channels.js";
|
||||
import type { FailedRetryNoticeCursor, FailedRetryNoticeScope } from "./chat-channels/failed-retry-notices.js";
|
||||
import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
|
||||
import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
|
||||
import { registerScopedDeadlineWork } from "./scoped-deadline-work.js";
|
||||
|
||||
type Entry = {
|
||||
id: string; companyId: string; issueId: string; dueAt: number;
|
||||
cursor: FailedRetryNoticeCursor | null; contended: boolean;
|
||||
};
|
||||
const batchSize = 100;
|
||||
|
||||
export function chatFailedRetryNoticeWork(service: Pick<ChatChannelService,
|
||||
"loadFailedRetryNoticeScopes" | "reconcileFailedRetryNotices">) {
|
||||
async function load(scope?: FailedRetryNoticeScope, signal?: AbortSignal): Promise<Entry[]> {
|
||||
return (await service.loadFailedRetryNoticeScopes(scope, signal)).map(row => ({
|
||||
...row, id: `${row.companyId}:${row.issueId}`, dueAt: 0, cursor: null, contended: false,
|
||||
}));
|
||||
}
|
||||
async function check(entry: Entry, signal?: AbortSignal): Promise<Entry | null> {
|
||||
const result = await service.reconcileFailedRetryNotices({
|
||||
scope: { companyId: entry.companyId, issueId: entry.issueId },
|
||||
limit: batchSize, cursor: entry.cursor, signal,
|
||||
});
|
||||
const contended = entry.contended || result.contended;
|
||||
if (result.nextCursor) return { ...entry, cursor: result.nextCursor, contended, dueAt: Date.now() };
|
||||
return contended ? { ...entry, cursor: null, contended: false, dueAt: Date.now() + 1000 } : null;
|
||||
}
|
||||
function register(coordinator: ReturnType<typeof createDeliveryWorkCoordinator>, canRun: () => boolean) {
|
||||
return registerScopedDeadlineWork(coordinator, {
|
||||
queue: DELIVERY_QUEUES.chatFailedRetryNotices,
|
||||
decodeScope: (hint: string): FailedRetryNoticeScope | undefined => hint === "*" ? undefined : JSON.parse(hint),
|
||||
matches: (entry: Entry, scope?: FailedRetryNoticeScope) => !scope ||
|
||||
(entry.companyId === scope.companyId && (!scope.issueId || entry.issueId === scope.issueId)),
|
||||
load, check,
|
||||
}, canRun);
|
||||
}
|
||||
return { load, check, register };
|
||||
}
|
||||
@@ -1,7 +1,8 @@
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
|
||||
import type { InboundWakeNoticeScope } from "./chat-channels/inbound/wakeup-notices.js";
|
||||
|
||||
export function notifyChatInboundWakeNotices(tx: object, scope: InboundWakeNoticeScope) {
|
||||
return notifyDeliveryWork(tx, DELIVERY_QUEUES.chatInboundWakeNotices,
|
||||
JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId }));
|
||||
export async function notifyChatInboundWakeNotices(tx: object, scope: InboundWakeNoticeScope) {
|
||||
const hint = JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId });
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatInboundWakeNotices, hint);
|
||||
await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatFailedRetryNotices, hint);
|
||||
}
|
||||
@@ -59,6 +59,7 @@ export const DELIVERY_QUEUES = {
|
||||
chatPublications: "chat-publications",
|
||||
chatNativeProgress: "chat-native-progress",
|
||||
chatInboundWakeNotices: "chat-inbound-wake-notices",
|
||||
chatFailedRetryNotices: "chat-failed-retry-notices",
|
||||
chatRunMilestones: "chat-run-milestones",
|
||||
chatReceipts: "chat-slack-receipts",
|
||||
browser: "browser-use",
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js";
|
||||
import { writeEnvironmentCleanup, notifyEnvironmentCleanup } from "./environment-cleanup-notifications.js";
|
||||
import { withoutCleanupVerification } from "./cleanup-verification-proof.js";
|
||||
import { randomUUID } from "node:crypto";
|
||||
@@ -1549,6 +1550,7 @@ export function environmentService(db: Db) {
|
||||
}
|
||||
}
|
||||
if (input.replacesReusableLeaseId) {
|
||||
await notifyChatInboundWakeNotices(tx, { companyId: input.companyId });
|
||||
const retired = await tx
|
||||
.update(environmentLeases)
|
||||
.set({
|
||||
@@ -1671,8 +1673,24 @@ export function environmentService(db: Db) {
|
||||
: sql`${environmentLeases.metadata} - 'remoteExecutionTermination'`;
|
||||
const cleanupSettled = options?.expectedPendingCleanupAttemptId && options.cleanupStatus === "success"
|
||||
&& status !== "pending_cleanup";
|
||||
const row = await writeEnvironmentCleanup(db, tx => tx
|
||||
.update(environmentLeases)
|
||||
const row = await writeEnvironmentCleanup(db, async tx => {
|
||||
const condition = and(eq(environmentLeases.id, id), options?.expectedPendingCleanupAttemptId
|
||||
? and(eq(environmentLeases.status, "pending_cleanup"),
|
||||
sql`${environmentLeases.metadata}->>'pendingCleanupAttemptId' = ${options.expectedPendingCleanupAttemptId}`)
|
||||
: undefined);
|
||||
const [previous] = await tx.select({ companyId: environmentLeases.companyId, issueId: environmentLeases.issueId,
|
||||
heartbeatRunId: environmentLeases.heartbeatRunId, status: environmentLeases.status,
|
||||
cleanupStatus: environmentLeases.cleanupStatus, releasedAt: environmentLeases.releasedAt,
|
||||
}).from(environmentLeases).where(condition).for("update");
|
||||
if (!previous) return null;
|
||||
if (previous.heartbeatRunId && (previous.status !== status ||
|
||||
(options?.cleanupStatus !== undefined && previous.cleanupStatus !== options.cleanupStatus) ||
|
||||
(status !== "retained" && !previous.releasedAt))) {
|
||||
await notifyChatInboundWakeNotices(tx, {
|
||||
companyId: previous.companyId, issueId: previous.issueId ?? undefined,
|
||||
});
|
||||
}
|
||||
return tx.update(environmentLeases)
|
||||
.set({
|
||||
status,
|
||||
releasedAt: status === "retained" ? null : now,
|
||||
@@ -1686,12 +1704,10 @@ export function environmentService(db: Db) {
|
||||
? sql`coalesce(${releaseMetadata}, '{}'::jsonb) || '{"pendingCleanupInFlight":false,"pendingCleanupLeaseExpiresAtMs":0}'::jsonb`
|
||||
: releaseMetadata,
|
||||
})
|
||||
.where(and(eq(environmentLeases.id, id), options?.expectedPendingCleanupAttemptId
|
||||
? and(eq(environmentLeases.status, "pending_cleanup"),
|
||||
sql`${environmentLeases.metadata}->>'pendingCleanupAttemptId' = ${options.expectedPendingCleanupAttemptId}`)
|
||||
: undefined))
|
||||
.where(condition)
|
||||
.returning()
|
||||
.then((rows) => rows[0] ?? null));
|
||||
.then((rows) => rows[0] ?? null);
|
||||
});
|
||||
return row ? toEnvironmentLease(row) : null;
|
||||
},
|
||||
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { notifyChatInboundWakeNotices } from "../chat-inbound-wake-notice-notifications.js";
|
||||
import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js";
|
||||
import { assertTaskRetryConversationAllowed, readTaskRetryIntent, readTaskRetryBoardMessageOrigin, retireSupersededTaskRetry, taskRetryMutationAllowed } from "../task-retry.js";
|
||||
import { insertIssueComment } from "../issue-comment-writes.js";
|
||||
@@ -2498,12 +2499,24 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
|
||||
const waitIssueId = issueId;
|
||||
return db.transaction(async (tx) => {
|
||||
await tx.execute(sql`select id from issues where id = ${waitIssueId} and company_id = ${agent.companyId} for update`);
|
||||
if (durableRequest) await notifyChatInboundWakeNotices(tx, {
|
||||
companyId: durableRequest.companyId, issueId: durableRequest.issueId,
|
||||
});
|
||||
return recordExecutionWait(tx as unknown as Db, {
|
||||
issueId: waitIssueId, request, condition: waitCondition, coalesce: coalesceExecutionWait,
|
||||
});
|
||||
});
|
||||
}
|
||||
await db.insert(agentWakeupRequests).values(request);
|
||||
if (durableRequest) {
|
||||
await db.transaction(async tx => {
|
||||
await notifyChatInboundWakeNotices(tx, {
|
||||
companyId: durableRequest.companyId, issueId: durableRequest.issueId,
|
||||
});
|
||||
await tx.insert(agentWakeupRequests).values(request);
|
||||
});
|
||||
} else {
|
||||
await db.insert(agentWakeupRequests).values(request);
|
||||
}
|
||||
return { created: true };
|
||||
};
|
||||
const writeSkippedHeartbeatRequest = async (
|
||||
|
||||
Reference in new issue
Block a user