fix: preserve chat composer submissions and session state

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-09-11 17:20:12 -05:00
commit c0d2c2ed3e
37 files changed
+436 -191

No files matched your search

@@ -13,6 +13,7 @@ import { randomUUID } from "node:crypto";
import { and, eq } from "drizzle-orm";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import {
activityLog,
authUsers,
agents,
agentTaskSessions,
@@ -236,6 +237,21 @@ const support = await getEmbeddedPostgresTestSupport();
).status,
).toBe(422);
const chatId = resolved[0].body.id;
for (const body of ["Hello", "/new"]) {
expect((await request(appFor(colleague)).post(`/api/issues/${chatId}/comments`)
.send({ body, clientRequestId: randomUUID() })).status).toBe(403);
}
expect((await request(appFor(colleague))
.post(`/api/companies/${companyId}/issues/${chatId}/attachments`)
.attach("file", Buffer.from("foreign upload"), "note.txt")).status).toBe(403);
const [planReview] = await db.insert(issueThreadInteractions).values({
companyId, issueId: chatId, kind: "request_confirmation", status: "pending",
continuationPolicy: "wake_assignee_on_accept", payload: { version: 1, prompt: "Hand off this plan?" },
}).returning();
expect((await request(appFor(colleague))
.post(`/api/issues/${chatId}/interactions/${planReview.id}/accept`).send({})).status).toBe(403);
expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, planReview.id)))[0].status).toBe("pending");
const [pause] = await db.insert(issueTreeHolds).values({
companyId, rootIssueId: chatId, mode: "pause", status: "active",
createdByActorType: "user", createdByUserId: owner,
@@ -245,9 +261,17 @@ const support = await getEmbeddedPostgresTestSupport();
.send({ body: "Continue working", clientRequestId: randomUUID() });
expect(blockedSend.status).toBe(409);
expect(await db.select().from(issueComments).where(eq(issueComments.issueId, chatId))).toHaveLength(0);
const reset = await request(app).post(`/api/issues/${chatId}/comments`)
.send({ body: "/new", clientRequestId: randomUUID() });
const resetRequest = { body: "/new", clientRequestId: randomUUID() };
const reset = await request(app).post(`/api/issues/${chatId}/comments`).send(resetRequest);
expect(reset.status).toBe(201);
const retriedResets = await Promise.all(Array.from({ length: 3 }, () =>
request(app).post(`/api/issues/${chatId}/comments`).send(resetRequest)));
expect(retriedResets.every((response) => response.status === 201 && response.body.id === reset.body.id)).toBe(true);
const addedEvents = await db.select().from(activityLog).where(and(
eq(activityLog.entityId, chatId), eq(activityLog.action, "issue.comment_added"),
));
expect(addedEvents).toHaveLength(1);
expect((await db.select().from(issueTreeHolds).where(eq(issueTreeHolds.id, pause.id)))[0].status).toBe("released");
const local = await request(appFor()).post(path);
expect(local.body.conversationUserId).toBe("local-board");
@@ -577,7 +601,7 @@ const support = await getEmbeddedPostgresTestSupport();
runningProcesses.delete(active.id);
}
});
it("runs real process turns, processes /new without invocation, and leaves the chat idle", async () => {
it.each([false, true])("runs real process turns and resets without replaying pre-Stop queued input (queued=%s)", async (queuedBeforeStop) => {
const runtimeCompany = randomUUID();
const runtimeAgent = randomUUID();
await db
@@ -585,7 +609,7 @@ const support = await getEmbeddedPostgresTestSupport();
.values({
id: runtimeCompany,
name: "Runtime chat",
issuePrefix: "RCHAT",
issuePrefix: queuedBeforeStop ? "RCHATQ" : "RCHAT",
requireBoardApprovalForNewAgents: false,
});
const generations: unknown[] = [];
@@ -682,6 +706,19 @@ const support = await getEmbeddedPostgresTestSupport();
await db.insert(issueThreadInteractions).values({ companyId: runtimeCompany, issueId: chat.id,
kind: "ask_user_questions", status: "pending", title: "Old topic", payload: { version: 1, questions: [{ id: "old", prompt: "Old topic?", options: [{ id: "yes", label: "Yes" }], selectionMode: "single", required: true }], supersedeOnUserComment: false },
});
let stoppedQueuedWakeId: string | null = null;
if (queuedBeforeStop) {
const [pending] = await db.insert(issueComments).values({ companyId: runtimeCompany,
issueId: chat.id, authorUserId: "local-board", body: "Old topic queued before Stop",
}).returning();
const [stoppedQueuedWake] = await db.insert(agentWakeupRequests).values({ companyId: runtimeCompany, agentId: runtimeAgent,
source: "on_demand", reason: "issue_execution_deferred", status: "deferred_issue_execution",
requestedByActorType: "user", requestedByActorId: "local-board",
payload: { issueId: chat.id, commentId: pending.id,
_paperclipWakeContext: { issueId: chat.id, wakeReason: "issue_commented", wakeCommentId: pending.id, wakeCommentIds: [pending.id] } },
}).returning();
stoppedQueuedWakeId = stoppedQueuedWake.id;
}
await send("/new");
await waitIdle();
expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.issueId, chat.id)))[0].status).toBe("expired");
@@ -690,6 +727,11 @@ const support = await getEmbeddedPostgresTestSupport();
await send("A fresh idea");
await waitIdle();
expect(generations).toEqual([0, 1]);
if (stoppedQueuedWakeId) {
const [stoppedWake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, stoppedQueuedWakeId));
expect(stoppedWake).toMatchObject({ status: "cancelled", runId: null });
}
expect(
await heartbeat.wakeup(runtimeAgent, {
source: "automation",
@@ -700,7 +742,7 @@ const support = await getEmbeddedPostgresTestSupport();
.select()
.from(issueComments)
.where(eq(issueComments.issueId, chat.id));
expect(history).toHaveLength(5);
expect(history).toHaveLength(queuedBeforeStop ? 6 : 5);
} finally {
await new Promise<void>((resolve) => listener.close(() => resolve()));
await rm(cwd, { recursive: true, force: true });
@@ -6710,13 +6710,20 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
await vi.waitFor(async () => expect((await heartbeat.getRun(next!.id))?.status).not.toBe("running"));
});
it.each(["dedicated deferred donor", "non-coalescing recipient"] as const)(
it.each(["dedicated deferred donor", "non-coalescing recipient", "persistent agent conversation"] as const)(
"does not adopt unrelated queued comments for a %s after Stop",
async (direction) => {
const { companyId, agentId, issueId, runId } = await seedRunFixture({
runtimeMode: "legacy",
agentStatus: "running",
});
const persistentConversation = direction === "persistent agent conversation";
if (persistentConversation) {
await instanceSettingsService(db).updateExperimental({ enableAgentChat: true });
await db.update(issues).set({
conversationAgentId: agentId, conversationUserId: "responsible-user", conversationState: "active",
}).where(eq(issues.id, issueId));
}
const heartbeat = heartbeatService(db);
const [pending, go] = await db
.insert(issueComments)
@@ -6801,11 +6808,11 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
reason: "issue_commented",
requestedByActorType: "user",
requestedByActorId: "responsible-user",
...(dedicatedDonor ? {} : { allowRunCoalescing: false }),
...(direction === "non-coalescing recipient" ? { allowRunCoalescing: false } : {}),
payload: {
issueId,
commentId: go!.id,
...(dedicatedDonor
...(dedicatedDonor || persistentConversation
? {}
: { mutation: "interaction", ...interaction }),
},
@@ -6813,12 +6820,12 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
issueId,
commentId: go!.id,
wakeReason: "issue_commented",
...(dedicatedDonor ? {} : interaction),
...(dedicatedDonor || persistentConversation ? {} : interaction),
},
});
expect(next).not.toBeNull();
expect(next?.contextSnapshot?.wakeCommentIds).toEqual([go!.id]);
if (!dedicatedDonor)
if (!dedicatedDonor && !persistentConversation)
expect(next?.contextSnapshot).toMatchObject(interaction);
const [retained] = await db
.select()
@@ -6830,6 +6837,7 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
});
expect(retained?.payload).toEqual(deferredPayload);
} finally {
if (persistentConversation) await instanceSettingsService(db).updateExperimental({ enableAgentChat: false });
// Keep this fixture's parked donor from being scheduled during teardown.
await db
.update(agents)
@@ -1,3 +1,4 @@
import { currentConversationCommentCondition } from "../../../services/agent-conversations.js";
import { getExecutionBlocker } from "../../../services/execution-blocker.js";
import { and, asc, eq, inArray, isNull, notInArray, or, sql } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
@@ -261,7 +262,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru
const rows = await tx
.select({ id: issueComments.id, deletedAt: issueComments.deletedAt, createdByRunId: issueComments.createdByRunId })
.from(issueComments)
.where(and(eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId), inArray(issueComments.id, queuedCommentIds)));
.where(and(eq(issueComments.companyId, companyId), eq(issueComments.issueId, issueId), inArray(issueComments.id, queuedCommentIds), currentConversationCommentCondition()));
const targetsFinishingRunAgent = wakeAgentId === finishingRunAgentId;
const liveNonSelfCommentIds = queuedCommentIds.filter((commentId) => {
const row = rows.find((candidate) => candidate.id === commentId);
@@ -384,6 +384,50 @@ describe("releaseIssueExecution", () => {
expect(promotedContextSnapshot.acceptedPlanWakeRouting).toEqual({ targetAgentId: "agent-1" });
});
it.each(["done", "cancelled"])("cancels stale assignee continuations before claiming promotion on %s tasks", async (status) => {
const queue = [wakeCandidate({ agentId: ISSUE.assigneeAgentId!, requestedByActorType: "agent" })];
const transaction = createFakeTransaction({
findNextDeferredWake: vi.fn(async () => queue.shift() ?? null),
});
const release = createReleaseIssueExecution({
issueLock: createFakeIssueLock(createFakeHost(), transaction, { ...ISSUE, status }),
recovery: createFakeRecovery(),
});
const result = await release({ companyId: RUN.companyId, runId: RUN.id, now: new Date() });
expect(transaction.cancelDeferredWake).toHaveBeenCalledWith(expect.objectContaining({
wakeId: "wake-1",
reason: "Deferred execution wake no longer applies to a terminal task",
}));
expect(transaction.claimDeferredWakeForPromotion).not.toHaveBeenCalled();
expect(transaction.finalizePromotedWake).not.toHaveBeenCalled();
expect(result.outcome.kind).toBe("released");
});
it("reopens a completed task before promoting its assignee's human follow-up", async () => {
const queue = [wakeCandidate({
agentId: ISSUE.assigneeAgentId!,
requestedByActorType: "user",
deferredCommentIds: ["human-follow-up"],
})];
const transaction = createFakeTransaction({
findNextDeferredWake: vi.fn(async () => queue.shift() ?? null),
reopenIssue: vi.fn(async () => ({ ...ISSUE, status: "todo" })),
});
const release = createReleaseIssueExecution({
issueLock: createFakeIssueLock(createFakeHost(), transaction, { ...ISSUE, status: "done" }),
recovery: createFakeRecovery(),
});
const result = await release({ companyId: RUN.companyId, runId: RUN.id, now: new Date() });
expect(transaction.cancelDeferredWake).not.toHaveBeenCalled();
expect(transaction.reopenIssue).toHaveBeenCalledTimes(1);
expect(transaction.finalizePromotedWake).toHaveBeenCalledTimes(1);
expect(result.outcome.kind).toBe("promoted");
});
it("never reopens the issue when the promotion claim loses the race, and moves on to the next wake", async () => {
const doneIssue: IssueSnapshot = { ...ISSUE, status: "done" };
const queue = [
@@ -271,21 +271,8 @@ async function promoteDeferredWake(
postCommitEffects: PostCommitEffect[],
input: ReleaseIssueExecutionInput,
): Promise<ReleaseTransactionResult | null> {
// Claim the wake for promotion before any other write in this branch
// (design choice: claim first, then reopen). A reopen write, or its
// `issue_reopened` post-commit effect, must never survive a lost race on
// this compare-and-set. When the claim fails, a concurrent writer already
// changed the wake's status, so this candidate is gone; the caller moves
// on to the next one instead of ending the drain.
const claimedForPromotion = await ports.transaction.claimDeferredWakeForPromotion({
companyId: run.companyId,
wakeId: workingCandidate.id,
now: input.now,
});
if (!claimedForPromotion) return null;
let currentIssue = issue;
let shouldReopen = false;
if (
!workingCandidate.authorizedFailedChatRetry &&
workingCandidate.deferredCommentIds.length > 0 &&
@@ -297,28 +284,56 @@ async function promoteDeferredWake(
finishingRunId: run.id,
commentIds: workingCandidate.deferredCommentIds,
});
const shouldReopen =
shouldReopen =
!selfAuthorship.allSelfAuthored &&
(workingCandidate.requestedByActorType === "user" ||
workingCandidate.wakeReason === "issue_reopened_via_comment");
if (shouldReopen) {
const reopened = await ports.transaction.reopenIssue({
companyId: run.companyId,
issueId: currentIssue.id,
}
// Agent continuations can outlive the work they addressed. Only a human
// reopen can revive assignee execution; other agents may still receive
// notifications about the closed task. Cancel before claiming promotion so
// the compare-and-set still sees the deferred wake.
if (
!shouldReopen &&
(currentIssue.status === "done" || currentIssue.status === "cancelled") &&
workingCandidate.agentId === currentIssue.assigneeAgentId
) {
await ports.transaction.cancelDeferredWake({
companyId: run.companyId,
wakeId: workingCandidate.id,
reason: "Deferred execution wake no longer applies to a terminal task",
now: input.now,
});
return null;
}
// Claim before reopening. A reopen write and its post-commit effect must
// never survive a lost race on this compare-and-set.
const claimedForPromotion = await ports.transaction.claimDeferredWakeForPromotion({
companyId: run.companyId,
wakeId: workingCandidate.id,
now: input.now,
});
if (!claimedForPromotion) return null;
if (shouldReopen) {
const reopened = await ports.transaction.reopenIssue({
companyId: run.companyId,
issueId: currentIssue.id,
runId: run.id,
});
if (reopened) {
postCommitEffects.push({
kind: "issue_reopened",
companyId: reopened.companyId,
agentId: invokableAgent.id,
runId: run.id,
issueId: reopened.id,
identifier: reopened.identifier,
reopenedFrom: currentIssue.status,
});
if (reopened) {
postCommitEffects.push({
kind: "issue_reopened",
companyId: reopened.companyId,
agentId: invokableAgent.id,
runId: run.id,
issueId: reopened.id,
identifier: reopened.identifier,
reopenedFrom: currentIssue.status,
});
currentIssue = reopened;
}
currentIssue = reopened;
}
}
+28 -6
View File
@@ -5569,9 +5569,12 @@ export function issueRoutes(
async function getIssueThreadInteractionResolutionAuthorization(
req: Request,
res: Response,
issue: Parameters<typeof assertAgentIssueMutationAllowed>[2],
issue: Parameters<typeof assertAgentIssueMutationAllowed>[2] & { conversationAgentId?: string | null; conversationUserId?: string | null },
interactionId: string,
) {
if (issue.conversationAgentId && req.actor.type === "board" && req.actor.userId !== issue.conversationUserId) {
throw forbidden("Only the conversation owner can respond to chat interactions");
}
// Actor-only gates deliberately precede the interaction lookup. An actor
// outside the issue's trusted/watchdog scope must not learn whether an
// interaction id exists on that issue.
@@ -16959,16 +16962,32 @@ export function issueRoutes(
if (issue.conversationAgentId && req.actor.type === "board") {
if (!(await instanceSettings.getExperimental()).enableAgentChat) throw notFound("Agent Chat is disabled");
if (!req.actor.userId) throw forbidden("Board user access required");
if (req.actor.userId !== issue.conversationUserId) throw forbidden("Only the conversation owner can send messages or start a new session");
if (!req.body.clientRequestId) throw unprocessable("Chat messages require a clientRequestId for safe retries");
if (!(await assertAgentIssueCommentAllowed(req, res, issue))) return;
if (req.body.body.trim() !== "/new" && !(await assertBoardCommentNotPaused(req, res, issue))) return;
const actor = getActorInfo(req);
const comment = await svc.addComment(issue.id, req.body.body, { userId: req.actor.userId }, {
clientRequestId: req.body.clientRequestId, authorType: "user", attachmentIds: req.body.attachmentIds,
const userId = req.actor.userId;
const publications: ActivityPublication[] = [];
const comment = await db.transaction(async (tx) => {
await tx.select({ id: issueRows.id }).from(issueRows).where(and(
eq(issueRows.id, issue.id), eq(issueRows.companyId, issue.companyId),
)).for("update");
const [existing] = await tx.select({ id: issueComments.id }).from(issueComments).where(and(
eq(issueComments.issueId, issue.id), eq(issueComments.authorUserId, userId),
eq(issueComments.clientRequestId, req.body.clientRequestId),
));
const saved = await svc.addComment(issue.id, req.body.body, { userId }, {
clientRequestId: req.body.clientRequestId, authorType: "user", attachmentIds: req.body.attachmentIds,
}, tx);
if (!existing) await logActivity(tx as unknown as Db, {
companyId: issue.companyId, actorType: actor.actorType, actorId: actor.actorId,
action: "issue.comment_added", entityType: "issue", entityId: issue.id,
details: { commentId: saved.id, identifier: issue.identifier },
}, publications);
return saved;
});
await logActivity(db, { companyId: issue.companyId, actorType: actor.actorType, actorId: actor.actorId,
action: "issue.comment_added", entityType: "issue", entityId: issue.id,
details: { commentId: comment.id, identifier: issue.identifier } });
for (const publication of publications) publishActivity(publication);
await issueReferencesSvc.syncComment(comment.id);
await deliverConversationComments(db, issue, heartbeat.wakeup);
res.status(201).json(comment);
@@ -18132,6 +18151,9 @@ export function issueRoutes(
if (issue.conversationAgentId && req.actor.type === "board" && !(await instanceSettings.getExperimental()).enableAgentChat) {
throw notFound("Agent Chat is disabled");
}
if (issue.conversationAgentId && req.actor.type === "board" && req.actor.userId !== issue.conversationUserId) {
throw forbidden("Only the conversation owner can upload attachments");
}
if (!(await assertAgentIssueMutationAllowed(req, res, issue))) return;
if (
!(await assertDeliverableMutationAllowedByRunContext(req, res, issue))
@@ -89,6 +89,21 @@ Create ordinary assigned tasks, never subtasks of this conversation. Give each t
Keep discussion here and leave the conversation available for the next message. Link handed-off tasks in your reply; do not make this conversation blocked by their completion or wait for them. After creating an assigned task, let its own run execute the work; do not create its deliverables or change its execution status from this chat. Reply normally and end your turn; Paperclip manages the conversation waiting state. Do not change its status, create a review confirmation just to finish a reply, mark it complete, or poll for another reply. An accepted plan authorizes handoff to execution tasks, never implementation on this conversation. Honor normal approvals. Ask mode is non-mutating. Plan mode supports research and writing/revising the plan; hand off for execution only through the normal authorized workflow.`;
/** A reset keeps history visible, but parked input from a stopped session cannot become a new turn. */
export function currentConversationCommentCondition() {
return sql`not exists (
select 1 from ${issues} conversation_issue
join ${issueComments} conversation_boundary
on conversation_boundary.id = conversation_issue.conversation_boundary_comment_id
and conversation_boundary.company_id = conversation_issue.company_id
and conversation_boundary.issue_id = conversation_issue.id
where conversation_issue.id = ${issueComments.issueId}
and conversation_issue.company_id = ${issueComments.companyId}
and conversation_issue.conversation_agent_id is not null
and (${issueComments.createdAt}, ${issueComments.id}) < (conversation_boundary.created_at, conversation_boundary.id)
)`;
}
/** Runs under the normal issue execution lock, before any provider session is read. */
export async function prepareConversationTurn(
db: Db,
+1 -1
View File
@@ -26579,7 +26579,7 @@ export function heartbeatService(
.then((rows) => rows[0]);
const pendingComments =
opts.allowRunCoalescing !== false &&
!isConversation(issue) && opts.allowRunCoalescing !== false &&
!(await getExecutionBlocker(tx as unknown as Db, issue.companyId, issue.id))
? await tx
.select()
+1 -1
View File
@@ -12025,7 +12025,7 @@ export function issueService(db: Db) {
if (!issue) throw notFound("Issue not found");
if (issue.conversationAgentId && actor.userId && !(await instanceSettings.getExperimental()).enableAgentChat) {
if (issue.conversationAgentId && actor.userId && !(await instanceSettingsService(dbOrTx).getExperimental()).enableAgentChat) {
throw unprocessable("Agent Chat is disabled in Experimental settings");
}
const currentUserRedactionOptions = {
@@ -5053,10 +5053,12 @@ describe("native session same-turn steering", () => {
describe("native warm session supervision", () => {
it.each([true, false])(
"preserves chat reply grace for per-turn providers: chat=%s",
"uses provider turn completion without a semantic-result cutoff: chat=%s",
async (conversationMode) => {
state.execute.mockReset().mockImplementationOnce(async (options) => {
expect(options.semanticResultTerminalGraceMs).toBe(conversationMode ? 30_000 : undefined);
// The provider must finish streaming its reply after task tools return.
// A semantic-result grace timer would truncate that output.
expect(options).not.toHaveProperty("semanticResultTerminalGraceMs");
return {
result: { summary: "Reply completed" },
terminal: { runTerminalState: "succeeded" },
@@ -503,11 +503,10 @@ const recoveryFakeCodex = resolve(
bundle = createCapabilityRunnerdCodexTransport({
stateDirectory: root,
sourceCodexHome: home,
codexCommand: resolve(
import.meta.dirname,
"../../../../packages/paperclip-runner/runner/target/debug/fake-codex-app-server",
),
codexArgs: ["--state-file", join(scratch, "fake.json"), "--hold-turn"],
// The packaged runnerd does not imply local Rust test binaries exist.
// Reuse the credential-free provider fixture available in every checkout.
codexCommand: process.execPath,
codexArgs: [recoveryFakeCodex, join(scratch, "fake.json"), "16"],
prpIdentity: {
runId,
runnerInstanceId,