mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 01:50:13 +01:00
fix(queue): clear interaction receipts a run already delivered (#19376)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Agents ask people for input through thread cards: questions, confirmations, plan approvals and connection requests > - When a person resolves a card, Paperclip queues the response for the agent. The task chat shows it above the composer as a read-only receipt, for example "Answered: Questions" > - A second wake can carry the same receipt. One example is a recovery continuation that raced the response's own wake. The agent receives the response from the first run, but the second wake stays deferred > - The composer then shows a receipt that the agent already has. The user cannot edit, discard or clear it. The same receipt also blocks the agent's handback with `interaction_response_queued` > - This pull request treats a receipt as pending only until a run that carries the interaction has started > - The benefit is that an answered card leaves the queue once the agent has it, for every card kind ## Linked Issues or Issue Description No public issue exists. A hands-on onboarding test on a staging cloud workspace (Claude subscription path) found this bug. **What happened?** During onboarding, a queued item stayed above the task-chat composer. It said "Answered: Questions", but the user had already answered the questions. Later items also stayed after the interview answers and after the user connected GitHub. The user could not trash, steer or edit them. Evidence from the staging workspace (interactions, wake diagnostics and run records for the onboarding task): - The first questions card (no title, so its label is "Answered: Questions") was answered at 17:37:48. - Run `ba3bc7d6` (`issue.interaction.respond`) delivered that answer from 17:37:48 to 17:38:27. - A recovery wake (`issue.interaction_continuation_recovery`) for the same card was requested at 17:37:48 too. It stayed deferred behind `ba3bc7d6` and then ran again as `cacb095d` (17:38:33 to 17:39:17). During that time the composer showed the stale "Answered: Questions" receipt. - The plan approval had the same pattern from 17:42:06 to 18:20 (fixed at its source by #19332). The declined Slack connection receipt sat behind that chain. **Expected behavior** A receipt leaves the composer queue as soon as a run has received it. The agent can hand back the task after it has received the response. **Steps to reproduce** 1. An agent asks a question card on a task. 2. A user answers it. The response wake starts a run that carries the answer. 3. A second wake for the same card is deferred behind that run or a later run. 4. `GET /api/issues/:id/queued-comments` returns the second wake's receipt as an immutable entry until it runs. ## What Changed - `server/src/services/queued-interaction-response.ts`: `readQueuedInteractionResponse` returns no receipt when a run for the same issue has started with this interaction in its context and is `running` or `succeeded`. A cancelled or failed run does not count, so that response is still owed. `hasQueuedInteractionResponse` (the handback guard) uses the same rule. - `server/src/routes/issues.ts`: `findQueuedCommentWake` skips a delivered receipt, so a later undelivered receipt can still show. - `server/src/__tests__/issue-queued-comments-routes.test.ts`: red/green tests for every card kind and outcome: answered questions, accepted and rejected confirmation, accepted plan (checkbox) approval, connected and declined connection. Each runs with a `succeeded` and with a `running` delivering run. Controls check that an undelivered receipt still shows, that a later undelivered receipt shows instead of a delivered one, and that the handback guard ignores a delivered receipt. - N-02 (hover shows raw JSON): `master` already shows only the one-line summary in the label and hover (`queuedMessageLabel`). This PR does not change the UI. ## Verification Red, on `master` code with only the new tests added: ``` $ cd server && npx vitest run src/__tests__/issue-queued-comments-routes.test.ts -t "receipt" × clears a ask_user_questions answered receipt once a succeeded run already delivered it 1309ms × clears a ask_user_questions answered receipt once a running run already delivered it 1041ms × clears a request_confirmation accepted receipt once a succeeded run already delivered it 1436ms × clears a request_confirmation accepted receipt once a running run already delivered it 829ms × clears a request_confirmation rejected receipt once a succeeded run already delivered it 666ms × clears a request_confirmation rejected receipt once a running run already delivered it 776ms × clears a request_checkbox_confirmation accepted receipt once a succeeded run already delivered it 1179ms × clears a request_checkbox_confirmation accepted receipt once a running run already delivered it 4227ms × clears a connection_intent accepted receipt once a succeeded run already delivered it 1954ms × clears a connection_intent accepted receipt once a running run already delivered it 1634ms × clears a connection_intent rejected receipt once a succeeded run already delivered it 1505ms × clears a connection_intent rejected receipt once a running run already delivered it 1064ms × shows a later undelivered receipt instead of an earlier delivered one 480ms × does not hold agent handback for a receipt the agent already received 1272ms ⎯⎯⎯⎯⎯⎯ Failed Tests 14 ⎯⎯⎯⎯⎯⎯⎯ AssertionError: expected [ 'Answered: Questions' ] to deeply equal [] AssertionError: expected [ 'Accepted: Approve hiring 2 agents' ] to deeply equal [] AssertionError: expected [ 'Rejected: Approve hiring 2 agents' ] to deeply equal [] AssertionError: expected [ Array(1) ] to deeply equal [] AssertionError: expected [ 'Connected: GitHub connection' ] to deeply equal [] AssertionError: expected [ 'Declined: Slack connection' ] to deeply equal [] AssertionError: expected [ Array(1) ] to deeply equal [ 'Declined: Slack connection' ] AssertionError: expected true to be false // Object.is equality Tests 14 failed | 11 passed | 85 skipped (110) ``` Green, with the fix: ``` $ cd server && npx vitest run src/__tests__/issue-queued-comments-routes.test.ts -t "receipt" Tests 25 passed | 85 skipped (110) ``` Neighbouring suites, with the fix: - `issue-queued-comments-routes.test.ts` (full file, on current `master`): 110 passed. - `question-response-delivery`, `queued-comment-consumption`, `issue-thread-interaction-routes`, `satisfied-connection-intents`, `issue-comment-reopen-routes`, `issue-execution-policy-routes` together with the queued-comments file: 494 tests passed. One run hit an embedded-Postgres `afterAll` cleanup timeout under 7-file parallel load. The same file passed alone twice. - `heartbeat-process-recovery.test.ts -t "plan approval|accepted interaction|answered question|answered interaction|onboarding first task"`: 14 passed. - `pnpm --filter @paperclipai/server typecheck`: passes. ## Risks - Low risk. The change only hides a receipt that a started run already carries. An undelivered receipt keeps its current behavior. - The redundant wake itself is not cancelled here. It still runs later as a normal continuation, but it no longer holds the composer queue or blocks handback. #19332 removes the main source of these duplicate wakes. - Each receipt read adds one indexed-by-company lookup on `heartbeat_runs` for the issue. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used in Claude Code with tool use (shell, file edits, browser inspection of the staging workspace API). ## 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 - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Opus 5.5 authored and GitHub committed
1 parent
434d224c3a
commit
de1143448b
3 files changed
+184
-10
No files matched your search
@@ -29,6 +29,7 @@ import { issueRoutes } from "../routes/issues.js";
|
||||
import { heartbeatService } from "../services/heartbeat.js";
|
||||
import { issueService } from "../services/issues.js";
|
||||
import { issueThreadInteractionService } from "../services/issue-thread-interactions.js";
|
||||
import { hasQueuedInteractionResponse } from "../services/queued-interaction-response.js";
|
||||
import { remoteTerminationReceipt } from "../services/remote-execution-termination.js";
|
||||
import { initializeRunIdentity, reconcileSteeredIdentity } from "../services/run-identity.js";
|
||||
import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js";
|
||||
@@ -668,6 +669,136 @@ describeEmbeddedPostgres("issue queued-comment routes", () => {
|
||||
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)))[0].status).toBe("running");
|
||||
});
|
||||
|
||||
// N-01: a redundant continuation wake (for example a recovery sweep that raced
|
||||
// the response's own wake) carries a receipt the agent already received. It
|
||||
// must not sit in the composer queue as an immutable, unremovable item.
|
||||
const deliveredReceiptCases = [
|
||||
{ kind: "ask_user_questions", status: "answered", title: null, label: "Answered: Questions",
|
||||
payload: { version: 1, questions: [{ id: "goal", prompt: "What are you building?", selectionMode: "single", required: true,
|
||||
options: [{ id: "shop", label: "A shop" }] }] }, result: { version: 1, answers: [{ questionId: "goal", optionIds: ["shop"] }] } },
|
||||
{ kind: "request_confirmation", status: "accepted", title: "Approve hiring 2 agents", label: "Accepted: Approve hiring 2 agents",
|
||||
payload: { version: 1, prompt: "Hire 2 agents?" }, result: { version: 1, outcome: "accepted" } },
|
||||
{ kind: "request_confirmation", status: "rejected", title: "Approve hiring 2 agents", label: "Rejected: Approve hiring 2 agents",
|
||||
payload: { version: 1, prompt: "Hire 2 agents?" }, result: { version: 1, outcome: "rejected" } },
|
||||
{ kind: "request_checkbox_confirmation", status: "accepted", title: "Approve storefront prototype plan",
|
||||
label: "Accepted: Approve storefront prototype plan",
|
||||
payload: { version: 1, prompt: "Accept scope", options: [{ id: "scope", label: "Build the app" }] },
|
||||
result: { version: 1, outcome: "accepted", selectedOptionIds: ["scope"] } },
|
||||
{ kind: "connection_intent", status: "accepted", title: "Connect GitHub", label: "Connected: GitHub connection",
|
||||
payload: { version: 1, serviceName: "GitHub" }, result: { version: 1, outcome: "connected" } },
|
||||
{ kind: "connection_intent", status: "rejected", title: "Connect Slack", label: "Declined: Slack connection",
|
||||
payload: { version: 1, serviceName: "Slack" }, result: { version: 1, outcome: "declined" } },
|
||||
] as const;
|
||||
|
||||
async function seedDeliveredReceipt(
|
||||
c: (typeof deliveredReceiptCases)[number],
|
||||
delivering: "succeeded" | "running" | "failed" | "cancelled" | null,
|
||||
runContext: "response" | "card_creation" = "response",
|
||||
) {
|
||||
const seeded = await seedQueue();
|
||||
await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
||||
const interactionId = randomUUID();
|
||||
const resolvedAt = new Date("2026-08-22T15:05:00.000Z");
|
||||
await db.insert(issueThreadInteractions).values({
|
||||
id: interactionId, companyId: seeded.companyId, issueId: seeded.issueId,
|
||||
kind: c.kind, status: c.status, createdByAgentId: seeded.agentId, sourceRunId: seeded.runId,
|
||||
resolvedByUserId: "queue-owner", resolvedAt,
|
||||
continuationPolicy: "wake_assignee", requestedResolverPolicy: "human_only",
|
||||
effectiveResolverPolicy: "human_only", resolverPolicyProvenance: "explicit",
|
||||
// Fixture payloads cover several interaction kinds; the schema types each kind separately.
|
||||
title: c.title, payload: c.payload as never, result: c.result as never,
|
||||
});
|
||||
const context = { issueId: seeded.issueId, interactionId, interactionKind: c.kind, interactionStatus: c.status };
|
||||
// The response's own wake already started a run that carries the answer.
|
||||
const deliveringRunId = randomUUID();
|
||||
if (delivering) {
|
||||
await db.insert(heartbeatRuns).values({
|
||||
id: deliveringRunId, companyId: seeded.companyId, agentId: seeded.agentId,
|
||||
invocationSource: "automation", triggerDetail: "system", status: delivering, runtimeMode: "native",
|
||||
startedAt: new Date("2026-08-22T15:05:01.000Z"),
|
||||
finishedAt: delivering === "running" ? null : new Date("2026-08-22T15:06:00.000Z"),
|
||||
// The card-creation wake carries the same interaction id while the
|
||||
// card is still pending, without a resolved status.
|
||||
contextSnapshot: runContext === "card_creation"
|
||||
? { issueId: seeded.issueId, interactionId, interactionKind: c.kind, wakeReason: "interaction_pending" }
|
||||
: { ...context, source: "issue.interaction.respond", wakeReason: "issue_commented" },
|
||||
});
|
||||
}
|
||||
// The redundant continuation, still deferred behind the issue's active run.
|
||||
const staleWakeId = randomUUID();
|
||||
await db.insert(agentWakeupRequests).values({
|
||||
id: staleWakeId, companyId: seeded.companyId, agentId: seeded.agentId,
|
||||
source: "automation", triggerDetail: "system", reason: "issue_continuation_needed",
|
||||
status: "deferred_issue_execution", requestedByActorType: "system", requestedByActorId: "recovery",
|
||||
requestedAt: new Date("2026-08-22T15:05:02.000Z"),
|
||||
payload: { issueId: seeded.issueId, mutation: "interaction", interactionId, interactionStatus: c.status,
|
||||
_paperclipWakeContext: { ...context, source: "issue.interaction_continuation_recovery" } },
|
||||
});
|
||||
return { ...seeded, interactionId, staleWakeId, deliveringRunId };
|
||||
}
|
||||
|
||||
it.each(deliveredReceiptCases.flatMap(c => (["succeeded", "running"] as const).map(delivering => ({ ...c, delivering }))))(
|
||||
"clears a $kind $status receipt once a $delivering run already delivered it",
|
||||
async (c) => {
|
||||
const seeded = await seedDeliveredReceipt(c, c.delivering);
|
||||
const client = app(seeded.companyId);
|
||||
const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
||||
expect(queue.body.entries.map((entry: { source?: { summary?: string } }) => entry.source?.summary)).toEqual([]);
|
||||
},
|
||||
);
|
||||
|
||||
it.each(deliveredReceiptCases)("keeps a pending $kind $status receipt until a run delivers it", async (c) => {
|
||||
const seeded = await seedDeliveredReceipt(c, null);
|
||||
const queue = await request(app(seeded.companyId)).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
||||
expect(queue.body.entries).toHaveLength(1);
|
||||
expect(queue.body.entries[0]).toMatchObject({
|
||||
source: { kind: "interaction", interactionId: seeded.interactionId, interactionKind: c.kind, summary: c.label },
|
||||
canEdit: false, canDiscard: false,
|
||||
});
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ delivering: "failed", runContext: "response" },
|
||||
{ delivering: "cancelled", runContext: "response" },
|
||||
{ delivering: "succeeded", runContext: "card_creation" },
|
||||
{ delivering: "running", runContext: "card_creation" },
|
||||
] as const)("keeps a receipt owed after a $delivering $runContext run", async ({ delivering, runContext }) => {
|
||||
const seeded = await seedDeliveredReceipt(deliveredReceiptCases[0], delivering, runContext);
|
||||
const queue = await request(app(seeded.companyId)).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
||||
expect(queue.body.entries.map((entry: { source?: { summary?: string } }) => entry.source?.summary))
|
||||
.toEqual(["Answered: Questions"]);
|
||||
expect(await hasQueuedInteractionResponse(db, seeded.companyId, seeded.issueId, seeded.agentId)).toBe(true);
|
||||
});
|
||||
|
||||
it("shows a later undelivered receipt instead of an earlier delivered one", async () => {
|
||||
const seeded = await seedDeliveredReceipt(deliveredReceiptCases[3], "succeeded");
|
||||
const pendingId = randomUUID();
|
||||
await db.insert(issueThreadInteractions).values({
|
||||
id: pendingId, companyId: seeded.companyId, issueId: seeded.issueId,
|
||||
kind: "connection_intent", status: "rejected", createdByAgentId: seeded.agentId,
|
||||
resolvedByUserId: "queue-owner", resolvedAt: new Date("2026-08-22T15:10:00.000Z"),
|
||||
continuationPolicy: "wake_assignee", requestedResolverPolicy: "human_only",
|
||||
effectiveResolverPolicy: "human_only", resolverPolicyProvenance: "explicit",
|
||||
title: "Connect Slack", payload: { version: 1, serviceName: "Slack" } as never, result: { version: 1, outcome: "declined" } as never,
|
||||
});
|
||||
await db.insert(agentWakeupRequests).values({
|
||||
companyId: seeded.companyId, agentId: seeded.agentId, source: "automation", triggerDetail: "system",
|
||||
reason: "issue_commented", status: "deferred_issue_execution", requestedAt: new Date("2026-08-22T16:00:00.000Z"),
|
||||
payload: { issueId: seeded.issueId, mutation: "interaction", interactionId: pendingId, interactionStatus: "rejected" },
|
||||
});
|
||||
const queue = await request(app(seeded.companyId)).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
||||
expect(queue.body.entries.map((entry: { source?: { summary?: string } }) => entry.source?.summary))
|
||||
.toEqual(["Declined: Slack connection"]);
|
||||
});
|
||||
|
||||
it("does not hold agent handback for a receipt the agent already received", async () => {
|
||||
const pending = await seedDeliveredReceipt(deliveredReceiptCases[1], null);
|
||||
expect(await hasQueuedInteractionResponse(db, pending.companyId, pending.issueId, pending.agentId)).toBe(true);
|
||||
await db.execute(sql`TRUNCATE TABLE companies CASCADE`);
|
||||
const delivered = await seedDeliveredReceipt(deliveredReceiptCases[1], "succeeded");
|
||||
expect(await hasQueuedInteractionResponse(db, delivered.companyId, delivered.issueId, delivered.agentId)).toBe(false);
|
||||
});
|
||||
|
||||
it("promotes an approval after normal completion without a click", async () => {
|
||||
const seeded = await seedResponseQueue();
|
||||
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, seeded.runId));
|
||||
|
||||
@@ -42,7 +42,7 @@ import { resumeNativeWorkspaceAfterOwnerStop } from "../services/native-runtime/
|
||||
import { retryNativeWorkspaceExport } from "../services/native-runtime/native-workspace-export-retry.js";
|
||||
import { reconstructNativeWorkspace, unreconciledNativeExportPredicate } from "../services/native-runtime/native-workspace-reconstruction.js";
|
||||
import { reconstructWorkspaceSchema } from "@paperclipai/shared";
|
||||
import { queuedInteractionId, readQueuedInteractionResponse, hasQueuedInteractionResponse } from "../services/queued-interaction-response.js";
|
||||
import { queuedInteractionId, readQueuedInteractionResponse, hasQueuedInteractionResponse, type QueuedInteractionResponse } from "../services/queued-interaction-response.js";
|
||||
import { deliverConversationComments, isConversation } from "../services/agent-conversations.js";
|
||||
import { assertOnboardingFirstTaskDependenciesAllowed } from "../services/onboarding-first-task-admission.js";
|
||||
import { resolveChatTaskAssignment } from "../services/chat-task-assignment.js";
|
||||
@@ -7403,6 +7403,8 @@ export function issueRoutes(
|
||||
wake: Pick<IssueQueueWake, "id" | "companyId" | "payload" | "status" | "runId">;
|
||||
state: "deferred" | "queued";
|
||||
queueRun: Pick<IssueQueueRun, "runtimeMode"> | null;
|
||||
/** The receipt already read for an interaction wake, reused by the queue builder. */
|
||||
response?: QueuedInteractionResponse | null;
|
||||
};
|
||||
|
||||
async function findQueuedCommentWake(
|
||||
@@ -7454,8 +7456,14 @@ export function issueRoutes(
|
||||
continue;
|
||||
const isReceipt = Boolean(queuedInteractionId(wake.payload));
|
||||
if (isReceipt && receipt) continue;
|
||||
// A receipt a run already received is not pending; a later one may be.
|
||||
const readReceipt = () => readQueuedInteractionResponse(executor as Db, wake.companyId, issue.id, wake.payload);
|
||||
if (wake.status === "deferred_issue_execution") {
|
||||
if (isReceipt) { receipt = { wake, state: "deferred", queueRun: null }; continue; }
|
||||
if (isReceipt) {
|
||||
const response = await readReceipt();
|
||||
if (response) receipt = { wake, state: "deferred", queueRun: null, response };
|
||||
continue;
|
||||
}
|
||||
return { wake, state: "deferred", queueRun: null };
|
||||
}
|
||||
if (!wake.runId) continue;
|
||||
@@ -7473,7 +7481,10 @@ export function issueRoutes(
|
||||
)
|
||||
.limit(1)
|
||||
.then((runRows) => runRows[0] ?? null);
|
||||
if (queueRun && isReceipt) receipt = { wake, state: "queued", queueRun };
|
||||
if (queueRun && isReceipt) {
|
||||
const response = await readReceipt();
|
||||
if (response) receipt = { wake, state: "queued", queueRun, response };
|
||||
}
|
||||
else if (queueRun) return { wake, state: "queued", queueRun };
|
||||
}
|
||||
return receipt;
|
||||
@@ -7483,11 +7494,9 @@ export function issueRoutes(
|
||||
executor: IssueQueueDb,
|
||||
issueId: string,
|
||||
wake: Pick<IssueQueueWake, "companyId" | "payload"> | null,
|
||||
response: QueuedInteractionResponse | null,
|
||||
) {
|
||||
if (wake && queuedInteractionId(wake.payload)) {
|
||||
const response = await readQueuedInteractionResponse(executor as Db, wake.companyId, issueId, wake.payload);
|
||||
return response ? [response.comment] : [];
|
||||
}
|
||||
if (wake && queuedInteractionId(wake.payload)) return response ? [response.comment] : [];
|
||||
const ids = queuedCommentIdsFromWakePayload(wake?.payload);
|
||||
if (ids.length === 0) return [];
|
||||
const rows = await executor
|
||||
@@ -7516,10 +7525,17 @@ export function issueRoutes(
|
||||
? await findQueuedCommentWake(input.executor, input.issue)
|
||||
: input.queueState;
|
||||
const wake = queueState?.wake ?? null;
|
||||
// Read an interaction receipt once per snapshot.
|
||||
const response = wake && queuedInteractionId(wake.payload)
|
||||
? queueState?.response !== undefined
|
||||
? queueState.response
|
||||
: await readQueuedInteractionResponse(input.executor as Db, input.issue.companyId, input.issue.id, wake.payload)
|
||||
: null;
|
||||
const comments = await queueCommentsForWake(
|
||||
input.executor,
|
||||
input.issue.id,
|
||||
wake,
|
||||
response,
|
||||
);
|
||||
const assignedAgent = input.issue.assigneeAgentId
|
||||
? await input.executor
|
||||
@@ -7563,7 +7579,6 @@ export function issueRoutes(
|
||||
actorId: input.actor.actorId,
|
||||
});
|
||||
if (wake && queuedInteractionId(wake.payload)) {
|
||||
const response = await readQueuedInteractionResponse(input.executor as Db, input.issue.companyId, input.issue.id, wake.payload);
|
||||
if (response?.source.requiresFreshSession) queue.steeringDisposition = "unsupported";
|
||||
queue.entries = response ? [{ comment: response.comment, source: response.source, position: 0, canEdit: false, canDiscard: false }] : [];
|
||||
}
|
||||
|
||||
@@ -34,15 +34,42 @@ export function queuedInteractionSummary(interaction: {
|
||||
return `${label}: ${subject}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* A run that started with this exact response in its context already received
|
||||
* it. The run must carry the resolved status: the card-creation wake carries
|
||||
* the same interaction id while the card is still pending, so it never counts.
|
||||
* Any other wake that still carries the same receipt (for example a recovery
|
||||
* continuation that raced the response's own wake) is redundant: it is not
|
||||
* pending input, so it must not hold the composer queue or the issue.
|
||||
* A cancelled or failed start does not count; that response is still owed.
|
||||
*/
|
||||
// Table-qualified on purpose: inside a select list Drizzle renders columns
|
||||
// unqualified, which would bind them to the subquery's own run row.
|
||||
function deliveredInteractionRunExists(companyId: string, issueId: string) {
|
||||
return sql<boolean>`EXISTS (SELECT 1 FROM heartbeat_runs delivered_run
|
||||
WHERE delivered_run.company_id = ${companyId}
|
||||
AND delivered_run.context_snapshot->>'issueId' = ${issueId}
|
||||
AND delivered_run.context_snapshot->>'interactionId' = issue_thread_interactions.id::text
|
||||
AND delivered_run.context_snapshot->>'interactionStatus' = issue_thread_interactions.status
|
||||
AND delivered_run.started_at IS NOT NULL
|
||||
AND delivered_run.status IN ('running', 'succeeded'))`;
|
||||
}
|
||||
|
||||
export type QueuedInteractionResponse = NonNullable<Awaited<ReturnType<typeof readQueuedInteractionResponse>>>;
|
||||
|
||||
export async function readQueuedInteractionResponse(db: Db, companyId: string, issueId: string, payload: unknown) {
|
||||
const id = queuedInteractionId(payload);
|
||||
if (!id) return null;
|
||||
const [interaction] = await db.select().from(issueThreadInteractions).where(and(
|
||||
const [row] = await db.select({
|
||||
interaction: issueThreadInteractions,
|
||||
delivered: deliveredInteractionRunExists(companyId, issueId),
|
||||
}).from(issueThreadInteractions).where(and(
|
||||
eq(issueThreadInteractions.id, id), eq(issueThreadInteractions.companyId, companyId),
|
||||
eq(issueThreadInteractions.issueId, issueId),
|
||||
inArray(issueThreadInteractions.status, ["accepted", "answered", "rejected"]),
|
||||
));
|
||||
if (!interaction || interaction.status !== record(payload).interactionStatus) return null;
|
||||
const interaction = row?.interaction;
|
||||
if (!interaction || interaction.status !== record(payload).interactionStatus || row.delivered) return null;
|
||||
const p = record(interaction.payload);
|
||||
const result = record(interaction.result);
|
||||
const heading = queuedInteractionSummary(interaction);
|
||||
@@ -80,6 +107,7 @@ export async function hasQueuedInteractionResponse(db: Db, companyId: string, is
|
||||
inArray(agentWakeupRequests.status, ["deferred_issue_execution", "queued"]),
|
||||
sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
|
||||
sql`${agentWakeupRequests.payload}->>'mutation' = 'interaction'`,
|
||||
sql`NOT ${deliveredInteractionRunExists(companyId, issueId)}`,
|
||||
)).limit(1);
|
||||
return Boolean(row);
|
||||
}
|
||||
Reference in new issue
Block a user