diff --git a/server/src/__tests__/agent-conversation-delivery.test.ts b/server/src/__tests__/agent-conversation-delivery.test.ts index 819c52a1545..8c7d22f0cba 100644 --- a/server/src/__tests__/agent-conversation-delivery.test.ts +++ b/server/src/__tests__/agent-conversation-delivery.test.ts @@ -1,9 +1,10 @@ import { randomUUID } from "node:crypto"; import { eq } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; -import { agents, agentWakeupRequests, companies, createDb, issueComments, issues } from "@paperclipai/db"; +import { agents, agentWakeupRequests, companies, createDb, heartbeatRuns, issueComments, issues } from "@paperclipai/db"; import { conflict } from "../errors.js"; -import { deliverConversationComments, undeliveredConversationComments } from "../services/agent-conversations.js"; +import { deliverConversationComments, heldConversationComments, undeliveredConversationComments } from "../services/agent-conversations.js"; +import { heartbeatService } from "../services/heartbeat.js"; import { instanceSettingsService } from "../services/instance-settings.js"; import { recoveryService } from "../services/recovery/service.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; @@ -79,6 +80,86 @@ const support = await getEmbeddedPostgresTestSupport(); expect(await f.pending()).toHaveLength(0); }); + // recovery.browser-reconnect: the fourth chat message after a failed-model + // phase hit the agent's daily run limit; admission wrote a skipped receipt + // under the outbox key, so the message counted as delivered and never ran. + async function capReached(f: Awaited>, startedAt = new Date()) { + await db.update(agents).set({ runtimeConfig: { heartbeat: { maxDailyRuns: 1 } } }).where(eq(agents.id, f.agentId)); + const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, + status: "succeeded", startedAt, finishedAt: startedAt }).returning(); + return run; + } + const nextUtcDay = () => { const now = new Date(); return Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate() + 1); }; + + it("holds a message at the agent's daily limit, shows why, and sends it once the limit resets", async () => { + const f = await fixture("idle"); + const run = await capReached(f); + const wake = enqueue(), deadlines: number[] = []; + expect(await deliverConversationComments(db, f.chat, wake, { onDeferredDeadline: at => deadlines.push(at) })).toBe("deferred"); + expect(wake).not.toHaveBeenCalled(); + expect(deadlines).toEqual([nextUtcDay()]); + expect((await f.pending()).map(row => row.id)).toEqual([f.comment.id]); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, f.agentId))).toEqual([]); + const held = await heldConversationComments(db, f.chat); + expect(held?.hold).toMatchObject({ reason: "heartbeat.daily_run_limit", retryAt: nextUtcDay() }); + expect(held?.hold.message).toContain("Your message is saved"); + expect(held?.comments.map(row => row.id)).toEqual([f.comment.id]); + + // The limit counts the current UTC day; yesterday's run no longer holds it. + await db.update(heartbeatRuns).set({ startedAt: new Date(Date.now() - 86_400_000) }).where(eq(heartbeatRuns.id, run.id)); + expect(await heldConversationComments(db, f.chat)).toBeNull(); + expect(await deliverConversationComments(db, f.chat, wake)).toBe("delivered"); + expect(wake).toHaveBeenCalledTimes(1); + expect(await f.pending()).toHaveLength(0); + }); + + it("schedules stranded-work recovery for the limit reset instead of consuming the message", async () => { + const f = await fixture("idle"); + await capReached(f); + const wake = enqueue(), deadlines: number[] = []; + await recoveryService(db, { enqueueWakeup: wake }).reconcileStrandedAssignedIssues({ + companyId: f.companyId, issueId: f.chat.id, onDeferredDeadline: at => deadlines.push(at) }); + expect(wake).not.toHaveBeenCalled(); + expect(deadlines).toContain(nextUtcDay()); + expect(await f.pending()).toHaveLength(1); + }); + + it("keeps the message owed when the limit is reached while real queue admission runs", async () => { + const f = await fixture("idle"); + const heartbeat = heartbeatService(db); + const wake = vi.fn[2]>(async (agentId, options) => { + await capReached(f); + return heartbeat.wakeup(agentId, options); + }); + expect(await deliverConversationComments(db, f.chat, wake)).toBe("deferred"); + expect(wake).toHaveBeenCalledTimes(1); + expect((await f.pending()).map(row => row.id)).toEqual([f.comment.id]); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, f.agentId))).toEqual([]); + }); + + it("makes a message owed again when dispatch cancels its accepted run at the daily limit", async () => { + const f = await fixture("idle"); + await db.update(agents).set({ runtimeConfig: { heartbeat: { maxDailyRuns: 1 } } }).where(eq(agents.id, f.agentId)); + const [receipt] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId, + source: "on_demand", triggerDetail: "manual", reason: "issue_commented", status: "queued", + idempotencyKey: `conversation-comment:${f.comment.id}`, requestedByActorType: "user", requestedByActorId: "local-board", + payload: { issueId: f.chat.id, commentId: f.comment.id } }).returning(); + const [queued] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "queued", + wakeupRequestId: receipt.id, invocationSource: "on_demand", + contextSnapshot: { issueId: f.chat.id, wakeCommentId: f.comment.id, wakeCommentIds: [f.comment.id] } }).returning(); + await db.update(agentWakeupRequests).set({ runId: queued.id }).where(eq(agentWakeupRequests.id, receipt.id)); + expect(await f.pending()).toHaveLength(0); + // Another turn reaches the limit after the message was accepted. + await capReached(f); + await heartbeatService(db).resumeQueuedRunsForAgent(f.companyId, f.agentId); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, queued.id)))[0]) + .toMatchObject({ status: "cancelled", errorCode: "heartbeat.daily_run_limit" }); + expect((await f.pending()).map(row => row.id)).toEqual([f.comment.id]); + const wake = enqueue(); + expect(await deliverConversationComments(db, f.chat, wake)).toBe("deferred"); + expect(wake).not.toHaveBeenCalled(); + }); + it("keeps unrelated admission errors visible and comments pending", async () => { const f = await fixture("idle"); await expect(deliverConversationComments(db, f.chat, async () => { diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index cee0bb74056..4d78b47ebde 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -48,7 +48,7 @@ import { deferredWakeJoinsFirstTurn, firstTurnClaimedCommentIds, firstTurnTakesA import { reconstructNativeWorkspace, unreconciledNativeExportPredicate } from "../services/native-runtime/native-workspace-reconstruction.js"; import { reconstructWorkspaceSchema } from "@paperclipai/shared"; import { queuedInteractionId, readQueuedInteractionResponse, hasQueuedInteractionResponse, type QueuedInteractionResponse } from "../services/queued-interaction-response.js"; -import { deliverConversationComments, isConversation } from "../services/agent-conversations.js"; +import { deliverConversationComments, heldConversationComments, isConversation } from "../services/agent-conversations.js"; import { assertOnboardingFirstTaskDependenciesAllowed } from "../services/onboarding-first-task-admission.js"; import { resolveChatTaskAssignment } from "../services/chat-task-assignment.js"; import { issueRecoveryActionReadModel } from "../services/issue-recovery-actions.js"; @@ -17291,12 +17291,38 @@ export function issueRoutes( ); if (!issue) return; if (!(await assertIssueReadAllowed(req, res, issue))) return; - const queue = await buildQueuedCommentQueue({ + const actor = getActorInfo(req); + let queue = await buildQueuedCommentQueue({ executor: db, issue, activeRun: await resolveActiveIssueRun(issue, { queueStateOnly: true }), - actor: getActorInfo(req), + actor, }); + // A chat message held by the agent's daily limit has no wake receipt yet: + // show it as saved and waiting, with the reason, instead of as sent. + const held = await heldConversationComments(db, issue); + if (held && (queue.state !== null || queue.entries.length > 0)) { + // Earlier queued work keeps its controls; held messages join it read-only. + const queuedIds = new Set(queue.entries.map(entry => entry.comment.id)); + const extra = held.comments.filter(comment => !queuedIds.has(comment.id)); + const snapshot = buildQueuedCommentQueueSnapshot({ + issueId: issue.id, queueId: null, state: "deferred", activeRunId: null, + protocol: queue.protocol, steeringDisposition: "unsupported", comments: extra, + actorType: actor.actorType, actorId: actor.actorId, + }); + queue = { ...queue, executionWait: queue.executionWait ?? { reason: held.hold.reason, message: held.hold.message }, + entries: [...queue.entries, ...snapshot.entries.map((entry, index) => ({ + ...entry, position: queue.entries.length + index, canEdit: false, canDiscard: false }))] }; + } else if (held) { + const snapshot = buildQueuedCommentQueueSnapshot({ + issueId: issue.id, queueId: null, state: "deferred", activeRunId: null, + protocol: queue.protocol, steeringDisposition: "unsupported", comments: held.comments, + executionWait: { reason: held.hold.reason, message: held.hold.message }, + actorType: actor.actorType, actorId: actor.actorId, + }); + // The outbox owns delivery; there is no queue to edit, send or discard. + queue = { ...snapshot, entries: snapshot.entries.map(entry => ({ ...entry, canEdit: false, canDiscard: false })) }; + } res.json( await runRedactions.redactForIssue(issue.companyId, issue.id, queue), ); diff --git a/server/src/services/agent-conversations.ts b/server/src/services/agent-conversations.ts index 9bbf648df33..8ba5927de20 100644 --- a/server/src/services/agent-conversations.ts +++ b/server/src/services/agent-conversations.ts @@ -26,6 +26,7 @@ import { import { notifyChatRunMilestones } from "./chat-run-milestone-notifications.js"; import { sanitizeQuarantinedCommentForHigherTrust } from "./source-trust.js"; import { evaluateAgentInvokabilityFromDb } from "./agent-invokability.js"; +import { createHeartbeatTimerPolicy, parseHeartbeatPolicy } from "./heartbeat/timer-policy.js"; import { isAgentAwaitingSetup } from "../modules/agent-lifecycle/index.js"; import { HttpError } from "../errors.js"; @@ -610,6 +611,45 @@ export async function resumeConversationForReset(db: Db, comment: typeof issueCo for (const publication of publications) publishActivity(publication); } +/** Why the conversation outbox holds a person's message, and when the hold lifts. */ +export type ConversationDeliveryHold = { reason: string; message: string; retryAt: number }; + +/** + * The agent's daily run or cost limit holds a chat message, as queue + * admission would. The message stays owed in the outbox (never a skipped + * receipt) and is sent when the limit resets at the next UTC day. + */ +export async function conversationDeliveryHold( + db: Pick, + agent: typeof agents.$inferSelect, + now = new Date(), +): Promise { + const cap = await createHeartbeatTimerPolicy(db).getHeartbeatDailyCapBlock(agent, parseHeartbeatPolicy(agent), { now }); + if (!cap) return null; + const limit = cap.reason === "heartbeat.daily_cost_limit" ? "daily cost limit" : "daily run limit"; + return { + reason: cap.reason, + retryAt: Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), now.getUTCDate() + 1), + message: `${agent.name} reached its ${limit}. Your message is saved and sends automatically when the limit resets at 00:00 UTC.`, + }; +} + +/** Owed chat messages held by a daily limit, so the chat can show them as saved and waiting. */ +export async function heldConversationComments( + db: Db, + issue: { id: string; companyId: string; conversationAgentId?: string | null }, +) { + if (!issue.conversationAgentId) return null; + const [agent] = await db.select().from(agents).where(and( + eq(agents.id, issue.conversationAgentId), eq(agents.companyId, issue.companyId), + )); + if (!agent) return null; + const hold = await conversationDeliveryHold(db, agent); + if (!hold) return null; + const comments = await undeliveredConversationComments(db, issue.companyId, issue.id); + return comments.length ? { hold, comments } : null; +} + /** Serialize durable outbox delivery across API servers; the normal wake queue owns execution. */ export async function deliverConversationComments( db: Db, @@ -627,16 +667,24 @@ export async function deliverConversationComments( contextSnapshot: Record; }, ) => Promise, + options: { onDeferredDeadline?: (at: number) => void } = {}, ) { if (!issue.conversationAgentId) return; - const canDeliver = async (connection: Db) => { + const canDeliver = async (connection: Db): Promise<{ invokable: true } | { invokable: false; reason: string }> => { const [agent] = await connection.select().from(agents).where(and( eq(agents.id, issue.conversationAgentId!), eq(agents.companyId, issue.companyId), )); // Match queue admission: setup may retain accepted work, but a pause or // invalid reporting chain must leave the durable comment outbox untouched. - return evaluateAgentInvokabilityFromDb(connection, + const invokability = await evaluateAgentInvokabilityFromDb(connection, agent && isAgentAwaitingSetup(agent) ? { ...agent, status: "idle" } : agent); + if (!invokability.invokable || !agent) return invokability; + // A reached daily limit is a temporary hold, not a reason to consume the + // message: admission would record it as a terminal skip. + const hold = await conversationDeliveryHold(connection, agent); + if (!hold) return invokability; + options.onDeferredDeadline?.(hold.retryAt); + return { invokable: false, reason: hold.reason }; }; for (;;) { const delivered = await db.transaction(async (tx) => { diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index be86fd31205..4a230a73c3a 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -3,6 +3,7 @@ import { CONTEXT_COMPACTION_WAKE_REASON, ContextCompactionError, isContextCompac import { isProcessRunAdmissionClosed } from "./run-admission.js"; import { deliveredQueuedCommentIds } from "../queued-comment-delivery.js"; import { deferredWorkIntentWait } from "../work-intent-outbox.js"; +import { notifyIssueRecovery } from "../issue-recovery-notifications.js"; import { createHeartbeatTimerPolicy } from "./timer-policy.js"; import { notifyExecutionRunWork } from "../execution-control-notifications.js"; import { PersistedExecutionWaitRefusal, writeDeferredExecutionWait } from "../deferred-execution-wait.js"; @@ -357,6 +358,14 @@ export function shouldQueueFollowupForRunningIssueWake(input: { ); } +/** Agent Chat outbox receipts. The outbox, not a skipped receipt, owns a capped message. */ +const CONVERSATION_COMMENT_KEY_PREFIX = "conversation-comment:"; +function dailyLimitConversationRefusal(cap: { reason: string; observed: number; limit: number }) { + // No receipt is written, so the person's message stays owed in the outbox + // until the limit resets (agent-conversations.ts conversationDeliveryHold). + return conflict("Agent reached its daily limit", { reason: cap.reason, observed: cap.observed, limit: cap.limit }); +} + function readNonEmptyString(value: unknown): string | null { return typeof value === "string" && value.trim().length > 0 ? value : null; } @@ -473,6 +482,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende finishedAt: now, error: reason, }); + await releaseConversationCommentsForDailyCap(cancelled); await appendRunEvent(cancelled, { eventType: "lifecycle", @@ -493,6 +503,34 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende return cancelled; } + /** + * A chat message accepted before the limit was reached must not be consumed + * when dispatch then cancels its run at the limit. Releasing the outbox key + * on its receipts (the run's own and any coalesced into it) makes the + * message owed again; the outbox holds it until the limit resets. + */ + async function releaseConversationCommentsForDailyCap(run: typeof heartbeatRuns.$inferSelect) { + const owned = and( + eq(agentWakeupRequests.companyId, run.companyId), + eq(agentWakeupRequests.agentId, run.agentId), + or(eq(agentWakeupRequests.runId, run.id), + run.wakeupRequestId ? eq(agentWakeupRequests.id, run.wakeupRequestId) : undefined), + sql`${agentWakeupRequests.idempotencyKey} like ${CONVERSATION_COMMENT_KEY_PREFIX + "%"}`, + ); + await db.transaction(async tx => { + const receipts = await tx.select({ id: agentWakeupRequests.id }).from(agentWakeupRequests).where(owned).for("update"); + if (!receipts.length) return; + // Register the recovery signal before the write it announces. + await notifyIssueRecovery(tx, { companyId: run.companyId }); + await tx.update(agentWakeupRequests).set({ + idempotencyKey: null, + payload: sql`coalesce(${agentWakeupRequests.payload}, '{}'::jsonb) + || jsonb_build_object('releasedConversationCommentKey', ${agentWakeupRequests.idempotencyKey})`, + updatedAt: new Date(), + }).where(and(owned, inArray(agentWakeupRequests.id, receipts.map(row => row.id)))); + }); + } + async function markTimerHeartbeatChecked( agentId: string, source: WakeupOptions["source"], @@ -4431,6 +4469,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende message: "The agent has reached its daily limit. Your message is saved until work can resume." }; return deferBlockedExecution(executionBlocker); } + if (opts.idempotencyKey?.startsWith(CONVERSATION_COMMENT_KEY_PREFIX)) throw dailyLimitConversationRefusal(dailyCapBlock); const now = new Date(); await tx.insert(agentWakeupRequests).values({ ...durableReceiptFields, @@ -4840,6 +4879,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende tx, ); if (dailyCapBlock) { + if (opts.idempotencyKey?.startsWith(CONVERSATION_COMMENT_KEY_PREFIX)) throw dailyLimitConversationRefusal(dailyCapBlock); const now = new Date(); await tx.insert(agentWakeupRequests).values({ ...durableReceiptFields, diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 4d3bd439c78..f085bbb0d3c 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -4612,7 +4612,8 @@ export function recoveryService( } if (!(await instanceSettingsService(db).getExperimental()).enableAgentChat) { result.skipped += 1; continue; } { - if (await deliverConversationComments(db, issue, deps.enqueueWakeup) === "deferred") { + if (await deliverConversationComments(db, issue, deps.enqueueWakeup, + { onDeferredDeadline: opts?.onDeferredDeadline }) === "deferred") { result.skipped += 1; continue; } diff --git a/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx b/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx index fa9b4d6bd07..31e21fce020 100644 --- a/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx +++ b/ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx @@ -89,6 +89,17 @@ describe("TaskChatQueuedMessages", () => { expect(container.textContent).not.toContain(message); }); + it("says why a chat message held in the outbox waits, without offering queue actions", () => { + const message = "CEO reached its daily run limit. Your message is saved and sends automatically when the limit resets at 00:00 UTC."; + render({ queue: { ...queue, queueId: null, targetRunId: null, steeringDisposition: "unsupported", + executionWait: { reason: "heartbeat.daily_run_limit", message }, + entries: queue.entries.slice(0, 1).map(entry => ({ ...entry, canEdit: false, canDiscard: false })) } }); + expect([...container.querySelectorAll('p[role="status"]')].map(node => node.textContent)).toEqual([message]); + expect(container.textContent).toContain("First queued message"); + expect(container.querySelector('[data-testid="task-chat-queued-interrupt-comment-1"]')?.disabled ?? true).toBe(true); + expect(container.querySelector('[data-testid="task-chat-queued-discard-comment-1"]')?.disabled).toBe(true); + }); + it("renders each queued message once as a compact one-line row", () => { render(); const pane = container.querySelector( diff --git a/ui/src/components/task-chat/TaskChatQueuedMessages.tsx b/ui/src/components/task-chat/TaskChatQueuedMessages.tsx index 12067e4fae8..a1799df0f43 100644 --- a/ui/src/components/task-chat/TaskChatQueuedMessages.tsx +++ b/ui/src/components/task-chat/TaskChatQueuedMessages.tsx @@ -512,7 +512,14 @@ export function TaskChatQueuedMessages({ } } - if (entries.length === 0 && !visibleError && !deferredStatus) return null; + // A chat message held in the outbox (for example by the agent's daily + // limit) has no receipt or run whose task status could explain the wait, + // so the queue says why it waits. Receipt-backed waits keep the task status. + const statusMessage = deferredStatus?.message + ?? (queue.state === "deferred" && !queue.queueId && !queue.targetRunId && entries.length > 0 + ? queue.executionWait?.message ?? null : null); + + if (entries.length === 0 && !visibleError && !statusMessage) return null; return ( - {deferredStatus ?

{deferredStatus.message}

: null} + {statusMessage ?

{statusMessage}

: null} {visibleError ? (