mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-14 00:26:28 +01:00
fix(queue): show messages a booting run's first turn takes as sent (#20822)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Task chat lets a person send follow-up messages to a running agent, and steer them into the live turn > - A sandboxed run (for example on Daytona) can take minutes to boot before its provider session exists > - A message sent during that boot cannot be steered, so it waits in the steering pane > - The runner then puts the message into its first turn, and the server removes it from the queue, but the view has no signal for this, so the message stays in the steering pane as "queued" > - That stale local entry also hides the queue id, so Steer and Discard fail for every later message > - This pull request tells the view which messages the first turn takes, and drops local queue entries whose run has settled > - The benefit is that a message sent while the sandbox boots shows as sent at once, and later messages keep working Steer and Discard ## Linked Issues or Issue Description **What happened?** A person sent a follow-up while the agent's sandbox was booting. The message stayed in the steering pane with Steer unavailable. After the agent started, the message stayed there. The agent had received it in its first turn. Later, Discard on that message failed with "Couldn't discard. Message is still queued.", and a new message also stayed in the steering pane. **Steps to reproduce** 1. Assign a task to a Paperclip Runner agent that runs in a Daytona sandbox. 2. While the sandbox boots, send a follow-up message. 3. Wait for the agent's first response. The message is still in the steering pane. 4. After the run hands off to a successor run, send another message and try to discard the first one. **Expected behavior** A message sent while the run is starting shows as a sent message. Later messages keep working Steer and Discard. **Actual behavior** The first message stays in the steering pane. Discard fails, and the new message cannot be steered until the page reloads. ## What Changed - `firstTurnQueuedCommentIds` (server, `first-turn-queued-input.ts`) returns the messages in a deferred queue that the target run's first turn already sealed, or will take while it is still starting. It uses the same eligibility rule as `claimFirstTurnQueuedInput`. It reads only lean columns, so the queue poll does not transfer run snapshots or large wake payloads. - The queue read path and the queue mutation path both put these ids in a new optional `firstTurnCommentIds` field on `IssueQueuedCommentQueue`. - The task view puts each of those messages in the conversation as input of that run, at its submission time, and not in the steering pane. - The task view no longer treats a saved message as a pending local queue entry after its target run is no longer live. A request that is still in flight keeps its original target. ## Verification - Red-green server test: `issue-queued-comments-routes.test.ts` "reports the messages a preparing/native run's first turn takes, before and after it seals them" and "reports no first-turn messages once the run's turn already started". Before the fix, `firstTurnCommentIds` was `undefined`. - Red-green UI tests: `IssueDetail.test.tsx` "shows a message the booting run's first turn takes as sent, not queued" (still starting, and sealed) and "keeps queue controls for a new message after the run that took an earlier one hands off". Before the fix, the queue held `['new-message', 'boot-message']` and the queue id was withheld. - Reproduced on staging first: the run's context held a first-turn receipt for the stuck message, and the server queue had no entries, while the view still showed it as queued. - `npx vitest run` for `issue-queued-comments-routes`, `queued-comment-consumption`, `queued-comment-postgres`, `queued-comment-use-cases`: 172 passed. The existing queue-poll byte budget tests still pass. - `npx vitest run src/pages/IssueDetail.test.tsx src/components/task-chat src/lib/issue-queued-comment-queue src/lib/optimistic-issue-comments`: 1288 passed. - `tsc --noEmit` for shared, server, and ui: no new errors. - `pnpm verify` exits 2 because the change touches `packages/shared`. Required CI supplies the broad coverage. ## Risks - Low risk. The new field is optional and only for display. Delivery still belongs to the first-turn claim and the deferred wake. - If a starting run fails before its first turn, the field goes away and the message shows in the queue again for the next run. This is correct, because the deferred wake still holds it. - The queue read adds one lean run read and one comment read only while a deferred queue targets a running run. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, in Claude Code with tool use (shell, file edits, browser). ## 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
a4b872477e
commit
0b8ba1ac92
9 files changed
+311
-8
No files matched your search
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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<IssueQueueWake, "id" | "companyId" | "payload" | "status" | "runId">;
|
||||
wake: Pick<IssueQueueWake, "id" | "companyId" | "payload" | "status" | "runId" | "agentId" | "reason"
|
||||
| "requestedByActorType" | "requestedByActorId">;
|
||||
state: "deferred" | "queued";
|
||||
queueRun: Pick<IssueQueueRun, "runtimeMode"> | 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<Record<string, unknown>>`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;
|
||||
}
|
||||
|
||||
@@ -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<Db, "select">,
|
||||
input: {
|
||||
companyId: string;
|
||||
issueId: string;
|
||||
runId: string;
|
||||
wake: { agentId: string; status: string; reason: string | null; requestedByActorType: string | null;
|
||||
requestedByActorId: string | null; payload: unknown };
|
||||
},
|
||||
): Promise<string[]> {
|
||||
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<string | null>`${heartbeatRuns.runnerProfileJson}->'adapterDispatch'->>'adapterType'`,
|
||||
receipt: sql<unknown>`${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
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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 } : {}),
|
||||
};
|
||||
}
|
||||
@@ -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<void> };
|
||||
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({
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in new issue
Block a user