mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-04 02:32:07 +01:00
fix: dispatch queued legacy messages with operator identity and task permissions (#13315)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Task conversations save messages that arrive during an active turn. > - Legacy adapters deliver these messages in a later turn. > - A run can stop before the saved queue is delivered. > - The Interrupt button previously required an active run, so it could not release this queue. > - This pull request lets a board operator send the saved queue after the run stops and retries queues missed during finalization. > - Manual dispatch must use the clicking operator and must not require permission to create agents. > - The task can continue without a duplicate message or a second execution owner. ## Linked Issues or Issue Description **What happened?** A legacy task retained a queued message after its run stopped. Interrupt was disabled because the queue had no active target. Finalization and deferred message admission can also leave a queue without a successor. **Expected behavior** Interrupt sends the saved messages when no runner is active. Messages that arrive during normal completion are delivered automatically. An uncertain previous execution still requires proof that its process or sandbox stopped. **Steps to reproduce** 1. Queue a user message during a legacy conversation turn. 2. Let the turn stop or simulate a server restart before queue promotion. 3. Open the task with a deferred queue and no active run. 4. Try Interrupt. Before this change, the button is disabled. **Paperclip version or commit** Reproduced against `8f40b4ad4`. **Deployment mode** Legacy conversation adapter. The same persisted queue state is covered with an isolated PostgreSQL fixture. Related public work: #13275 adds active legacy interruption. #13291 addresses automatic sandbox conversation recovery. This change handles explicit saved-queue delivery and late queue promotion. ## What Changed - Accept a null Interrupt target while retaining queue identity, revision, company, and assignee checks. - Save the operator's request on the existing queue. Reuse normal admission after verified stop, including older messages, different authors, and queues whose original wake came from the system. - Strip interruption authority from caller-supplied wake payloads. Only the board queue route can persist that authority. - Retry durable interruption requests after restart and deferred queues after legacy cleanup. - Let an explicit Interrupt retry cleanup for its stopped run, including old ephemeral leases that recorded success without a provider stop receipt. Preserve retained resources, other lease owners, and the automatic retry limit. - Preserve the server's waiting explanation when normalizing and combining queue entries. - Revalidate the consumed board queue receipt at dispatch so a different message author does not cause setup failure. - Use the Interrupt user's execution identity for the new run. Preserve original message authors. Validate the receipt independently at startup and inherit the resulting identity on retry. - Persist authenticated board authority for ordinary manual wakes too. Adopting someone else's queued messages cannot switch a manual run to that author's permissions. Strip caller-supplied authority markers and retain private conversation ownership checks. - Keep the clicking user when a manual wake is merged into an older deferred receipt. Update its requester and payload in the same transaction. - Use the same current-queue/revision API on task details and pipeline conversations; show Interrupt after a legacy target stops. - Keep manual wakes out of active runs, including unscoped agent wakes. They receive their own execution identity; a matching receipt requester is not sufficient because an exact retry can retain a different originating identity. - Authorize both existing-agent wake endpoints with `agent:wake`, available to active non-viewer company members. Keep `agents:create` for hiring. Validate the stored task and current assignee before an exact task retry. - Reject viewer Interrupt requests before saving intent or stopping execution. Keep external chat retry authorization and per-action agent/user permission checks. - Preserve edits and discards until dispatch. Prevent another queue promotion when the same agent already has a successor. Keep independent reviewer recovery available. - Suppress cancelled/failed run toasts for intentional operator interruption. Keep ordinary runtime error notices. - Add UI, route, admission, restart, successor ownership, and toast regression tests. Document the behavior. - Reuse the existing socket reservation helper for both credential-quorum test cases after CI exposed an ambient-port collision. This changes test preparation only; production credential staging is still called exactly once. ## Verification - Failing regression tests reproduced the message-author identity bug and an operator's `agents:create` rejection before the fixes. - All 316 focused tests pass across eight route, queue, identity, authorization, continuation, and responsible-user suites, including the 44-test rerun of queue admission and actual startup after the final manual-wake restriction. Regressions reproduce cross-user merging both with and without a task, and same-requester receipt ambiguity. The cross-company existence guard also passes both tests. - Startup integration tests reach adapter execution under the clicking operator and retain that identity through follow-up. Coverage includes mixed authors, adopted queues, system-origin queues, restarts, forged or stale receipts, viewers, suspended memberships, changed assignees, private conversations, and caller-supplied authority markers. - The earlier queue/cleanup/UI regression suite passed 402 tests. The final review corrections pass another 180 tests across queue admission/persistence, real heartbeat startup, UI API, conversation rendering, and pipeline suites. Regression tests reproduced both review findings before correction. The final head has a 5/5 review with no unresolved threads. Full CI passes on `c2002979c`, including every general and serialized server shard, all browser shards, Paperclip Runner verification, typecheck, build, canary dry run, and the aggregate gates. - Full `pnpm -r typecheck`, `pnpm build`, and UI token gates pass after the final application changes. CI identified an outdated task-page API mock after the shared helper extraction; the fixture now exercises the real helper, and all 131 task-page/API tests pass. The final application build passes with the additional manual-wake restriction. - CI exposed a pre-existing port collision in the Codex credential-quorum fixture. It reproduced locally; both listener cases now use the existing bounded reservation helper. All 41 credential tests pass on rerun. One intervening local run hit a separate ambient bind collision in the two-occupied-port case. - The full local `pnpm test:run` attempt was stopped after host contention caused focused-suite timeouts. The affected focused tests passed on rerun. An expiring trace fixture and a missing private-conversation state were corrected. The successful full CI run is the complete-suite verification. - Hosted Interrupt previously cleared the original queue and produced exactly one successor with neutral interruption feedback. It exposed the dispatch authorization defect. Retry on that earlier build was rejected for missing `agents:create` before creating another run. - Deployed the final application build (`38257f391`) to the scoped hosted instance and verified readiness. The latest PR commit changes only the credential test fixture; application code matches that deployment. A live Retry by the same operator without `agents:create` created one successor attributed to that operator, passing the former dispatch permission gate. Startup then stopped at `configuration_incomplete` because that operator has not configured their required personal Claude Code OAuth secret; the post-deployment run page confirms the operator identity and no provider work started, and the My secrets UI still shows the token as not set. Provider execution remains unverified pending that credential. No permission grants or credentials were changed. ## Risks Queue admission and finalization can race. The task lock, durable queue receipt, current comment IDs, and successor guard prevent duplicate dispatch. Process and lease stop checks, task pauses, approvals, ownership, and budgets remain in force. The API change only allows null on legacy Interrupt; native steering still requires an active run. No schema migration is required. Active non-viewer board members can now invoke existing agents without agent-creation permission. Agent self-invocation rules, raw provider-trace admin access, task retry scope, external chat authorization, and action-specific user/agent permissions remain enforced. ## Model Used OpenAI GPT-6 through Codex. The session does not expose an exact backend model ID or context-window size. Used reasoning, repository search, code execution, tests, and browser tools. ## 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 #` 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
8d6232e7b0
commit
df984cbc2c
36 files changed
+1177
-102
No files matched your search
@@ -1784,6 +1784,7 @@ export function authorizationService(db: Db | DbTransaction) {
|
||||
}
|
||||
if (
|
||||
input.action === "agent:read" ||
|
||||
input.action === "agent:wake" ||
|
||||
input.action === "company_scope:read" ||
|
||||
input.action === "decision_queue:manage" ||
|
||||
input.action === "decision_queue:read" ||
|
||||
@@ -1798,6 +1799,7 @@ export function authorizationService(db: Db | DbTransaction) {
|
||||
// Mirroring the tasks:assign carve-out above, viewers keep the
|
||||
// read-only visibility actions but not the privileged ones.
|
||||
const requiresNonViewer =
|
||||
input.action === "agent:wake" ||
|
||||
input.action === "runtime:manage" ||
|
||||
input.action === "secrets:read" ||
|
||||
input.action === "decision_queue:manage" ||
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { and, asc, desc, eq, isNotNull, isNull, sql } from "drizzle-orm";
|
||||
import { and, asc, desc, eq, inArray, isNotNull, isNull, sql } from "drizzle-orm";
|
||||
import { z } from "zod";
|
||||
import {
|
||||
agentWakeupRequests,
|
||||
heartbeatRuns,
|
||||
@@ -11,6 +12,7 @@ import {
|
||||
import type { ExecutionContinuationEnvelope } from "@paperclipai/shared";
|
||||
import { sanitizeQuarantinedCommentForHigherTrust } from "./source-trust.js";
|
||||
import { hasConversationContinuationPolicy } from "./conversation-continuation.js";
|
||||
import { queuedCommentIdsFromWakePayload } from "./issue-queued-comment-queue.js";
|
||||
|
||||
const object = (v: unknown): Record<string, unknown> =>
|
||||
v && typeof v === "object" && !Array.isArray(v)
|
||||
@@ -268,18 +270,39 @@ export async function buildExecutionContinuation(input: {
|
||||
eq(agentWakeupRequests.reason, "retry_failed_run"), eq(agentWakeupRequests.requestedByActorType, "user"),
|
||||
sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
|
||||
)) : [];
|
||||
const authorization = reconciliations.map(row => object(row.evidence.explicitUserContinuation))
|
||||
.find(value => value.previousRunId === explicitUserSource &&
|
||||
// Admission records the board operator's authority separately from the
|
||||
// message author. At dispatch, prove that exact queue was adopted by this
|
||||
// run; caller-supplied continuation context cannot grant this authority.
|
||||
const continuationAuthorizations = reconciliations.map(row => object(row.evidence.explicitUserContinuation))
|
||||
.filter(value => value.previousRunId === explicitUserSource &&
|
||||
(!input.runId || value.runId === input.runId) &&
|
||||
value.commentId === explicitContinuation.commentId &&
|
||||
priorRuns.some(run => run.id === value.runId) &&
|
||||
(failedRunId
|
||||
priorRuns.some(run => run.id === value.runId));
|
||||
const interruptQueueIds = [...new Set(continuationAuthorizations.flatMap(value => {
|
||||
const parsed = z.string().guid().safeParse(value.queuedCommentInterruptId);
|
||||
return parsed.success ? [parsed.data] : [];
|
||||
}))];
|
||||
const interruptQueues = interruptQueueIds.length
|
||||
? await db.select().from(agentWakeupRequests).where(and(
|
||||
inArray(agentWakeupRequests.id, interruptQueueIds),
|
||||
eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.agentId, input.agentId),
|
||||
eq(agentWakeupRequests.status, "coalesced"),
|
||||
sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
|
||||
sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt' is not null`,
|
||||
)) : [];
|
||||
const authorization = continuationAuthorizations.find(value => failedRunId
|
||||
? value.failedRunId === failedRunId && retryWakes.some(wake =>
|
||||
wake.runId === value.runId && wake.requestedByActorId === value.actorId &&
|
||||
priorRuns.some(run => run.id === wake.runId && run.retryOfRunId === failedRunId))
|
||||
: rows.some(comment => comment.id === value.commentId &&
|
||||
comment.authorType === "user" && comment.authorUserId === value.actorId &&
|
||||
!comment.createdByRunId && !comment.deletedAt)));
|
||||
comment.authorType === "user" &&
|
||||
(value.queuedCommentInterruptId
|
||||
? interruptQueues.some(queue => queue.id === value.queuedCommentInterruptId &&
|
||||
queue.runId === value.runId &&
|
||||
object(object(queue.payload).queuedCommentInterrupt).actorId === value.actorId &&
|
||||
queuedCommentIdsFromWakePayload(queue.payload).includes(comment.id))
|
||||
: comment.authorUserId === value.actorId) &&
|
||||
!comment.createdByRunId && !comment.deletedAt));
|
||||
if (!predecessor || !authorization || explicitUserSource !== sourceRunId)
|
||||
throw new Error("continuation_user_authorization_missing");
|
||||
}
|
||||
|
||||
@@ -42,6 +42,143 @@ const support = await getEmbeddedPostgresTestSupport();
|
||||
actorType: "user", actorId: "board", reason: "issue_commented" };
|
||||
}
|
||||
type Fixture = Awaited<ReturnType<typeof seed>>;
|
||||
it.each(["pending", "failed", "historical", "shared", "retained"])("an explicit queued interrupt retries only its stopped sandbox, without granting automatic retries (%s)", async scenario => {
|
||||
const fails = scenario === "failed";
|
||||
const protectedLease = scenario === "shared" || scenario === "retained";
|
||||
const f = await seed(), other = await seed();
|
||||
await db.update(agents).set({ adapterType: "claude_local" }).where(eq(agents.id, f.agentId));
|
||||
await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "legacy", nativeIssueId: null, processPid: null })
|
||||
.where(eq(heartbeatRuns.id, f.sourceRunId));
|
||||
await db.update(issueRecoveryActions).set({ cause: "legacy_execution_requires_reconciliation" })
|
||||
.where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
|
||||
await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" });
|
||||
const queueId = randomUUID();
|
||||
await db.insert(agentWakeupRequests).values({ id: queueId, companyId: f.companyId, agentId: f.agentId,
|
||||
source: "automation", reason: "issue_commented", status: "deferred_issue_execution",
|
||||
requestedByActorType: "system", payload: { issueId: f.issueId, commentId: f.commentId,
|
||||
_paperclipWakeContext: { wakeCommentIds: [f.commentId] },
|
||||
queuedCommentInterrupt: { actorId: "board", requestedAt: new Date().toISOString() } },
|
||||
});
|
||||
const identities = [f, other].map(fixture => ({ id: randomUUID(), companyId: fixture.companyId,
|
||||
heartbeatRunId: fixture.sourceRunId, provider: "daytona", providerLeaseId: fixture.sourceRunId }));
|
||||
for (const identity of identities) await db.insert(environmentLeases).values({ ...identity,
|
||||
status: "pending_cleanup", leasePolicy: "ephemeral", releasedAt: new Date(), cleanupStatus: "failed",
|
||||
metadata: { pendingCleanupRetryAttempts: 5, pendingCleanupRetryCapWarned: true } });
|
||||
if (scenario === "historical" || protectedLease) await db.update(environmentLeases).set({
|
||||
status: "failed", cleanupStatus: "success",
|
||||
}).where(eq(environmentLeases.id, identities[0].id));
|
||||
if (scenario === "shared") await db.update(environmentLeases).set({
|
||||
providerLeaseId: identities[0].providerLeaseId, status: "active", releasedAt: null,
|
||||
}).where(eq(environmentLeases.id, identities[1].id));
|
||||
if (scenario === "retained") await db.update(environmentLeases).set({
|
||||
status: "retained", leasePolicy: "retain_on_failure",
|
||||
}).where(eq(environmentLeases.id, identities[0].id));
|
||||
const attempted: string[] = [];
|
||||
const heartbeat = heartbeatService(db, { environmentRuntime: {
|
||||
isPendingCleanupWorkerReady: async () => true,
|
||||
retryPendingSandboxTeardown: async ({ lease }: { lease: { id: string; providerLeaseId: string } }) => {
|
||||
attempted.push(lease.id);
|
||||
if (fails) throw new Error("Provider unavailable");
|
||||
return { providerLeaseId: lease.providerLeaseId, state: "destroyed" };
|
||||
},
|
||||
} as unknown as HeartbeatEnvironmentRuntime });
|
||||
try {
|
||||
await heartbeat.resumeQueuedCommentInterrupt(f.companyId, queueId);
|
||||
expect(attempted).toEqual([]);
|
||||
await heartbeat.resumeQueuedCommentInterrupt(f.companyId, queueId, { retryCleanup: true });
|
||||
expect(attempted).toEqual(protectedLease ? [] : [identities[0].id]);
|
||||
await heartbeat.resumeQueuedCommentInterrupt(f.companyId, queueId);
|
||||
expect(attempted).toHaveLength(protectedLease ? 0 : 1);
|
||||
const [queue] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, queueId));
|
||||
expect(queue.status).toBe(fails || protectedLease ? "deferred_issue_execution" : "coalesced");
|
||||
expect(Boolean(queue.runId)).toBe(!fails && !protectedLease);
|
||||
const [untouched] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, identities[1].id));
|
||||
expect(untouched).toMatchObject({ status: scenario === "shared" ? "active" : "pending_cleanup", metadata: { pendingCleanupRetryAttempts: 5 } });
|
||||
} finally {
|
||||
for (const identity of identities) await db.delete(environmentLeases).where(eq(environmentLeases.id, identity.id));
|
||||
}
|
||||
});
|
||||
it("a durable queue interrupt authorizes older legacy messages but still requires the provider to stop", async () => {
|
||||
const f = await seed();
|
||||
await db.update(agents).set({ adapterType: "claude_local" }).where(eq(agents.id, f.agentId));
|
||||
await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "legacy", nativeIssueId: null,
|
||||
processPid: process.pid,
|
||||
}).where(eq(heartbeatRuns.id, f.sourceRunId));
|
||||
await db.update(issueRecoveryActions).set({ cause: "legacy_execution_requires_reconciliation" })
|
||||
.where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
|
||||
await db.update(issueComments).set({ authorUserId: "original-author", createdAt: new Date("2026-09-11T09:00:00Z") })
|
||||
.where(eq(issueComments.id, f.commentId));
|
||||
const queueId = randomUUID();
|
||||
await db.insert(agentWakeupRequests).values({ id: queueId, companyId: f.companyId, agentId: f.agentId,
|
||||
source: "on_demand", reason: "issue_commented", status: "deferred_issue_execution",
|
||||
requestedByActorType: "user", requestedByActorId: "original-author",
|
||||
payload: { issueId: f.issueId, _paperclipWakeContext: { wakeCommentIds: [f.commentId] },
|
||||
queuedCommentInterrupt: { actorId: "board", requestedAt: new Date().toISOString() } },
|
||||
});
|
||||
const attempt = (queue = queueId) => db.transaction(async tx => {
|
||||
await tx.select().from(issues).where(eq(issues.id, f.issueId)).for("update");
|
||||
return admitExplicitNativeContinuation({ ...f, db: tx as unknown as typeof db,
|
||||
queuedCommentInterruptId: queue, dryRun: true });
|
||||
});
|
||||
expect(await attempt()).toBeNull();
|
||||
await db.update(heartbeatRuns).set({ processPid: 999999999 }).where(eq(heartbeatRuns.id, f.sourceRunId));
|
||||
expect(await attempt(randomUUID())).toBeNull();
|
||||
expect(await attempt()).toMatchObject({ previousRunId: f.sourceRunId, commentId: f.commentId });
|
||||
await db.update(agentWakeupRequests).set({ status: "cancelled" }).where(eq(agentWakeupRequests.id, queueId));
|
||||
expect(await attempt()).toBeNull();
|
||||
});
|
||||
it("dispatches another user's queued legacy message using the consumed board interrupt receipt", async () => {
|
||||
const f = await seed();
|
||||
await db.update(agents).set({ adapterType: "claude_local" }).where(eq(agents.id, f.agentId));
|
||||
await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "legacy", nativeIssueId: null })
|
||||
.where(eq(heartbeatRuns.id, f.sourceRunId));
|
||||
await db.update(issueRecoveryActions).set({ cause: "legacy_execution_requires_reconciliation" })
|
||||
.where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
|
||||
await db.update(issueComments).set({ authorUserId: "original-author", createdAt: new Date("2026-09-11T09:00:00Z") })
|
||||
.where(eq(issueComments.id, f.commentId));
|
||||
// Hold adapter startup so the test can exercise the real dispatch envelope
|
||||
// deterministically, without invoking a provider.
|
||||
await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" });
|
||||
const queueId = randomUUID();
|
||||
await db.insert(agentWakeupRequests).values({ id: queueId, companyId: f.companyId, agentId: f.agentId,
|
||||
source: "automation", reason: "issue_commented", status: "deferred_issue_execution",
|
||||
requestedByActorType: "system", payload: { issueId: f.issueId, commentId: f.commentId,
|
||||
_paperclipWakeContext: { wakeCommentIds: [f.commentId] },
|
||||
queuedCommentInterrupt: { actorId: "board", requestedAt: new Date().toISOString() } },
|
||||
});
|
||||
await heartbeatService(db).resumeQueuedCommentInterrupt(f.companyId, queueId);
|
||||
const [receipt] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, queueId));
|
||||
expect(receipt.status).toBe("coalesced");
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, receipt.runId!));
|
||||
const dispatch = (runId = run.id) => buildExecutionContinuation({ db, companyId: f.companyId,
|
||||
issueId: f.issueId, agentId: f.agentId, runId, context: run.contextSnapshot!,
|
||||
summary: null, exposeLowTrustRaw: false });
|
||||
const envelope = await dispatch();
|
||||
expect(envelope.interruptedRunId).toBe(f.sourceRunId);
|
||||
expect(envelope.originCommentIds).toContain(f.commentId);
|
||||
expect(envelope.messages).toEqual(expect.arrayContaining([expect.objectContaining({ id: f.commentId, body: "What happened?" })]));
|
||||
await expect(dispatch(randomUUID())).rejects.toThrow("continuation_user_authorization_missing");
|
||||
for (const patch of [
|
||||
{ status: "cancelled" }, { runId: f.sourceRunId },
|
||||
{ payload: { ...receipt.payload, issueId: randomUUID() } },
|
||||
{ payload: { ...receipt.payload, queuedCommentInterrupt: { actorId: "someone-else" } } },
|
||||
{ payload: { ...receipt.payload, _paperclipWakeContext: { wakeCommentIds: [] }, commentId: undefined } },
|
||||
]) {
|
||||
await db.update(agentWakeupRequests).set(patch).where(eq(agentWakeupRequests.id, queueId));
|
||||
await expect(dispatch()).rejects.toThrow("continuation_user_authorization_missing");
|
||||
await db.update(agentWakeupRequests).set({ status: receipt.status, runId: receipt.runId, payload: receipt.payload })
|
||||
.where(eq(agentWakeupRequests.id, queueId));
|
||||
}
|
||||
const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
|
||||
await db.update(issueRecoveryActions).set({ evidence: { ...action.evidence,
|
||||
explicitUserContinuation: { ...(action.evidence.explicitUserContinuation as Record<string, unknown>),
|
||||
queuedCommentInterruptId: "malformed-historical-receipt" },
|
||||
} }).where(eq(issueRecoveryActions.id, action.id));
|
||||
await expect(dispatch()).rejects.toThrow("continuation_user_authorization_missing");
|
||||
});
|
||||
const admit = (f: Fixture, dryRun = false) => db.transaction(async tx => {
|
||||
await tx.select().from(issues).where(eq(issues.id, f.issueId)).for("update");
|
||||
const result = await admitExplicitNativeContinuation({ ...f, dryRun, db: tx as unknown as typeof db });
|
||||
|
||||
@@ -5,7 +5,7 @@ import { hasRemoteTerminationReceipt, remoteLeaseCleanupScope } from "./remote-e
|
||||
import { z } from "zod";
|
||||
import { and, eq, inArray, isNull, ne, or, sql } from "drizzle-orm";
|
||||
import {
|
||||
agents, approvals, issueApprovals, issueThreadInteractions,
|
||||
agents, agentWakeupRequests, approvals, issueApprovals, issueThreadInteractions,
|
||||
environmentLeases, heartbeatRuns, issueComments, issueRecoveryActions,
|
||||
issues, nativeRunFinalizations, type Db,
|
||||
} from "@paperclipai/db";
|
||||
@@ -15,6 +15,7 @@ import { adapterExecutionControls } from "./adapter-execution-control.js";
|
||||
import { persistActivity } from "./activity-log.js";
|
||||
|
||||
import { historicalAdapterType, isConversationAdapter } from "./conversation-continuation.js";
|
||||
import { queuedCommentIdsFromWakePayload } from "./issue-queued-comment-queue.js";
|
||||
|
||||
type Run = typeof heartbeatRuns.$inferSelect;
|
||||
const terminal = ["failed", "interrupted", "timed_out", "cancelled"];
|
||||
@@ -33,6 +34,8 @@ export async function admitExplicitNativeContinuation(input: {
|
||||
actorType: string | null | undefined; actorId: string | null | undefined;
|
||||
reason: string | null; commentId: string | null; successorRunId: string;
|
||||
failedRunId?: string | null;
|
||||
/** Server-recorded board intent to send an existing legacy message queue. */
|
||||
queuedCommentInterruptId?: string;
|
||||
dryRun?: boolean;
|
||||
onBlocked?: (reason: string, message: string) => void;
|
||||
}): Promise<{ previousRunId: string; commentId: string | null; failedRunId?: string } | null> {
|
||||
@@ -47,16 +50,27 @@ export async function admitExplicitNativeContinuation(input: {
|
||||
eq(issues.companyId, companyId), eq(issues.id, issueId),
|
||||
));
|
||||
if (!task || task.assigneeAgentId !== agentId || ["done", "cancelled"].includes(task.status)) return null;
|
||||
const [interruptQueue] = input.queuedCommentInterruptId ? await db.select().from(agentWakeupRequests).where(and(
|
||||
eq(agentWakeupRequests.id, input.queuedCommentInterruptId),
|
||||
eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.agentId, agentId),
|
||||
eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
||||
sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
|
||||
sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt'->>'actorId' = ${actorId}`,
|
||||
)) : [];
|
||||
const queuedInterrupt = Boolean(interruptQueue && commentId &&
|
||||
queuedCommentIdsFromWakePayload(interruptQueue.payload).includes(commentId));
|
||||
if (input.queuedCommentInterruptId && !queuedInterrupt) return null;
|
||||
const [comment] = retry ? [] : await db.select().from(issueComments).where(and(
|
||||
eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId),
|
||||
eq(issueComments.id, commentId!), eq(issueComments.authorType, "user"),
|
||||
eq(issueComments.authorUserId, actorId), isNull(issueComments.createdByRunId),
|
||||
queuedInterrupt ? undefined : eq(issueComments.authorUserId, actorId), isNull(issueComments.createdByRunId),
|
||||
isNull(issueComments.deletedAt),
|
||||
));
|
||||
if (!retry && !comment?.body.trim()) return null;
|
||||
const authorizedAt = comment?.createdAt ?? new Date();
|
||||
const [agent] = await db.select().from(agents).where(and(eq(agents.companyId, companyId), eq(agents.id, agentId)));
|
||||
if (!agent || (!isConversationAdapter(agent.adapterType) && agent.adapterType !== "paperclip_runner")) return null;
|
||||
if (queuedInterrupt && !isConversationAdapter(agent.adapterType)) return null;
|
||||
const actions = await db.select().from(issueRecoveryActions).where(and(
|
||||
eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.sourceIssueId, issueId),
|
||||
executionBlockerPredicate(),
|
||||
@@ -86,7 +100,7 @@ export async function admitExplicitNativeContinuation(input: {
|
||||
if (!run || run.agentId !== agentId || !terminal.includes(run.status) ||
|
||||
(run.nativeIssueId ?? run.contextSnapshot?.issueId) !== issueId ||
|
||||
!run.finishedAt) return blocked("source_unavailable", "The previous execution has not finished or its owner changed. Your message is saved.");
|
||||
if (authorizedAt <= run.finishedAt) return blocked("message_predates_stop", "This message arrived before the previous run stopped. Send a new message to continue.");
|
||||
if (!queuedInterrupt && authorizedAt <= run.finishedAt) return blocked("message_predates_stop", "This message arrived before the previous run stopped. Send a new message to continue.");
|
||||
if (adapterExecutionControls.has(run.id)) return blocked("execution_settling", "Waiting for the previous run to stop. Your message will start automatically.");
|
||||
const unusedAdmission = run.status === "cancelled" && !run.startedAt &&
|
||||
run.errorCode === "execution_reconciliation_required" &&
|
||||
@@ -94,6 +108,7 @@ export async function admitExplicitNativeContinuation(input: {
|
||||
const legacyUserTurn = run.runtimeMode === "legacy" &&
|
||||
action.cause === "legacy_execution_requires_reconciliation" &&
|
||||
isConversationAdapter(agent.adapterType);
|
||||
if (queuedInterrupt && !legacyUserTurn) return null;
|
||||
if (legacyUserTurn) {
|
||||
const historicalAdapter = await historicalAdapterType(db, run);
|
||||
// A settings change never converts a known process/webhook execution into
|
||||
@@ -161,7 +176,8 @@ export async function admitExplicitNativeContinuation(input: {
|
||||
context: { previousRunId: previous.id, wakeCommentId: commentId },
|
||||
summary: null, exposeLowTrustRaw: false });
|
||||
if (input.dryRun) return { previousRunId: previous.id, commentId, ...(retry ? { failedRunId: input.failedRunId! } : {}) };
|
||||
const authorization = { actorId, commentId, ...(retry ? { failedRunId: input.failedRunId } : {}), runId: input.successorRunId,
|
||||
const authorization = { actorId, commentId, ...(retry ? { failedRunId: input.failedRunId } : {}),
|
||||
...(queuedInterrupt ? { queuedCommentInterruptId: input.queuedCommentInterruptId } : {}), runId: input.successorRunId,
|
||||
previousRunId: previous.id, recordedAt: new Date().toISOString() };
|
||||
for (const runId of cancelledStartupIds) {
|
||||
await db.update(nativeRunFinalizations).set({
|
||||
|
||||
@@ -2,7 +2,7 @@ import { AGENT_CHAT_DIRECTIVE, conversationReplay, isConversation, isConversatio
|
||||
import { PROCESS_IDENTITY_RECORDED, recordNativeLocalProcessStop } from "./native-local-process-stop.js";
|
||||
import { legacyControllerBootId, legacyControllerClaim, renewLegacyControllerLease, hasLiveLegacyController, revokeExpiredLegacyController, watchLegacyControllerLease } from "./legacy-controller-lease.js";
|
||||
import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js";
|
||||
import { remoteExecutionHasStopped, remoteTerminationReceipt, stoppedRemoteCleanupScopes } from "./remote-execution-termination.js";
|
||||
import { hasRemoteTerminationReceipt, remoteExecutionHasStopped, remoteTerminationReceipt, stoppedRemoteCleanupScopes } from "./remote-execution-termination.js";
|
||||
import { applyConnectorSkills, prepareConnectorSkillDelivery, resolveConnectorAssignments } from "./connector-runtime.js";
|
||||
import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js";
|
||||
import { connectionIntentService } from "./connection-intents.js";
|
||||
@@ -27,7 +27,7 @@ import { buildHeartbeatRunStatusLiveEventPayload } from "./heartbeat-run-status-
|
||||
export { buildHeartbeatRunStatusLiveEventPayload } from "./heartbeat-run-status-payload.js";
|
||||
import { buildExecutionContinuation } from "./execution-continuation.js";
|
||||
import { renderPaperclipWakePrompt } from "@paperclipai/adapter-utils/server-utils";
|
||||
import { initializeRunIdentity } from "./run-identity.js";
|
||||
import { initializeRunIdentity, explicitOperatorRunIdentity } from "./run-identity.js";
|
||||
import {
|
||||
assertDurableChatWakeupReceipt,
|
||||
assertDurableChatWakeupRequest,
|
||||
@@ -61,6 +61,7 @@ import {
|
||||
gte,
|
||||
inArray,
|
||||
isNull,
|
||||
isNotNull,
|
||||
lt,
|
||||
lte,
|
||||
ne,
|
||||
@@ -3501,6 +3502,10 @@ function normalizeMaxConcurrentRuns(value: unknown) {
|
||||
}
|
||||
|
||||
interface WakeupOptions {
|
||||
/** Set only by authenticated board wake routes; never copied from caller payloads. */
|
||||
manualUserWake?: boolean;
|
||||
/** Internal resume of a queue with persisted board interruption intent. */
|
||||
queuedCommentInterruptId?: string;
|
||||
/** Exact failed run selected by an authenticated board Retry request. */
|
||||
failedRunId?: string | null;
|
||||
durableChatRequest?: DurableChatWakeupRequest;
|
||||
@@ -10055,6 +10060,10 @@ export function heartbeatService(
|
||||
for (const wake of pending) {
|
||||
if (wake.idempotencyKey?.startsWith("chat-inbound:")) continue;
|
||||
const payload = parseObject(wake.payload);
|
||||
if (payload.queuedCommentInterrupt) {
|
||||
await resumeQueuedCommentInterrupt(wake.companyId, wake.id);
|
||||
continue;
|
||||
}
|
||||
const context = parseObject(payload[DEFERRED_WAKE_CONTEXT_KEY]);
|
||||
const commentId = deriveCommentId(context, payload);
|
||||
if (legacyContinuation) {
|
||||
@@ -10094,6 +10103,93 @@ export function heartbeatService(
|
||||
}
|
||||
}
|
||||
|
||||
async function resumeQueuedCommentInterrupt(companyId: string, queueId: string, opts?: { retryCleanup?: boolean }) {
|
||||
const [wake] = await db.select().from(agentWakeupRequests).where(and(
|
||||
eq(agentWakeupRequests.id, queueId), eq(agentWakeupRequests.companyId, companyId),
|
||||
eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
||||
));
|
||||
if (!wake) return;
|
||||
const payload = parseObject(wake.payload);
|
||||
const actorId = readNonEmptyString(parseObject(payload.queuedCommentInterrupt).actorId);
|
||||
const commentIds = queuedCommentIdsFromWakePayload(payload);
|
||||
const issueId = readNonEmptyString(payload.issueId);
|
||||
if (!actorId || !issueId || !commentIds.length) return;
|
||||
const agent = await getAgent(wake.agentId);
|
||||
if (!agent || agent.companyId !== companyId || agent.adapterType === "paperclip_runner") return;
|
||||
const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, companyId),
|
||||
sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`,
|
||||
inArray(heartbeatRuns.status, ["running", "queued", "scheduled_retry"]),
|
||||
)).limit(1);
|
||||
if (active) return;
|
||||
if (opts?.retryCleanup) {
|
||||
// Only the HTTP click grants an extra cleanup attempt. Periodic retries
|
||||
// reuse the intent to deliver, never a fresh provider teardown budget.
|
||||
const sourceRun = await db.transaction(async tx => {
|
||||
const [task] = await tx.select().from(issues).where(and(
|
||||
eq(issues.companyId, companyId), eq(issues.id, issueId),
|
||||
)).for("update");
|
||||
const [current] = await tx.select().from(agentWakeupRequests).where(and(
|
||||
eq(agentWakeupRequests.id, queueId), eq(agentWakeupRequests.companyId, companyId),
|
||||
eq(agentWakeupRequests.agentId, wake.agentId), eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
||||
sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
|
||||
sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt'->>'actorId' = ${actorId}`,
|
||||
));
|
||||
if (!task || task.assigneeAgentId !== wake.agentId || ["done", "cancelled"].includes(task.status) ||
|
||||
!current || !queuedCommentIdsFromWakePayload(current.payload).length) return null;
|
||||
const [successor] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, companyId),
|
||||
sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`,
|
||||
inArray(heartbeatRuns.status, ["running", "queued", "scheduled_retry"]),
|
||||
)).limit(1);
|
||||
if (successor) return null;
|
||||
const blocker = await getExecutionBlocker(tx as unknown as Db, companyId, issueId);
|
||||
const run = blocker?.runId ? await tx.select().from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, blocker.runId),
|
||||
eq(heartbeatRuns.agentId, wake.agentId), eq(heartbeatRuns.runtimeMode, "legacy"),
|
||||
inArray(heartbeatRuns.status, ["failed", "timed_out", "interrupted", "cancelled"]),
|
||||
sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`,
|
||||
)).then(rows => rows[0]) : null;
|
||||
if (!run || activeRunExecutions.has(run.id) || adapterExecutionControls.has(run.id)) return null;
|
||||
// Older ephemeral leases recorded successful cleanup without a provider
|
||||
// receipt. Re-verify them through the recorded teardown path; a timestamp
|
||||
// alone never certifies termination. Retained/reusable resources stay put.
|
||||
const historical = await tx.select().from(environmentLeases).where(and(
|
||||
eq(environmentLeases.companyId, companyId), eq(environmentLeases.heartbeatRunId, run.id),
|
||||
eq(environmentLeases.leasePolicy, "ephemeral"), isNotNull(environmentLeases.releasedAt),
|
||||
inArray(environmentLeases.status, ["released", "expired", "failed"]),
|
||||
)).for("update");
|
||||
for (const lease of historical) {
|
||||
if (!lease.provider || lease.provider === "local" || !lease.providerLeaseId || hasRemoteTerminationReceipt(lease)) continue;
|
||||
const [otherOwner] = await tx.select({ id: environmentLeases.id }).from(environmentLeases).where(and(
|
||||
ne(environmentLeases.id, lease.id), eq(environmentLeases.provider, lease.provider),
|
||||
eq(environmentLeases.providerLeaseId, lease.providerLeaseId),
|
||||
or(isNull(environmentLeases.releasedAt), inArray(environmentLeases.status, ["active", "retained", "pending_cleanup"])),
|
||||
)).limit(1);
|
||||
if (otherOwner) continue;
|
||||
await tx.update(environmentLeases).set({ status: "pending_cleanup", updatedAt: new Date() })
|
||||
.where(eq(environmentLeases.id, lease.id));
|
||||
}
|
||||
return run;
|
||||
});
|
||||
if (sourceRun) await sweepPendingCleanupLeases({ explicitRetry: {
|
||||
companyId, runId: sourceRun.id, actorId, reason: "queued_comment_interrupt",
|
||||
} });
|
||||
}
|
||||
const deliveryPayload = { ...payload };
|
||||
delete deliveryPayload.queuedCommentInterrupt;
|
||||
await enqueueWakeup(wake.agentId, {
|
||||
source: "on_demand", triggerDetail: "manual", reason: "issue_commented",
|
||||
payload: deliveryPayload, contextSnapshot: withQueuedCommentIdsInRunContext({
|
||||
issueId, triggeredBy: "board", actorId, responsibleUserId: actorId,
|
||||
}, commentIds),
|
||||
requestedByActorType: "user", requestedByActorId: actorId,
|
||||
queuedCommentInterruptId: queueId,
|
||||
issueStateGuard: { assigneeAgentId: wake.agentId, statuses: ["todo", "in_progress", "in_review", "blocked"] },
|
||||
idempotencyKey: `queued-comment-interrupt:${queueId}`,
|
||||
}, queueId);
|
||||
}
|
||||
|
||||
async function resumeExecutionWaitComments() {
|
||||
if ((await getSchedulingSuppression()).suppressed) return;
|
||||
const waits = await db.select({ wake: agentWakeupRequests })
|
||||
@@ -10108,6 +10204,7 @@ export function heartbeatService(
|
||||
))), eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
||||
eq(agentWakeupRequests.requestedByActorType, "user"),
|
||||
sql`${agentWakeupRequests.payload}->'executionWait' is not null`,
|
||||
sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt' is null`,
|
||||
lte(agentWakeupRequests.updatedAt, new Date(Date.now() - 30_000)),
|
||||
notInArray(issues.status, ["done", "cancelled"])))
|
||||
.orderBy(asc(agentWakeupRequests.updatedAt)).limit(50);
|
||||
@@ -10613,7 +10710,8 @@ export function heartbeatService(
|
||||
ReturnType<typeof getRoutineEnvForExecutionIssue>
|
||||
>;
|
||||
}) {
|
||||
const responsibleUserId = await resolveResponsibleUserIdForRunSeed({
|
||||
const operatorIdentity = await explicitOperatorRunIdentity(db, input.run);
|
||||
const responsibleUserId = operatorIdentity?.actorId ?? await resolveResponsibleUserIdForRunSeed({
|
||||
companyId: input.run.companyId,
|
||||
contextSnapshot: input.contextSnapshot,
|
||||
issueContext: input.issueContext,
|
||||
@@ -17834,7 +17932,7 @@ export function heartbeatService(
|
||||
* A later user Retry may try again after a provider failure; automatic
|
||||
* sweeps retain their exhausted budget and never gain extra attempts.
|
||||
*/
|
||||
explicitRetry?: { companyId: string; runId: string; actorId: string };
|
||||
explicitRetry?: { companyId: string; runId: string; actorId: string; reason?: "retry_failed_run" | "queued_comment_interrupt" };
|
||||
}): Promise<{
|
||||
swept: number;
|
||||
destroyed: number;
|
||||
@@ -17964,7 +18062,7 @@ export function heartbeatService(
|
||||
if (opts?.explicitRetry) await logActivity(db, {
|
||||
companyId: row.companyId, actorType: "user", actorId: opts.explicitRetry.actorId,
|
||||
action: "environment_lease.cleanup_retried", entityType: "environment_lease", entityId: row.id,
|
||||
runId: opts.explicitRetry.runId, details: { attempt: attempts + 1, reason: "retry_failed_run" },
|
||||
runId: opts.explicitRetry.runId, details: { attempt: attempts + 1, reason: opts.explicitRetry.reason ?? "retry_failed_run" },
|
||||
});
|
||||
|
||||
try {
|
||||
@@ -18676,6 +18774,52 @@ export function heartbeatService(
|
||||
if ((await getSchedulingSuppression()).suppressed) return;
|
||||
await resumeExecutionWaitComments();
|
||||
const cutoff = await getWorktreeExecutionCutoff();
|
||||
const pendingInterrupts = await db.select({ id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId })
|
||||
.from(agentWakeupRequests).innerJoin(companies, eq(companies.id, agentWakeupRequests.companyId))
|
||||
.where(and(eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
||||
eq(companies.status, "active"),
|
||||
sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt' is not null`,
|
||||
lte(agentWakeupRequests.updatedAt, new Date(Date.now() - 30_000)),
|
||||
cutoff ? gte(agentWakeupRequests.requestedAt, cutoff) : undefined))
|
||||
.orderBy(asc(agentWakeupRequests.updatedAt)).limit(50);
|
||||
for (const wake of pendingInterrupts) {
|
||||
await db.update(agentWakeupRequests).set({ updatedAt: new Date() }).where(and(
|
||||
eq(agentWakeupRequests.id, wake.id), eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
||||
));
|
||||
await resumeQueuedCommentInterrupt(wake.companyId, wake.id).catch(err => {
|
||||
logger.warn({ err, queueId: wake.id }, "failed to resume interrupted comment queue");
|
||||
});
|
||||
}
|
||||
// A server restart or a message/cleanup race can leave a deferred wake
|
||||
// after its owner has released the issue lock. Revisit it through the same
|
||||
// release admission, so recovery holds and operator Stops still apply.
|
||||
const strandedQueues = await db.select({ wake: agentWakeupRequests })
|
||||
.from(agentWakeupRequests)
|
||||
.innerJoin(issues, and(eq(issues.companyId, agentWakeupRequests.companyId),
|
||||
sql`${issues.id}::text = ${agentWakeupRequests.payload}->>'issueId'`,
|
||||
eq(issues.assigneeAgentId, agentWakeupRequests.agentId)))
|
||||
.innerJoin(companies, and(eq(companies.id, issues.companyId), eq(companies.status, "active")))
|
||||
.where(and(eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
||||
isNull(issues.executionRunId),
|
||||
sql`jsonb_typeof(${agentWakeupRequests.payload} #> '{_paperclipWakeContext,wakeCommentIds}') = 'array'`,
|
||||
sql`${agentWakeupRequests.payload} #> '{_paperclipWakeContext,wakeCommentIds}' <> '[]'::jsonb`,
|
||||
sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt' is null`,
|
||||
cutoff ? gte(agentWakeupRequests.requestedAt, cutoff) : undefined))
|
||||
.orderBy(asc(agentWakeupRequests.updatedAt)).limit(50);
|
||||
for (const { wake } of strandedQueues) {
|
||||
if (!queuedCommentIdsFromWakePayload(wake.payload).length) continue;
|
||||
const [latest] = await db.select().from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, wake.companyId), eq(heartbeatRuns.agentId, wake.agentId),
|
||||
sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${String(wake.payload?.issueId)}`,
|
||||
)).orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)).limit(1);
|
||||
await db.update(agentWakeupRequests).set({ updatedAt: new Date() }).where(and(
|
||||
eq(agentWakeupRequests.id, wake.id), eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
||||
));
|
||||
if (!latest || latest.runtimeMode !== "legacy" || !isHeartbeatRunTerminalStatus(latest.status)) continue;
|
||||
await releaseIssueExecutionAndPromote(latest, { suppressImmediateRecovery: true }).catch(err => {
|
||||
logger.warn({ err, queueId: wake.id }, "failed to promote stranded legacy comments");
|
||||
});
|
||||
}
|
||||
|
||||
// The cancellation marker is durable intent. Retry while its exact queue
|
||||
// is still deferred, including after a failed cleanup promotion or restart.
|
||||
@@ -25142,18 +25286,6 @@ export function heartbeatService(
|
||||
${JSON.stringify({ startupPreparationSettledAt: new Date().toISOString() })}::jsonb`,
|
||||
}).where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.status, "cancelled")));
|
||||
}
|
||||
// Interrupting a queued message explicitly authorizes the pending queue.
|
||||
// Retry its normal promotion after leases and adapter cleanup have settled;
|
||||
// the earlier terminal write can still have an execution blocker here.
|
||||
if (
|
||||
latestRun?.status === "cancelled" &&
|
||||
latestRun.runtimeMode !== "native" &&
|
||||
readNonEmptyString(latestRun.resultJson?.queuedCommentInterruptQueueId)
|
||||
) {
|
||||
await releaseIssueExecutionAndPromote(latestRun, { suppressImmediateRecovery: true }).catch((err) => {
|
||||
logger.error({ err, runId: run.id }, "failed to promote interrupted comment queue after cleanup");
|
||||
});
|
||||
}
|
||||
} finally {
|
||||
controllerLease.stop();
|
||||
activeRunExecutions.delete(run.id);
|
||||
@@ -25166,6 +25298,20 @@ export function heartbeatService(
|
||||
adapterExecutionControls.delete(run.id);
|
||||
}
|
||||
}
|
||||
// Terminalization precedes lease and adapter cleanup. Only now is the
|
||||
// owner gone; retry pending input for ordinary completions as well as Stop.
|
||||
if (latestRun?.runtimeMode === "legacy" && isHeartbeatRunTerminalStatus(latestRun.status)) {
|
||||
const [pending] = await db.select({ id: agentWakeupRequests.id, payload: agentWakeupRequests.payload }).from(agentWakeupRequests).where(and(
|
||||
eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId),
|
||||
eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
||||
sql`${agentWakeupRequests.payload}->>'issueId' = ${String(latestRun.contextSnapshot?.issueId)}`,
|
||||
)).limit(1);
|
||||
if (pending) await (pending.payload?.queuedCommentInterrupt
|
||||
? resumeQueuedCommentInterrupt(run.companyId, pending.id)
|
||||
: releaseIssueExecutionAndPromote(latestRun, { suppressImmediateRecovery: true })).catch(err => {
|
||||
logger.error({ err, runId: run.id }, "failed to promote legacy comment queue after cleanup");
|
||||
});
|
||||
}
|
||||
if (
|
||||
!nativeSessionResumeScheduled &&
|
||||
!nativeWorkspaceFinalizeScheduled &&
|
||||
@@ -25211,7 +25357,19 @@ export function heartbeatService(
|
||||
...(opts.contextSnapshot ?? {}),
|
||||
};
|
||||
const reason = opts.reason ?? null;
|
||||
const payload = opts.payload ?? null;
|
||||
let payload = opts.payload ? { ...opts.payload } : null;
|
||||
// Only the board queue route can record interruption authority on an
|
||||
// existing receipt. Never accept this internal marker from a wake caller.
|
||||
if (payload) {
|
||||
delete payload.queuedCommentInterrupt;
|
||||
delete payload.manualUserWake;
|
||||
}
|
||||
if (opts.manualUserWake) {
|
||||
if (opts.requestedByActorType !== "user" || !opts.requestedByActorId || opts.failedRunId) {
|
||||
throw new HttpError(403, "Manual wake requires an authenticated user");
|
||||
}
|
||||
payload = { ...payload, manualUserWake: true };
|
||||
}
|
||||
const executionReconciliationWake =
|
||||
contextSnapshot.source === "execution.reconciled" ||
|
||||
opts.idempotencyKey?.startsWith("execution-reconciliation:") === true;
|
||||
@@ -25236,6 +25394,9 @@ export function heartbeatService(
|
||||
if (issueId) {
|
||||
const conversation = await getIssueExecutionContext(agent.companyId, issueId);
|
||||
if (isConversation(conversation)) {
|
||||
if (opts.manualUserWake && conversation!.conversationUserId !== opts.requestedByActorId) {
|
||||
throw new HttpError(403, "Only the conversation owner can start a chat run");
|
||||
}
|
||||
if (isConversationExecutionWake(conversation, reason ?? readNonEmptyString(enrichedContextSnapshot.wakeReason))) return null;
|
||||
if (agent.id !== conversation!.conversationAgentId) return null;
|
||||
if (!(await instanceSettings.getExperimental()).enableAgentChat) return null;
|
||||
@@ -25557,8 +25718,10 @@ export function heartbeatService(
|
||||
const isolatedWorkspacesEnabled = issueId
|
||||
? (await instanceSettings.getExperimental()).enableIsolatedWorkspaces
|
||||
: false;
|
||||
let operatorResponsibleUserId: string | null = opts.manualUserWake ? opts.requestedByActorId! : null;
|
||||
let queuedResponsibleUserIdPromise: Promise<string> | null = null;
|
||||
const resolveQueuedResponsibleUserId = () => {
|
||||
if (operatorResponsibleUserId) return Promise.resolve(operatorResponsibleUserId);
|
||||
queuedResponsibleUserIdPromise ??= (async () => {
|
||||
const queuedIssueContext = issueId
|
||||
? await getIssueExecutionContext(agent.companyId, issueId)
|
||||
@@ -25738,8 +25901,13 @@ export function heartbeatService(
|
||||
const [pending] = await tx.select().from(agentWakeupRequests).where(and(
|
||||
eq(agentWakeupRequests.id, executionWaitRequestId), eq(agentWakeupRequests.companyId, agent.companyId),
|
||||
eq(agentWakeupRequests.agentId, agentId), eq(agentWakeupRequests.status, "deferred_issue_execution"),
|
||||
eq(agentWakeupRequests.requestedByActorType, "user"),
|
||||
eq(agentWakeupRequests.requestedByActorId, opts.requestedByActorId ?? ""),
|
||||
// A user message can join a queue originally created by a
|
||||
// system wake. The recorded board click supplies fresh authority.
|
||||
opts.queuedCommentInterruptId === executionWaitRequestId
|
||||
? undefined : eq(agentWakeupRequests.requestedByActorType, "user"),
|
||||
opts.queuedCommentInterruptId === executionWaitRequestId
|
||||
? sql`${agentWakeupRequests.payload}->'queuedCommentInterrupt'->>'actorId' = ${opts.requestedByActorId ?? ""}`
|
||||
: eq(agentWakeupRequests.requestedByActorId, opts.requestedByActorId ?? ""),
|
||||
sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
|
||||
));
|
||||
// The issue lock serializes cleanup callbacks and periodic workers.
|
||||
@@ -25747,6 +25915,21 @@ export function heartbeatService(
|
||||
if (!pending || !wakeCommentId || !queuedCommentIdsFromWakePayload(pending.payload).includes(wakeCommentId)) {
|
||||
return { kind: "deferred" as const };
|
||||
}
|
||||
if (!opts.queuedCommentInterruptId && pending.payload?.manualUserWake === true) {
|
||||
// A persisted manual wake keeps its actor when an execution wait
|
||||
// resumes. The locked receipt above has revalidated that actor.
|
||||
payload = { ...payload, manualUserWake: true };
|
||||
operatorResponsibleUserId = opts.requestedByActorId!;
|
||||
}
|
||||
if (opts.queuedCommentInterruptId) {
|
||||
// The locked board receipt supplies execution authority even when
|
||||
// another user authored the messages. Dispatch revalidates the receipt.
|
||||
operatorResponsibleUserId = opts.requestedByActorId!;
|
||||
// Edits/discards between the click and dispatch remain authoritative.
|
||||
Object.assign(enrichedContextSnapshot, withQueuedCommentIdsInRunContext(
|
||||
enrichedContextSnapshot, queuedCommentIdsFromWakePayload(pending.payload),
|
||||
));
|
||||
}
|
||||
}
|
||||
let automaticParentRunId: string | null = null;
|
||||
if (
|
||||
@@ -26124,6 +26307,7 @@ export function heartbeatService(
|
||||
db: tx as unknown as Db, companyId: issue.companyId, issueId: issue.id,
|
||||
agentId, actorType: opts.requestedByActorType, actorId: opts.requestedByActorId,
|
||||
reason, commentId: wakeCommentId ?? null, failedRunId: opts.failedRunId, successorRunId: explicitContinuationRunId,
|
||||
queuedCommentInterruptId: opts.queuedCommentInterruptId,
|
||||
dryRun: true,
|
||||
onBlocked: (reason, message) => { continuationWait = { reason, message }; },
|
||||
}))) return deferBlockedExecution(executionBlocker);
|
||||
@@ -26886,6 +27070,7 @@ export function heartbeatService(
|
||||
db: tx as unknown as Db, companyId: issue.companyId, issueId: issue.id,
|
||||
agentId, actorType: opts.requestedByActorType, actorId: opts.requestedByActorId,
|
||||
reason, commentId: wakeCommentId ?? null, failedRunId: opts.failedRunId, successorRunId: explicitContinuationRunId,
|
||||
queuedCommentInterruptId: opts.queuedCommentInterruptId,
|
||||
});
|
||||
if (!explicitContinuation && executionBlocker) return deferBlockedExecution(executionBlocker);
|
||||
if (explicitContinuation) {
|
||||
@@ -26929,6 +27114,7 @@ export function heartbeatService(
|
||||
.orderBy(asc(agentWakeupRequests.requestedAt))
|
||||
: [];
|
||||
const adoptedComments = pendingComments.filter((wake) => {
|
||||
if (wake.id === opts.queuedCommentInterruptId) return true;
|
||||
const deferredPayload = parseObject(wake.payload);
|
||||
const deferredContext = parseObject(
|
||||
deferredPayload[DEFERRED_WAKE_CONTEXT_KEY],
|
||||
@@ -27103,8 +27289,9 @@ export function heartbeatService(
|
||||
contextSnapshot: enrichedContextSnapshot,
|
||||
wakeCommentId,
|
||||
});
|
||||
// Unscoped manual wakes need their own receipt and execution identity too.
|
||||
const rawCoalescedTarget =
|
||||
opts.allowRunCoalescing === false
|
||||
opts.allowRunCoalescing === false || opts.manualUserWake
|
||||
? null
|
||||
: (sameScopeQueuedRun ??
|
||||
sameScopeScheduledRetryRun ??
|
||||
@@ -28508,6 +28695,7 @@ export function heartbeatService(
|
||||
|
||||
releaseEnvironmentLeasesForRun,
|
||||
resumeRemoteStopComments,
|
||||
resumeQueuedCommentInterrupt,
|
||||
resumeExecutionWaitComments,
|
||||
|
||||
sweepStaleIssueLocks,
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { and, asc, desc, eq, inArray, sql } from "drizzle-orm";
|
||||
import {
|
||||
agentWakeupRequests,
|
||||
heartbeatRuns,
|
||||
heartbeatRunEvents,
|
||||
issueComments,
|
||||
@@ -10,6 +11,48 @@ import {
|
||||
} from "@paperclipai/db";
|
||||
import { conflict, forbidden } from "../errors.js";
|
||||
import { isUuidLike } from "@paperclipai/shared";
|
||||
import { queuedCommentIdsFromRunContext, queuedCommentIdsFromWakePayload } from "./issue-queued-comment-queue.js";
|
||||
|
||||
/** Resolve an explicit click from persisted receipts, never caller context or message authors. */
|
||||
export async function explicitOperatorRunIdentity(
|
||||
executor: Pick<Db, "select">,
|
||||
run: Pick<typeof heartbeatRuns.$inferSelect, "id" | "companyId" | "agentId" | "contextSnapshot" | "wakeupRequestId">,
|
||||
) {
|
||||
const [request] = run.wakeupRequestId ? await executor.select().from(agentWakeupRequests).where(and(
|
||||
eq(agentWakeupRequests.id, run.wakeupRequestId), eq(agentWakeupRequests.companyId, run.companyId),
|
||||
eq(agentWakeupRequests.agentId, run.agentId), eq(agentWakeupRequests.runId, run.id),
|
||||
)) : [];
|
||||
if (request?.payload?.manualUserWake === true) {
|
||||
if (request.requestedByActorType !== "user" || !request.requestedByActorId) {
|
||||
throw forbidden("Manual wake requires an authenticated user");
|
||||
}
|
||||
return { actorId: request.requestedByActorId, cause: "manual_user_wake" };
|
||||
}
|
||||
const prefix = "queued-comment-interrupt:";
|
||||
if (!request?.idempotencyKey?.startsWith(prefix)) return null;
|
||||
const queueId = request.idempotencyKey.slice(prefix.length);
|
||||
// The key only locates a candidate. The consumed queue, actor, run, company,
|
||||
// agent, task, and delivered messages must all independently agree.
|
||||
if (!isUuidLike(queueId)) {
|
||||
throw forbidden("Queued-message interrupt authority is unavailable");
|
||||
}
|
||||
const [receipt] = await executor.select().from(agentWakeupRequests).where(and(
|
||||
eq(agentWakeupRequests.id, queueId), eq(agentWakeupRequests.companyId, run.companyId),
|
||||
eq(agentWakeupRequests.agentId, run.agentId), eq(agentWakeupRequests.runId, run.id),
|
||||
eq(agentWakeupRequests.status, "coalesced"),
|
||||
));
|
||||
const marker = receipt?.payload?.queuedCommentInterrupt;
|
||||
const actorId = marker && typeof marker === "object" && "actorId" in marker ? marker.actorId : null;
|
||||
const ids = queuedCommentIdsFromWakePayload(receipt?.payload);
|
||||
const deliveredIds = queuedCommentIdsFromRunContext(run.contextSnapshot);
|
||||
if (typeof actorId !== "string" || !actorId || !ids.length ||
|
||||
receipt?.payload?.issueId !== run.contextSnapshot?.issueId ||
|
||||
!ids.every(id => deliveredIds.includes(id)) ||
|
||||
request.requestedByActorType !== "user" || request.requestedByActorId !== actorId) {
|
||||
throw forbidden("Queued-message interrupt authority is unavailable");
|
||||
}
|
||||
return { actorId, cause: "queued_comment_interrupt" };
|
||||
}
|
||||
|
||||
export type RunIdentityContext = typeof runIdentityContexts.$inferSelect;
|
||||
type Executor = Pick<Db, "select" | "insert" | "update">;
|
||||
@@ -147,6 +190,7 @@ export async function initializeRunIdentity(
|
||||
.where(eq(runIdentityContexts.id, run.activeIdentityContextId));
|
||||
return current!;
|
||||
}
|
||||
const operatorIdentity = await explicitOperatorRunIdentity(tx, run);
|
||||
const [parent] = input.parentRunId
|
||||
? await tx
|
||||
.select()
|
||||
@@ -171,10 +215,9 @@ export async function initializeRunIdentity(
|
||||
),
|
||||
)
|
||||
: [];
|
||||
const parentId =
|
||||
interaction?.sourceIdentityContextId ??
|
||||
input.parentContextId ??
|
||||
parent?.activeIdentityContextId;
|
||||
const parentId = operatorIdentity ? null : (
|
||||
interaction?.sourceIdentityContextId ?? input.parentContextId ?? parent?.activeIdentityContextId
|
||||
);
|
||||
const [origin] = parentId
|
||||
? await tx
|
||||
.select()
|
||||
@@ -192,12 +235,10 @@ export async function initializeRunIdentity(
|
||||
let current = await append(tx, {
|
||||
companyId: input.companyId,
|
||||
runId: input.runId,
|
||||
responsibleUserId: origin
|
||||
? origin.responsibleUserId
|
||||
: input.responsibleUserId,
|
||||
responsibleUserId: operatorIdentity?.actorId ?? (origin ? origin.responsibleUserId : input.responsibleUserId),
|
||||
parentContextId: origin?.id ?? null,
|
||||
cause:
|
||||
origin?.cause === "company_default" ? "company_default" : input.cause,
|
||||
operatorIdentity ? operatorIdentity.cause : origin?.cause === "company_default" ? "company_default" : input.cause,
|
||||
correlationId: "dispatch",
|
||||
});
|
||||
const ids = [...new Set(input.messageIds ?? [])];
|
||||
@@ -220,10 +261,10 @@ export async function initializeRunIdentity(
|
||||
current = await append(tx, {
|
||||
companyId: input.companyId,
|
||||
runId: input.runId,
|
||||
responsibleUserId: comment.authorUserId,
|
||||
responsibleUserId: operatorIdentity?.actorId ?? comment.authorUserId,
|
||||
messageId: id,
|
||||
parentContextId: current.id,
|
||||
cause: "instruction",
|
||||
cause: operatorIdentity?.cause ?? "instruction",
|
||||
correlationId: `message:${id}`,
|
||||
});
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user