diff --git a/packages/shared/src/types/issue.ts b/packages/shared/src/types/issue.ts index 4598caecc50..725bd6f677b 100644 --- a/packages/shared/src/types/issue.ts +++ b/packages/shared/src/types/issue.ts @@ -1177,6 +1177,13 @@ export interface IssueQueuedCommentQueue { entries: IssueQueuedCommentEntry[]; /** Current admission condition for a saved user continuation. */ executionWait?: { reason: string; message: string } | null; + /** + * Messages the target run's first provider turn takes: ones it already + * sealed into that turn, and ones it will take while it is still starting + * (for example while its sandbox boots). They are sent, not waiting for + * Steer or Interrupt. Omitted when there are none. + */ + firstTurnCommentIds?: string[]; } interface IssueCommentMetadataRowBase { diff --git a/server/src/__tests__/issue-queued-comments-routes.test.ts b/server/src/__tests__/issue-queued-comments-routes.test.ts index 9487cc0e2f7..973aa0df559 100644 --- a/server/src/__tests__/issue-queued-comments-routes.test.ts +++ b/server/src/__tests__/issue-queued-comments-routes.test.ts @@ -1018,6 +1018,37 @@ describeEmbeddedPostgres("issue queued-comment routes", () => { }, ); + it.each(["preparing", "native"] as const)( + "reports the messages a %s run's first turn takes, before and after it seals them", + async (runtime) => { + const seeded = await seedFirstTurnQueue(runtime); + const client = app(seeded.companyId); + // The sandbox is still booting: the messages are sent, not waiting for Steer. + const before = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200); + expect(before.body.firstTurnCommentIds).toEqual(seeded.commentIds); + + if (runtime === "preparing") { + await db.update(heartbeatRuns).set({ runtimeMode: "native", runtimeModeResolvedAt: new Date(), + nativeIssueId: seeded.issueId }).where(eq(heartbeatRuns.id, seeded.runId)); + } + await claimFirstTurnQueuedInput(db, { companyId: seeded.companyId, runId: seeded.runId, + issueId: seeded.issueId, agentId: seeded.agentId }); + // After the seal the queue hides them; the view still needs to know where they went. + const after = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200); + expect(after.body.entries).toEqual([]); + expect(after.body.firstTurnCommentIds).toEqual(seeded.commentIds); + }, + ); + + it("reports no first-turn messages once the run's turn already started", async () => { + const seeded = await seedFirstTurnQueue("native"); + await appendHeartbeatRunEvent(db, { companyId: seeded.companyId, runId: seeded.runId, agentId: seeded.agentId, + eventType: "turn.accepted", stream: "system", level: "info", message: "turn accepted" }); + const queue = await request(app(seeded.companyId)).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200); + expect(queue.body.entries).toHaveLength(2); + expect(queue.body.firstTurnCommentIds ?? []).toEqual([]); + }); + it("still interrupts a preparing run when its queue holds another user's message", async () => { const seeded = await seedFirstTurnQueue("native"); // Ordinary comments from two users merge into one wake that keeps the diff --git a/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts b/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts index de0e5674605..b601825a87e 100644 --- a/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts +++ b/server/src/modules/wake-queue/adapters/queued-comment-postgres.ts @@ -22,7 +22,7 @@ import { } from "../../../services/issue-queued-comment-queue.js"; import { logActivity as persistActivityLogRow, type ActivityPublication } from "../../../services/activity-log.js"; import { decideQueuedCommentWakeLookup } from "../domain/policy.js"; -import { firstTurnClaimedCommentIds } from "../../../services/native-runtime/first-turn-queued-input.js"; +import { firstTurnClaimedCommentIds, firstTurnQueuedCommentIds, firstTurnWakeColumns } from "../../../services/native-runtime/first-turn-queued-input.js"; import { parseObject, readNonEmptyString } from "../domain/values.js"; import { QueuedCommentMutationError } from "../application/queued-comment-use-cases.js"; import type { @@ -208,7 +208,7 @@ function buildTransaction(tx: Db, companyId: string, issueId: string, deps: Queu const steeringDisposition: IssueQueuedCommentQueue["steeringDisposition"] = steering.kind === "probe" ? "temporarily_unavailable" : steering.kind; - return buildQueuedCommentQueueSnapshot({ + const queue = buildQueuedCommentQueueSnapshot({ issueId: issue.id, queueId: wake?.id ?? null, state, @@ -219,6 +219,18 @@ function buildTransaction(tx: Db, companyId: string, issueId: string, deps: Queu actorType: actor.actorType, actorId: actor.actorId, }); + // Same view as the read path: what the starting run's first turn takes is sent. + const firstTurnWake = wake && state === "deferred" && activeRun?.status === "running" && wakeCommentIds.length > 0 + ? await tx.select(firstTurnWakeColumns).from(agentWakeupRequests) + .where(and(eq(agentWakeupRequests.id, wake.id), eq(agentWakeupRequests.companyId, companyId))) + .limit(1).then((wakeRows) => wakeRows[0] ?? null) + : null; + const firstTurnCommentIds = wake && firstTurnWake && activeRun + ? await firstTurnQueuedCommentIds(tx, { companyId, issueId: issue.id, runId: activeRun.id, + wake: { ...firstTurnWake, status: wake.status, payload: wake.payload } }) + : []; + if (firstTurnCommentIds.length > 0) queue.firstTurnCommentIds = firstTurnCommentIds; + return queue; }, async syncCommentReferences(commentId) { diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index ba8255ea1f0..4e77a6040b0 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -50,7 +50,7 @@ import { confirmNativeControllerStopped, NATIVE_CONTROLLER_HOST_UNVERIFIED } fro import { confirmOwnerGone, OWNER_UNPROVABLE_CAUSE } from "../services/recovery/owner-gone-attestation.js"; import { OWNABLE_KINDS } from "../services/recovery/ownership-registry.js"; import { retryNativeWorkspaceExport } from "../services/native-runtime/native-workspace-export-retry.js"; -import { deferredWakeJoinsFirstTurn, firstTurnClaimedCommentIds, firstTurnTakesAllQueuedComments, runAcceptsFirstTurnQueuedInput } from "../services/native-runtime/first-turn-queued-input.js"; +import { deferredWakeJoinsFirstTurn, firstTurnClaimedCommentIds, firstTurnQueuedCommentIds, firstTurnTakesAllQueuedComments, firstTurnWakeColumns, firstTurnWakeContextFields, firstTurnWakePayloadFields, runAcceptsFirstTurnQueuedInput } from "../services/native-runtime/first-turn-queued-input.js"; 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"; @@ -7541,7 +7541,8 @@ export function issueRoutes( type IssueQueueWake = typeof agentWakeupRequests.$inferSelect; type IssueQueueRun = typeof heartbeatRuns.$inferSelect; type IssueQueueState = { - wake: Pick; + wake: Pick; state: "deferred" | "queued"; queueRun: Pick | null; /** The receipt already read for an interaction wake, reused by the queue builder. */ @@ -7557,16 +7558,18 @@ export function issueRoutes( .select({ id: agentWakeupRequests.id, companyId: agentWakeupRequests.companyId, status: agentWakeupRequests.status, runId: agentWakeupRequests.runId, + ...firstTurnWakeColumns, payload: sql>`jsonb_build_object( + ${firstTurnWakePayloadFields(agentWakeupRequests.payload)}, 'issueId', ${agentWakeupRequests.payload}->'issueId', 'mutation', ${agentWakeupRequests.payload}->'mutation', - 'interactionId', ${agentWakeupRequests.payload}->'interactionId', 'interactionStatus', ${agentWakeupRequests.payload}->'interactionStatus', 'externalChatContinuation', ${agentWakeupRequests.payload}->'externalChatContinuation', 'executionWait', jsonb_build_object( 'reason', ${agentWakeupRequests.payload}->'executionWait'->'reason', 'message', ${agentWakeupRequests.payload}->'executionWait'->'message'), '_paperclipWakeContext', jsonb_build_object( + ${firstTurnWakeContextFields(agentWakeupRequests.payload)}, 'wakeCommentIds', ${agentWakeupRequests.payload}->'_paperclipWakeContext'->'wakeCommentIds', 'forceFreshSession', ${agentWakeupRequests.payload}->'_paperclipWakeContext'->'forceFreshSession') )`, @@ -7725,6 +7728,13 @@ export function issueRoutes( if (wake && queuedInteractionId(wake.payload)) { if (response?.source.requiresFreshSession) queue.steeringDisposition = "unsupported"; queue.entries = response ? [{ comment: response.comment, source: response.source, position: 0, canEdit: false, canDiscard: false }] : []; + } else if (wake && queueState?.state === "deferred" && input.activeRun?.status === "running") { + // A starting run (a booting sandbox, say) cannot be steered yet, but its + // first turn takes these messages. The view shows them as sent. + const firstTurnCommentIds = await firstTurnQueuedCommentIds(input.executor, { + companyId: input.issue.companyId, issueId: input.issue.id, runId: input.activeRun.id, wake, + }); + if (firstTurnCommentIds.length > 0) queue.firstTurnCommentIds = firstTurnCommentIds; } return queue; } diff --git a/server/src/services/native-runtime/first-turn-queued-input.ts b/server/src/services/native-runtime/first-turn-queued-input.ts index 216b173c276..5f494b4c5b3 100644 --- a/server/src/services/native-runtime/first-turn-queued-input.ts +++ b/server/src/services/native-runtime/first-turn-queued-input.ts @@ -250,6 +250,81 @@ export async function runAcceptsFirstTurnQueuedInput( return !accepted; } +/** The wake facts `deferredWakeJoinsFirstTurn` reads, without the rest of a large payload. */ +export const firstTurnWakeColumns = { + agentId: agentWakeupRequests.agentId, + reason: agentWakeupRequests.reason, + requestedByActorType: agentWakeupRequests.requestedByActorType, + requestedByActorId: agentWakeupRequests.requestedByActorId, +}; + +/** The lean payload fields `deferredWakeJoinsFirstTurn` reads, for a `jsonb_build_object` projection. */ +export function firstTurnWakePayloadFields(payload: typeof agentWakeupRequests.payload) { + return sql`'interactionId', ${payload}->'interactionId', + 'queuedCommentInterrupt', ${payload}->'queuedCommentInterrupt'->'requestedAt', + 'externalChatProvider', ${payload}->'externalChatProvider'`; +} + +/** The lean `_paperclipWakeContext` fields `deferredWakeJoinsFirstTurn` reads. */ +export function firstTurnWakeContextFields(payload: typeof agentWakeupRequests.payload) { + return sql`'wakeReason', ${payload}->'_paperclipWakeContext'->'wakeReason', + 'interactionId', ${payload}->'_paperclipWakeContext'->'interactionId', + 'interactionStatus', ${payload}->'_paperclipWakeContext'->'interactionStatus', + 'resumeIntent', ${payload}->'_paperclipWakeContext'->'resumeIntent', + 'chatConversationId', ${payload}->'_paperclipWakeContext'->'chatConversationId'`; +} + +/** + * The messages in one deferred queue that the running run's first turn takes: + * the ones it already sealed, or, while it is still starting, the ones its + * claim will take under the same rule as `claimFirstTurnQueuedInput`. A queue + * view shows these as sent instead of waiting for Steer, which cannot reach a + * run whose provider session (for example a booting sandbox) is not up yet. + * Presentation only: delivery still belongs to the claim and the deferred wake. + */ +export async function firstTurnQueuedCommentIds( + tx: Pick, + input: { + companyId: string; + issueId: string; + runId: string; + wake: { agentId: string; status: string; reason: string | null; requestedByActorType: string | null; + requestedByActorId: string | null; payload: unknown }; + }, +): Promise { + const { wake } = input; + if (wake.status !== DEFERRED_WAKE_STATUS) return []; + const wakeIds = queuedCommentIdsFromWakePayload(wake.payload); + if (wakeIds.length === 0) return []; + // Read only the receipt and dispatch choice; a run snapshot can be large. + const [row] = await tx.select({ + id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + status: heartbeatRuns.status, runtimeMode: heartbeatRuns.runtimeMode, + runtimeModeResolvedAt: heartbeatRuns.runtimeModeResolvedAt, responsibleUserId: heartbeatRuns.responsibleUserId, + dispatchAdapterType: sql`${heartbeatRuns.runnerProfileJson}->'adapterDispatch'->>'adapterType'`, + receipt: sql`${heartbeatRuns.contextSnapshot}->${FIRST_TURN_QUEUED_INPUT_KEY}`, + executionRunId: issues.executionRunId, conversationAgentId: issues.conversationAgentId, + }).from(heartbeatRuns) + .leftJoin(issues, and(eq(issues.id, input.issueId), eq(issues.companyId, input.companyId))) + .where(and(eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId))).limit(1); + if (!row || row.status !== "running" || row.agentId !== wake.agentId) return []; + const run = { ...row, runnerProfileJson: { adapterDispatch: { adapterType: row.dispatchAdapterType } }, + contextSnapshot: { [FIRST_TURN_QUEUED_INPUT_KEY]: row.receipt } }; + const receipt = readFirstTurnQueuedInputReceipt(run); + if (receipt) { + const sealed = new Set(receipt.commentIds); + return wakeIds.filter(id => sealed.has(id)); + } + if (row.conversationAgentId || row.executionRunId !== run.id || !run.responsibleUserId) return []; + if (!deferredWakeJoinsFirstTurn(wake, run)) return []; + if (!(await runAcceptsFirstTurnQueuedInput(tx, run))) return []; + const eligible = await tx.select({ id: issueComments.id }).from(issueComments).where(and( + inArray(issueComments.id, wakeIds), + firstTurnEligibleComment({ companyId: input.companyId, issueId: input.issueId, responsibleUserId: run.responsibleUserId }))); + const ids = new Set(eligible.map(row => row.id)); + return wakeIds.filter(id => ids.has(id)); +} + /** * True when the first turn would take every pending message in this queue, * under the same rule as the claim. Another user's message, an agent's, or one diff --git a/tests/runner-e2e/source-sdk-public-inputs.json b/tests/runner-e2e/source-sdk-public-inputs.json index 0697ff2f739..105248e26d5 100644 --- a/tests/runner-e2e/source-sdk-public-inputs.json +++ b/tests/runner-e2e/source-sdk-public-inputs.json @@ -402,7 +402,7 @@ "packages/shared/src/types/index.ts": "01845e2a041eb892c3a193685471d635da2c1be9", "packages/shared/src/types/instance.ts": "2c40a18c05010991d93d719937faa4a43e47104f", "packages/shared/src/types/issue-tree-control.ts": "75489fffd491b8350b05caf57d7796419797d9df", - "packages/shared/src/types/issue.ts": "4598caecc50e60450a1204de832a4fcd7991096d", + "packages/shared/src/types/issue.ts": "725bd6f677b9d461460697d1ce10dee739a66ebe", "packages/shared/src/types/live.ts": "b90e55e4de8600c6fc17d7fa8d89079d4ae655ec", "packages/shared/src/types/native-finalization.ts": "c026a73003ae232fea4f7624d561493b383df358", "packages/shared/src/types/plugin.ts": "36e0f50d97f96532cb1fbdf3223368d92a7e9b1e", diff --git a/ui/src/lib/issue-queued-comment-queue.ts b/ui/src/lib/issue-queued-comment-queue.ts index e3b43757bc6..bfbf93b73b0 100644 --- a/ui/src/lib/issue-queued-comment-queue.ts +++ b/ui/src/lib/issue-queued-comment-queue.ts @@ -113,6 +113,9 @@ export function normalizeIssueQueuedCommentQueue( executionWait: typeof wait?.reason === "string" && typeof wait?.message === "string" ? { reason: wait.reason, message: wait.message } : null, + ...(Array.isArray(source?.firstTurnCommentIds) ? { + firstTurnCommentIds: source.firstTurnCommentIds.filter((id): id is string => typeof id === "string" && id.length > 0), + } : {}), }; } @@ -182,5 +185,6 @@ export function mergePendingIssueQueuedComments(params: { : "unsupported"), entries, executionWait: params.authoritativeQueue?.executionWait ?? null, + ...(params.authoritativeQueue?.firstTurnCommentIds ? { firstTurnCommentIds: params.authoritativeQueue.firstTurnCommentIds } : {}), }; } diff --git a/ui/src/pages/IssueDetail.test.tsx b/ui/src/pages/IssueDetail.test.tsx index 1bc7cab7e9f..e4d90a82224 100644 --- a/ui/src/pages/IssueDetail.test.tsx +++ b/ui/src/pages/IssueDetail.test.tsx @@ -4604,6 +4604,134 @@ describe("IssueDetail", () => { }); }); + it.each(["still starting", "sealed into its first turn"] as const)( + "shows a message the booting run's first turn takes as sent, not queued (%s)", + async (stage) => { + const followUp = createIssueComment({ id: "boot-follow-up", body: "Steer while the sandbox boots", + createdAt: new Date("2026-04-21T00:00:05.000Z"), updatedAt: new Date("2026-04-21T00:00:05.000Z") }); + mockIssuesApi.get.mockResolvedValue(createIssue({ status: "in_progress", assigneeAgentId: "agent-1", + executionRunId: "run-boot" })); + mockIssuesApi.listComments.mockResolvedValue([followUp]); + mockIssuesApi.recentComments.mockResolvedValue({ comments: [followUp], latestCommentId: followUp.id, hasMore: false }); + const bootingRun = { + id: "run-boot", status: "running", invocationSource: "assignment", triggerDetail: null, + contextCommentId: null, contextWakeCommentId: null, + startedAt: "2026-04-21T00:00:00.000Z", finishedAt: null, createdAt: "2026-04-21T00:00:00.000Z", + agentId: "agent-1", agentName: "Codie", adapterType: "paperclip_runner", issueId: "issue-1", + }; + mockHeartbeatsApi.activeRunForIssue.mockResolvedValue(bootingRun); + mockHeartbeatsApi.liveRunsForIssue.mockResolvedValue([bootingRun]); + // Staging PAP-4454: the server hides a sealed message from the queue, + // and steering is unavailable until the sandbox's session is up. + mockIssuesApi.getQueuedComments.mockResolvedValue(createQueuedCommentQueue({ + queueId: "boot-wake", state: "deferred", targetRunId: "run-boot", + steeringDisposition: "temporarily_unavailable", + entries: stage === "still starting" ? [{ comment: followUp, position: 0, canEdit: true, canDiscard: true }] : [], + firstTurnCommentIds: [followUp.id], + })); + await act(async () => renderIssueView()); + await waitForAssertion(() => { + expect(mockIssuesApi.getQueuedComments).toHaveBeenCalled(); + const props = mockIssueChatThreadRender.mock.calls.at(-1)?.[0]; + const bubble = props?.comments.find((comment: IssueComment) => comment.id === followUp.id); + expect(bubble).toMatchObject({ body: followUp.body, consumedByRunId: "run-boot" }); + expect(bubble?.queueState).toBeUndefined(); + expect(props?.queuedCommentQueue?.entries ?? []).toEqual([]); + }); + }, + ); + + it("keeps first-turn messages the view hides in a reorder of the rest", async () => { + const at = (second: number) => new Date(`2026-04-21T00:00:0${second}.000Z`); + const joining = createIssueComment({ id: "joins-first-turn", body: "Plain text", createdAt: at(5), updatedAt: at(5) }); + const withFileA = createIssueComment({ id: "with-file-a", body: "First file", createdAt: at(6), updatedAt: at(6) }); + const withFileB = createIssueComment({ id: "with-file-b", body: "Second file", createdAt: at(7), updatedAt: at(7) }); + mockIssuesApi.get.mockResolvedValue(createIssue({ status: "in_progress", assigneeAgentId: "agent-1", + executionRunId: "run-boot" })); + mockIssuesApi.listComments.mockResolvedValue([joining, withFileA, withFileB]); + mockIssuesApi.recentComments.mockResolvedValue({ comments: [joining, withFileA, withFileB], + latestCommentId: withFileB.id, hasMore: false }); + const bootingRun = { + id: "run-boot", status: "running", invocationSource: "assignment", triggerDetail: null, + contextCommentId: null, contextWakeCommentId: null, + startedAt: "2026-04-21T00:00:00.000Z", finishedAt: null, createdAt: "2026-04-21T00:00:00.000Z", + agentId: "agent-1", agentName: "Codie", adapterType: "paperclip_runner", issueId: "issue-1", + }; + mockHeartbeatsApi.activeRunForIssue.mockResolvedValue(bootingRun); + mockHeartbeatsApi.liveRunsForIssue.mockResolvedValue([bootingRun]); + // Until the seal, the server queue still lists the message the first turn will take. + const queue = createQueuedCommentQueue({ + queueId: "boot-wake", state: "deferred", targetRunId: "run-boot", steeringDisposition: "temporarily_unavailable", + entries: [joining, withFileA, withFileB].map((comment, position) => ({ comment, position, canEdit: true, canDiscard: true })), + firstTurnCommentIds: [joining.id], + }); + mockIssuesApi.getQueuedComments.mockResolvedValue(queue); + mockIssuesApi.reorderQueuedComments.mockResolvedValue(queue); + await act(async () => renderIssueView()); + await waitForAssertion(() => { + const props = mockIssueChatThreadRender.mock.calls.at(-1)?.[0]; + expect(props?.queuedCommentQueue?.entries.map((entry: { comment: IssueComment }) => entry.comment.id)) + .toEqual([withFileA.id, withFileB.id]); + }); + const props = mockIssueChatThreadRender.mock.calls.at(-1)?.[0]; + await act(async () => props.onReorderQueuedComments([withFileB.id, withFileA.id], queue.revision)); + expect(mockIssuesApi.reorderQueuedComments).toHaveBeenCalledWith("issue-1", { + orderedCommentIds: [joining.id, withFileB.id, withFileA.id], queueId: "boot-wake", revision: queue.revision, + }); + }); + + it("keeps queue controls for a new message after the run that took an earlier one hands off", async () => { + const runnerRun = (id: string, startedAt: string) => ({ + id, status: "running" as const, invocationSource: "assignment" as const, triggerDetail: null, + contextCommentId: null, contextWakeCommentId: null, startedAt, finishedAt: null, createdAt: startedAt, + agentId: "agent-1", agentName: "Codie", adapterType: "paperclip_runner", issueId: "issue-1", + }); + const bootRun = runnerRun("run-boot", "2026-04-21T00:00:00.000Z"); + const bootMessage = createIssueComment({ id: "boot-message", body: "Sent while the sandbox boots", + createdAt: new Date("2026-04-21T00:00:05.000Z"), updatedAt: new Date("2026-04-21T00:00:05.000Z") }); + mockIssuesApi.get.mockResolvedValue(createIssue({ status: "in_progress", assigneeAgentId: "agent-1", + executionRunId: "run-boot" })); + mockIssuesApi.addComment.mockResolvedValue(bootMessage); + mockHeartbeatsApi.activeRunForIssue.mockResolvedValue(bootRun); + mockHeartbeatsApi.liveRunsForIssue.mockResolvedValue([bootRun]); + mockIssuesApi.getQueuedComments.mockResolvedValue(createQueuedCommentQueue({ + queueId: null, state: null, targetRunId: null, steeringDisposition: "temporarily_unavailable", entries: [], + })); + await act(async () => renderIssueView()); + await flushReact(); + const bootProps = mockIssueChatThreadRender.mock.calls.at(-1)?.[0] as { onAdd: (body: string) => Promise }; + await act(async () => { await bootProps.onAdd(bootMessage.body); }); + await flushReact(); + + // The boot run's first turn took the message and finished; a successor runs. + // The server queue never lists the boot message, only the new follow-up. + const nextRun = runnerRun("run-next", "2026-04-21T00:10:00.000Z"); + const newMessage = createIssueComment({ id: "new-message", body: "Merge the PR when checks pass", + createdAt: new Date("2026-04-21T00:11:00.000Z"), updatedAt: new Date("2026-04-21T00:11:00.000Z") }); + mockIssuesApi.listComments.mockResolvedValue([bootMessage, newMessage]); + mockIssuesApi.recentComments.mockResolvedValue({ comments: [bootMessage, newMessage], + latestCommentId: newMessage.id, hasMore: false }); + const nextQueue = createQueuedCommentQueue({ queueId: "next-wake", state: "deferred", targetRunId: "run-next", + steeringDisposition: "available", + entries: [{ comment: newMessage, position: 0, canEdit: true, canDiscard: true }] }); + mockIssuesApi.getQueuedComments.mockResolvedValue(nextQueue); + mockHeartbeatsApi.activeRunForIssue.mockResolvedValue(nextRun); + mockHeartbeatsApi.liveRunsForIssue.mockResolvedValue([nextRun]); + mockIssuesApi.get.mockResolvedValue(createIssue({ status: "in_progress", assigneeAgentId: "agent-1", + executionRunId: "run-next" })); + await act(async () => { + await queryClient.invalidateQueries(); + await queryClient.invalidateQueries({ queryKey: queryKeys.issues.comments("issue-1") }); + }); + await waitForAssertion(() => { + const props = mockIssueChatThreadRender.mock.calls.at(-1)?.[0]; + // A stale local entry must not withhold Steer and Discard from the real queue. + expect(props?.queuedCommentQueue?.entries.map((entry: { comment: IssueComment }) => entry.comment.id)) + .toEqual([newMessage.id]); + expect(props?.queuedCommentQueue?.queueId).toBe("next-wake"); + }); + }); + it("does not mark the wake comment for the current live run as queued when active-run cache is stale", async () => { mockIssuesApi.get.mockResolvedValue( createIssue({ diff --git a/ui/src/pages/IssueDetail.tsx b/ui/src/pages/IssueDetail.tsx index 28beb0e70ab..713cc8adb5c 100644 --- a/ui/src/pages/IssueDetail.tsx +++ b/ui/src/pages/IssueDetail.tsx @@ -1851,6 +1851,26 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ queueIds: inputPlacementByCommentId.get(commentId)?.queueIds ?? new Set(), }); } + // A run that is still starting (a booting sandbox, say) cannot be steered, + // but its first turn takes these messages. They are sent: show them at + // their submission time in the conversation, not in the queue. + const firstTurnRunId = authoritativeQueuedCommentQueue?.targetRunId ?? null; + if (firstTurnRunId) { + const createdAtById = new Map(comments.map((comment) => [comment.id, comment.createdAt])); + authoritativeQueuedCommentQueue?.firstTurnCommentIds?.forEach((commentId, sequence) => { + const createdAt = createdAtById.get(commentId); + if (inputPlacementByCommentId.has(commentId) || !createdAt) return; + const anchorAt = new Date(createdAt).toISOString(); + inputPlacementByCommentId.set(commentId, { + runId: firstTurnRunId, + anchorAt, + sequence, + anchorMs: new Date(anchorAt).getTime(), + kind: "run_start", + queueIds: new Set(authoritativeQueuedCommentQueue.queueId ? [authoritativeQueuedCommentQueue.queueId] : []), + }); + }); + } for (const [commentId, placement] of localSteeringPlacements) { if (inputPlacementByCommentId.has(commentId)) continue; inputPlacementByCommentId.set(commentId, { @@ -2136,6 +2156,7 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ }); return [...projectedComments, ...responseComments]; }, [ + authoritativeQueuedCommentQueue, comments, classicTaskInterfaceEnabled, interactions, @@ -2194,6 +2215,11 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ : comment.queueState === "queued" ? (comment.queueTargetRunId ?? null) : null); + // A saved message the server queue never listed (a run's first turn + // took it) is not pending once that run settles. Keeping it would + // withhold the queue id, and with it Steer and Discard, from every + // later message. A request still in flight keeps its original target. + if (targetRunId && !liveRunIds.has(targetRunId) && !comment.id.startsWith("optimistic-")) return []; return targetRunId ? [{ comment, targetRunId }] : []; }); const fallbackProtocol = @@ -2213,6 +2239,7 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ consumedQueuedCommentIds, discardedQueuedCommentIds, issueId, + liveRunIds, liveRuntimeRun, assigneeUsesPaperclipRunner, locallyQueuedCommentRunIds, @@ -2319,9 +2346,17 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ "The queued messages are awaiting server acknowledgement.", ); try { + // The server orders its whole queue. Messages this view hides (inputs + // a starting run already admitted, or ones its first turn takes) keep + // their place; the person reorders only the rest. + const firstTurnIds = new Set(authoritativeQueuedCommentQueue?.firstTurnCommentIds ?? []); + const hidden = (id: string) => startingCommentIds.has(id) || firstTurnIds.has(id); + const serverEntries = authoritativeQueuedCommentQueue?.queueId === queueId + ? authoritativeQueuedCommentQueue.entries + : effectiveQueuedCommentQueue.entries; let nextPosition = 0; - const fullOrder = effectiveQueuedCommentQueue.entries.map(entry => - startingCommentIds.has(entry.comment.id) ? entry.comment.id : orderedCommentIds[nextPosition++]!); + const fullOrder = serverEntries.map(entry => + hidden(entry.comment.id) ? entry.comment.id : orderedCommentIds[nextPosition++]!); storeQueuedCommentQueue( await issuesApi.reorderQueuedComments(issueId, { orderedCommentIds: fullOrder, @@ -2334,6 +2369,7 @@ const IssueDetailChatTab = memo(function IssueDetailChatTab({ } }, [ + authoritativeQueuedCommentQueue, effectiveQueuedCommentQueue, startingCommentIds, issueId,