mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
## Thinking Path > - Paperclip manages agents and their work. > - Task messages remain in a durable queue while an earlier run saves its result. > - That result can mark the task done before the user sends the saved messages. > - Send records current user intent, but its wake guard allowed only active task states. > - The old queue therefore remained pending with an obsolete finalization message. > - This change admits that exact Send receipt and reopens the task through the issue service in the successor transaction. > - The agent can receive the saved input once while the completed result remains unchanged. ## Linked Issues or Issue Description **What happened?** Send on a saved message can leave the message queued after its native predecessor succeeds and commits. The old wait message remains visible. Allowing admission alone also fails: continuation construction rejects a task that remains done. **Expected behavior** Explicit Send must deliver the saved input once through normal admission. It must retain the previous run and its result. **Steps to reproduce** 1. Save a follow-up while a native run finishes its result. 2. Let that run succeed, commit its accepted result, and mark the task done. 3. Send the saved follow-up. Before this change, the receipt stays deferred. Related public work: Refs #19542. That change selects recovered workspace files. This change handles the later saved-message admission. Refs #19602 for the independent workflow fixture repair required by CI. ## What Changed - Allow the exact explicit Send wake guard to accept a done task. Keep automatic recovery and cancelled task guards unchanged. - Reopen the task through the company-scoped issue service after admission gates pass. Commit reopening, queue consumption and the successor together. Publish the audit after commit. - Add a native succeeded/committed predecessor regression with unchanged saved work and one successor after two resume calls. - Verify real heartbeat startup under the clicking operator. Keep automatic input, cancelled tasks and paused agents held. - Check workspace preflight for the prospective reopened task. Use the normal company-scoped transition if preflight blocks a done task, so its completion metadata is cleared. - Check a successful native predecessor for its committed accepted result, current ownership and pending cleanup. Keep approvals and questions effective even without a recovery action. - Document the saved Send contract and update the queue test dependency. ## Verification - Final merge qualification for `675914f0fa9b5eb1d2c9c8fcd448e0669d0c4836`: all 56 applicable status checks pass, with no pending or failed checks. [Final CI](https://github.com/paperclipai/paperclip/actions/runs/42307623559) includes typecheck, build, all applicable test and E2E shards, and aggregate verification. Current-head Greptile is 5/5; all five review threads are resolved. Any paused or queued application review is not claimed as a new completed review. - Final base qualification at `675914f0fa9b5eb1d2c9c8fcd448e0669d0c4836`: official preflight passes (1.5s); all six integrated model-normalization cases pass (19.07s); both affected test typecheck layers pass. The integration changes no production server/shared/Runner/adapter inputs, so the preceding 12-case Send proof, two ordinary status guards, and unchanged production typecheck layers remain applicable. Fresh final-head CI supplies broad coverage. - New review regression: both failed and cancelled native predecessors on a task manually marked Done fail before the repair with `continuation_task_ownership_changed` through real `resumeQueuedCommentInterrupt` admission. At `bfb0eff7941090ac3ac43ce4366ba2e3d46612e9`, all 12 affected Send/source/cleanup/workspace/decision cases pass (246.80s), ordinary Done/cancelled rejection cases pass (2; 100.31s), and all seven server typecheck layers pass. History validation uses the prospective reopened status only for an already-verified explicit native Send. No status is changed during a dry run. - Prior `e64e4bb1` CI passed the affected serialized server suites, including the complete saved-message route suite, typecheck and build. General server shard 4 still failed two unrelated stale default-model expectations (`gpt-5.6-sol` versus the new `gpt-6.1-sol` default). The equivalent repair has already merged in #19610 as `8621bcf53ee2e4bbefb7d193add42dd544ca06e8`; the separately prepared #19645 is closed as superseded. This source fix does not weaken or edit those assertions. - Original RED: the committed-source and completed-task startup cases left the saved receipt deferred. - Review RED: Done plus an unrunnable isolated worktree created one successor instead of zero. A successful committed predecessor with pending lease cleanup also created one instead of zero. The new Done-to-blocked path retained stale completion time. Pending approvals and questions each allowed a successor. All failures were captured before their respective repairs. - Integrated production revision `99018264dd1ef002ff3500378184bcd167d9206b`: 417 affected cases pass across the complete native continuation, responsible-user startup, and workspace preflight suites (386 + 18 + 13). These include the original successful startup case and six cleanup/finalization faults that hold the exact saved receipt until settlement, then deliver it once. - `pnpm --filter @paperclipai/server typecheck` passes all seven layers on that integrated head, including the local Runner/SDK dependency build. Module-boundary, whitespace, and added-code-comment checks pass. - Normal master integrations preserve branch history. No force push or assertion weakening was used. Fresh CI and automated review are running on `675914f0fa9b5eb1d2c9c8fcd448e0669d0c4836`, which normally integrates the already-merged #19610 fixture repair after the reproduced inherited CI failure. - The earlier CI attempt failed in five unchanged workflow fixture cases because IMAGE_ONLY was unset. The merged #19602 repairs that fixture. A later inherited duplicate default in that independent fixture has a separate repair; this PR does not edit it. - Follow-up route-fixture RED: three existing native queue delivery cases failed because their fixture marked a run successful without an accepted result or committed finalization. The same three cases pass after the fixture records the complete native success contract (42.08s); delivery assertions and production gates are unchanged. The affected test-tree TypeScript layer also passes. The full route run passed all 119 assertions but failed its 30-second database teardown hook, so it is not recorded as a suite pass. That result is retained. One unchanged bounded rerun was stopped after more than 15 minutes without a final verdict; it is inconclusive, not a pass. - Final integrated head `e64e4bb1`: all 13 focused route and completed-task admission cases pass across two real-PostgreSQL suites (188.95s), including the three repaired route fixtures and every new cleanup/finalization/decision hold. Fresh preflight passes; all seven server typecheck layers pass. Required final CI supplies the broader route and server coverage. - These tests use real PostgreSQL and normal admission, cleanup, queue consumption, and startup services. The startup suite controls the provider boundary. Hosted provider acceptance remains pending normal deployment. ## Risks - Explicit Send can reopen a done task. This is limited to its validated saved receipt and current assignee. A cancelled task remains held. - Reopening uses the existing company-scoped issue transition and shares the successor transaction. A failed admission does not reopen the task. Provider Stop, workspace, approval, pause and budget checks retain their existing owners. - Cleanup and pending decisions retain the done task and saved receipt. An unrunnable workspace uses the existing blocked-task diagnostic. No schema or runtime package changes. ## Model Used - OpenAI Codex, GPT-6 family; exact serving snapshot/context window not exposed; repository analysis, code editing, tool execution. ## 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 - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
2198 lines
123 KiB
TypeScript
2198 lines
123 KiB
TypeScript
import { spawn, type ChildProcess } from "node:child_process";
|
|
import { runningProcesses } from "../adapters/index.js";
|
|
import { randomUUID } from "node:crypto";
|
|
import express from "express";
|
|
import request from "supertest";
|
|
import { eq, sql } from "drizzle-orm";
|
|
import { createFullDatabaseResponseCapture, installDatabaseResponseCapture } from "./helpers/full-database-response-capture.js";
|
|
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
|
|
import {
|
|
agentRuntimeState,
|
|
agentWakeupRequests,
|
|
agents,
|
|
activityLog,
|
|
companies,
|
|
companyMemberships,
|
|
companySkills,
|
|
completionContracts,
|
|
createDb,
|
|
environmentLeases,
|
|
heartbeatRuns,
|
|
heartbeatRunEvents,
|
|
issueComments,
|
|
issueThreadInteractions,
|
|
issueRecoveryActions,
|
|
nativeRunFinalizations,
|
|
nativeRunResults,
|
|
issues,
|
|
runIdentityContexts,
|
|
} from "@paperclipai/db";
|
|
import { errorHandler } from "../middleware/index.js";
|
|
import { issueRoutes } from "../routes/issues.js";
|
|
import { heartbeatService } from "../services/heartbeat.js";
|
|
import { issueService } from "../services/issues.js";
|
|
import { issueThreadInteractionService } from "../services/issue-thread-interactions.js";
|
|
import { hasQueuedInteractionResponse } from "../services/queued-interaction-response.js";
|
|
import { remoteTerminationReceipt } from "../services/remote-execution-termination.js";
|
|
import { initializeRunIdentity, reconcileSteeredIdentity } from "../services/run-identity.js";
|
|
import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js";
|
|
import {
|
|
getEmbeddedPostgresTestSupport,
|
|
startEmbeddedPostgresTestDatabase,
|
|
} from "./helpers/embedded-postgres.js";
|
|
|
|
const steerNativeSessionMock = vi.hoisted(() => vi.fn());
|
|
vi.mock("../services/native-runtime/native-session-executor.js", async (importOriginal) => {
|
|
const actual = await importOriginal<typeof import("../services/native-runtime/native-session-executor.js")>();
|
|
steerNativeSessionMock.mockImplementation(actual.steerNativeSession);
|
|
return { ...actual, steerNativeSession: steerNativeSessionMock };
|
|
});
|
|
const { NativeSessionSteeringError } = await import("../services/native-runtime/native-session-executor.js");
|
|
|
|
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
|
|
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
|
|
|
|
if (!embeddedPostgresSupport.supported) {
|
|
console.warn(
|
|
`Skipping queued-comment route tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`,
|
|
);
|
|
}
|
|
|
|
describeEmbeddedPostgres("issue queued-comment routes", () => {
|
|
let db!: ReturnType<typeof createDb>;
|
|
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
|
|
|
beforeAll(async () => {
|
|
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-queued-comments-");
|
|
db = createDb(tempDb.connectionString);
|
|
}, 30_000);
|
|
|
|
const testProcesses = new Map<string, ChildProcess>();
|
|
|
|
afterEach(async () => {
|
|
for (const [id, child] of testProcesses) {
|
|
runningProcesses.delete(id);
|
|
child.kill();
|
|
}
|
|
testProcesses.clear();
|
|
// Each case owns the entire disposable database. Clear the full company
|
|
// graph, including attribution rows and constraints added by migrations.
|
|
await db.execute(sql`TRUNCATE TABLE companies CASCADE`);
|
|
});
|
|
|
|
afterAll(async () => {
|
|
await db?.$client.end();
|
|
await tempDb?.cleanup();
|
|
});
|
|
|
|
function app(companyId: string, userId = "queue-owner", agentActor?: { agentId: string; runId: string }, database = db) {
|
|
const testApp = express();
|
|
testApp.use(express.json());
|
|
testApp.use((req, _res, next) => {
|
|
(req as any).actor = {
|
|
type: agentActor ? "agent" : "board",
|
|
...(agentActor ? { ...agentActor, companyId } : {}),
|
|
source: "session",
|
|
userId,
|
|
companyIds: [companyId],
|
|
memberships: [{ companyId, status: "active", membershipRole: "operator" }],
|
|
isInstanceAdmin: false,
|
|
};
|
|
next();
|
|
});
|
|
testApp.use("/api", issueRoutes(database, {} as any, {}));
|
|
testApp.use(errorHandler);
|
|
return testApp;
|
|
}
|
|
|
|
async function seedQueue() {
|
|
const companyId = randomUUID();
|
|
const agentId = randomUUID();
|
|
const issueId = randomUUID();
|
|
const runId = randomUUID();
|
|
const wakeId = randomUUID();
|
|
const commentIds = [randomUUID(), randomUUID()];
|
|
await db.insert(companies).values({
|
|
id: companyId,
|
|
name: "Queue Test Company",
|
|
issuePrefix: "QUE",
|
|
requireBoardApprovalForNewAgents: false,
|
|
});
|
|
await db.insert(agents).values({
|
|
id: agentId,
|
|
companyId,
|
|
name: "Paperclip Runner",
|
|
role: "engineer",
|
|
status: "idle",
|
|
adapterType: "paperclip_runner",
|
|
adapterConfig: {},
|
|
runtimeConfig: {},
|
|
permissions: {},
|
|
});
|
|
await db.insert(companyMemberships).values([
|
|
{
|
|
companyId,
|
|
principalType: "user",
|
|
principalId: "queue-owner",
|
|
status: "active",
|
|
membershipRole: "operator",
|
|
},
|
|
{
|
|
companyId,
|
|
principalType: "user",
|
|
principalId: "other-operator",
|
|
status: "active",
|
|
membershipRole: "operator",
|
|
},
|
|
]);
|
|
await db.insert(agentWakeupRequests).values({
|
|
id: wakeId,
|
|
companyId,
|
|
agentId,
|
|
source: "issue_comment",
|
|
reason: "Follow-up comments arrived during the active run",
|
|
status: "deferred_issue_execution",
|
|
requestedByActorType: "user",
|
|
requestedByActorId: "queue-owner",
|
|
payload: {
|
|
issueId,
|
|
commentId: commentIds[1],
|
|
_paperclipWakeContext: {
|
|
commentId: commentIds[1],
|
|
wakeCommentId: commentIds[1],
|
|
wakeCommentIds: commentIds,
|
|
},
|
|
},
|
|
});
|
|
await db.insert(heartbeatRuns).values({
|
|
id: runId,
|
|
companyId,
|
|
agentId,
|
|
invocationSource: "assignment",
|
|
triggerDetail: "queue route test",
|
|
status: "running",
|
|
runtimeMode: "native",
|
|
startedAt: new Date("2026-08-22T15:00:00.000Z"),
|
|
contextSnapshot: { issueId },
|
|
});
|
|
await db.insert(issues).values({
|
|
id: issueId,
|
|
companyId,
|
|
identifier: "QUE-1",
|
|
title: "Queued steering",
|
|
status: "in_progress",
|
|
priority: "medium",
|
|
assigneeAgentId: agentId,
|
|
executionRunId: runId,
|
|
});
|
|
await db.insert(issueComments).values(commentIds.map((id, index) => ({
|
|
id,
|
|
companyId,
|
|
issueId,
|
|
authorType: "user" as const,
|
|
authorUserId: "queue-owner",
|
|
body: index === 0 ? "First queued message" : "Second queued message",
|
|
createdAt: new Date(`2026-08-22T15:0${index + 1}:00.000Z`),
|
|
updatedAt: new Date(`2026-08-22T15:0${index + 1}:00.000Z`),
|
|
})));
|
|
return { companyId, agentId, issueId, runId, wakeId, commentIds };
|
|
}
|
|
|
|
async function commitNativeSuccess(f: Awaited<ReturnType<typeof seedQueue>>) {
|
|
const contractId = randomUUID(), resultId = randomUUID();
|
|
await db.insert(completionContracts).values({ id: contractId, companyId: f.companyId, issueId: f.issueId,
|
|
revision: 1, schemaVersion: "paperclip.completion-contract.v1", policyVersion: "test",
|
|
risk: "standard", completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal",
|
|
contractJson: {}, canonicalSha256: contractId, createdByActorType: "system", createdByActorId: "test" });
|
|
await db.update(heartbeatRuns).set({ nativeIssueId: f.issueId, completionContractId: contractId })
|
|
.where(eq(heartbeatRuns.id, f.runId));
|
|
await db.insert(nativeRunResults).values({ id: resultId, companyId: f.companyId, issueId: f.issueId,
|
|
runId: f.runId, completionContractId: contractId, serverFingerprint: resultId,
|
|
schemaStatus: "accepted", canonicalSha256: resultId,
|
|
resultJson: { terminal: { runTerminalState: "succeeded", reportedWorkDisposition: "completed" } } });
|
|
await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.issueId,
|
|
runId: f.runId, phase: "committed", resultId });
|
|
}
|
|
|
|
it.each([false, true])("polls queued comments without transferring run snapshots (agent fallback: %s)", async fallback => {
|
|
const f = await seedQueue(), payload = "saved-execution-input🙂".repeat(25_000);
|
|
await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issueId, prompt: payload },
|
|
runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner", adapterConfig: { instructions: payload } } },
|
|
stdoutExcerpt: payload, stderrExcerpt: payload,
|
|
}).where(eq(heartbeatRuns.id, f.runId));
|
|
if (fallback) await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, f.issueId));
|
|
const [before] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId));
|
|
const observed = createDb(tempDb!.connectionString), events: Array<Record<string, unknown>> = [], now = Date.now();
|
|
const capture = createFullDatabaseResponseCapture({ start: now - 100, end: now + 59_000 }, event => events.push(event));
|
|
installDatabaseResponseCapture(observed, { heartbeatRuns }, capture);
|
|
try {
|
|
const response = await request(app(f.companyId, "queue-owner", undefined, observed))
|
|
.get(`/api/issues/${f.issueId}/queued-comments`).expect(200);
|
|
expect(response.body.queueId).toBe(f.wakeId);
|
|
expect(response.body.targetRunId).toBe(f.runId);
|
|
expect(response.body.entries.map((entry: { comment: { id: string } }) => entry.comment.id)).toEqual(f.commentIds);
|
|
capture.flush();
|
|
const reads = events.filter(event => event.table === "heartbeat_runs" && event.operation === "select");
|
|
expect(reads.length).toBeGreaterThan(0);
|
|
expect(reads.reduce((bytes, event) => bytes + Number(event.responseValueBytes), 0)).toBeLessThan(2048);
|
|
await request(app(randomUUID(), "queue-owner", undefined, observed))
|
|
.get(`/api/issues/${f.issueId}/queued-comments`).expect(404);
|
|
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]).toEqual(before);
|
|
} finally { capture.flush(); await observed.$client.end(); }
|
|
});
|
|
|
|
it.each(["deferred", "queued"])("reads only queue display inputs from large %s wake payloads", async state => {
|
|
const f = await seedQueue();
|
|
if (state === "queued") await promoteQueue(f);
|
|
const client = app(f.companyId);
|
|
const path = `/api/issues/${f.issueId}/queued-comments`;
|
|
const expected = await request(client).get(path).expect(200);
|
|
const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.wakeId));
|
|
const payload = "saved-wake-input🙂".repeat(30_000);
|
|
const context = wake.payload?._paperclipWakeContext as Record<string, unknown>;
|
|
await db.update(agentWakeupRequests).set({ payload: {
|
|
...wake.payload, prompt: payload,
|
|
_paperclipWakeContext: { ...context, prompt: payload, nativeExecutionInput: { instructions: payload } },
|
|
} }).where(eq(agentWakeupRequests.id, f.wakeId));
|
|
await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issueId, prompt: payload },
|
|
runnerProfileJson: { instructions: payload }, resultJson: { output: payload },
|
|
}).where(eq(heartbeatRuns.companyId, f.companyId));
|
|
const [before] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.wakeId));
|
|
const observed = createDb(tempDb!.connectionString), events: Array<Record<string, unknown>> = [], now = Date.now();
|
|
const capture = createFullDatabaseResponseCapture({ start: now - 100, end: now + 59_000 }, event => events.push(event));
|
|
installDatabaseResponseCapture(observed, { agentWakeupRequests, heartbeatRuns }, capture);
|
|
try {
|
|
const response = await request(app(f.companyId, "queue-owner", undefined, observed)).get(path).expect(200);
|
|
expect(response.body).toEqual(expected.body);
|
|
capture.flush();
|
|
const reads = events.filter(event => ["agent_wakeup_requests", "heartbeat_runs"].includes(String(event.table)) && event.operation === "select");
|
|
expect(reads.some(event => event.table === "agent_wakeup_requests")).toBe(true);
|
|
expect(reads.reduce((bytes, event) => bytes + Number(event.responseValueBytes), 0)).toBeLessThan(4096);
|
|
expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.wakeId)))[0]).toEqual(before);
|
|
await request(app(randomUUID(), "queue-owner", undefined, observed)).get(path).expect(404);
|
|
} finally { capture.flush(); await observed.$client.end(); }
|
|
});
|
|
|
|
it.each(["deferred", "queued"])("filters unrelated task wakes before reading the %s comment queue", async state => {
|
|
const f = await seedQueue();
|
|
if (state === "queued") await promoteQueue(f);
|
|
const path = `/api/issues/${f.issueId}/queued-comments`;
|
|
const expected = await request(app(f.companyId)).get(path).expect(200);
|
|
const unrelated = Array.from({ length: 80 }, (_, index) => ({
|
|
companyId: f.companyId, agentId: f.agentId, source: "issue_comment",
|
|
status: index % 2 ? "queued" : "deferred_issue_execution",
|
|
requestedAt: new Date("2020-01-01T00:00:00Z"),
|
|
payload: { issueId: randomUUID(), _paperclipWakeContext: {
|
|
wakeCommentIds: Array.from({ length: 60 }, () => randomUUID()),
|
|
} },
|
|
}));
|
|
const malformed = [null, [f.issueId], { id: f.issueId }, f.issueId.toUpperCase()];
|
|
await db.insert(agentWakeupRequests).values([
|
|
...unrelated,
|
|
...malformed.map(issueId => ({ companyId: f.companyId, agentId: f.agentId,
|
|
source: "issue_comment", status: "deferred_issue_execution",
|
|
requestedAt: new Date("2020-01-01T00:00:00Z"),
|
|
payload: { issueId, _paperclipWakeContext: { wakeCommentIds: f.commentIds } },
|
|
})),
|
|
]);
|
|
const observed = createDb(tempDb!.connectionString), events: Array<Record<string, unknown>> = [], now = Date.now();
|
|
const capture = createFullDatabaseResponseCapture({ start: now - 100, end: now + 59_000 }, event => events.push(event));
|
|
installDatabaseResponseCapture(observed, { agentWakeupRequests }, capture);
|
|
try {
|
|
const response = await request(app(f.companyId, "queue-owner", undefined, observed)).get(path).expect(200);
|
|
expect(response.body).toEqual(expected.body);
|
|
capture.flush();
|
|
const reads = events.filter(event => event.table === "agent_wakeup_requests" && event.operation === "select");
|
|
expect(reads.length).toBeGreaterThan(0);
|
|
expect(reads.reduce((bytes, event) => bytes + Number(event.responseValueBytes), 0)).toBeLessThan(4096);
|
|
expect(reads.reduce((rows, event) => rows + Number(event.rows), 0)).toBe(1);
|
|
expect(await db.select({ id: agentWakeupRequests.id }).from(agentWakeupRequests)
|
|
.where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(85);
|
|
await request(app(randomUUID(), "queue-owner", undefined, observed)).get(path).expect(404);
|
|
} finally { capture.flush(); await observed.$client.end(); }
|
|
});
|
|
|
|
it.each(["cancelled", "running"])("rejects a late Done from an interrupted %s task run at the write boundary", async status => {
|
|
const seeded = await seedQueue();
|
|
await db.update(heartbeatRuns).set({ status, resultJson: { executionCancellation: { state: "requested" } } })
|
|
.where(eq(heartbeatRuns.id, seeded.runId));
|
|
// Model a request that passed middleware before cancellation and reached
|
|
// the service after the run lost its authority. Recovery may have released
|
|
// checkout and changed the task to Blocked in the meantime.
|
|
await db.update(issues).set({ status: "blocked", executionRunId: null }).where(eq(issues.id, seeded.issueId));
|
|
await expect(issueService(db).update(seeded.issueId, { status: "done",
|
|
actorAgentId: seeded.agentId, actorRunId: seeded.runId,
|
|
})).rejects.toMatchObject({ status: 403, details: { code: "agent_run_cancelled" } });
|
|
expect((await db.select().from(issues).where(eq(issues.id, seeded.issueId)))[0].status).toBe("blocked");
|
|
// A board disposition remains authoritative.
|
|
expect(await issueService(db).update(seeded.issueId, { status: "done", actorUserId: "queue-owner" }))
|
|
.toMatchObject({ status: "done" });
|
|
});
|
|
|
|
it("delivers a saved Grok interrupt exactly once after remote cleanup, including concurrent restart sweeps", async () => {
|
|
const seeded = await seedQueue();
|
|
await db.update(agents).set({ adapterType: "grok_local", runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } })
|
|
.where(eq(agents.id, seeded.agentId));
|
|
await db.update(heartbeatRuns).set({ runtimeMode: "legacy", status: "cancelled", errorCode: "operator_interrupted",
|
|
finishedAt: new Date("2026-08-22T15:03:00.000Z"),
|
|
contextSnapshot: { issueId: seeded.issueId, paperclipWorkspace: { remoteExecution: { transport: "sandbox" } } },
|
|
}).where(eq(heartbeatRuns.id, seeded.runId));
|
|
await db.update(issues).set({ status: "blocked", executionRunId: null }).where(eq(issues.id, seeded.issueId));
|
|
await db.insert(issueRecoveryActions).values({ companyId: seeded.companyId, sourceIssueId: seeded.issueId,
|
|
kind: "active_run_watchdog", cause: "legacy_execution_requires_reconciliation", fingerprint: seeded.runId,
|
|
status: "resolved", outcome: "blocked", nextAction: "Automatic recovery stopped.",
|
|
evidence: { runId: seeded.runId, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } },
|
|
});
|
|
const [lease] = await db.insert(environmentLeases).values({ companyId: seeded.companyId,
|
|
issueId: seeded.issueId, heartbeatRunId: seeded.runId, provider: "daytona", providerLeaseId: randomUUID(),
|
|
status: "active", leasePolicy: "ephemeral",
|
|
}).returning();
|
|
// Occupy the agent's slot on another task so this test never invokes a model.
|
|
await db.insert(heartbeatRuns).values({ companyId: seeded.companyId, agentId: seeded.agentId,
|
|
status: "running", contextSnapshot: { issueId: randomUUID() },
|
|
});
|
|
const client = app(seeded.companyId);
|
|
const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
|
await request(client).post(`/api/issues/${seeded.issueId}/queued-comments/interrupt`).send({
|
|
queueId: seeded.wakeId, revision: queue.body.revision, targetRunId: null,
|
|
}).expect(200);
|
|
expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId)))[0].status)
|
|
.toBe("deferred_issue_execution");
|
|
// Cleanup timestamps alone are not proof. A mismatched receipt is not proof either.
|
|
await db.update(environmentLeases).set({ status: "expired", releasedAt: new Date(), cleanupStatus: "success",
|
|
metadata: { remoteExecutionTermination: { ...remoteTerminationReceipt(lease, {
|
|
providerLeaseId: lease.providerLeaseId, state: "destroyed",
|
|
}), runId: randomUUID() } },
|
|
}).where(eq(environmentLeases.id, lease.id));
|
|
await heartbeatService(db).resumeQueuedCommentInterrupt(seeded.companyId, seeded.wakeId);
|
|
expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId)))[0].status)
|
|
.toBe("deferred_issue_execution");
|
|
await db.update(environmentLeases).set({ metadata: { remoteExecutionTermination: remoteTerminationReceipt(lease, {
|
|
providerLeaseId: lease.providerLeaseId, state: "destroyed",
|
|
}) } }).where(eq(environmentLeases.id, lease.id));
|
|
await db.update(agentWakeupRequests).set({ updatedAt: new Date(0) }).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
await Promise.all([heartbeatService(db).resumeQueuedRuns(), heartbeatService(db).resumeQueuedRuns()]);
|
|
const [delivered] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
expect(delivered.status).toBe("coalesced");
|
|
const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, delivered.runId!));
|
|
expect(successor.contextSnapshot).toMatchObject({ wakeCommentIds: seeded.commentIds, previousRunId: seeded.runId,
|
|
forceFreshSession: true });
|
|
await heartbeatService(db).resumeQueuedCommentInterrupt(seeded.companyId, seeded.wakeId);
|
|
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, seeded.companyId))).toHaveLength(3);
|
|
});
|
|
|
|
it("allows only the exact task mutation that stopped its own run to commit", async () => {
|
|
const seeded = await seedQueue();
|
|
const stopId = randomUUID();
|
|
await db.update(heartbeatRuns).set({ status: "cancelled", errorCode: "issue_reassigned",
|
|
resultJson: { reassignmentStopConfirmed: true, issueMutationStopId: stopId },
|
|
}).where(eq(heartbeatRuns.id, seeded.runId));
|
|
const data = { status: "todo", actorAgentId: seeded.agentId, actorRunId: seeded.runId };
|
|
await expect(issueService(db).update(seeded.issueId, { ...data, actorRunStopId: randomUUID() }))
|
|
.rejects.toMatchObject({ status: 403, details: { code: "agent_run_cancelled" } });
|
|
expect(await issueService(db).update(seeded.issueId, { ...data, actorRunStopId: stopId }))
|
|
.toMatchObject({ status: "todo" });
|
|
await expect(issueService(db).update(seeded.issueId, { ...data, status: "done" }))
|
|
.rejects.toMatchObject({ status: 403, details: { code: "agent_run_cancelled" } });
|
|
});
|
|
|
|
it.each(["questions", "verdicts"] as const)("rejects admitted %s responses after Stop revokes the run", async kind => {
|
|
const seeded = await seedQueue();
|
|
const service = issueThreadInteractionService(db);
|
|
const issue = { id: seeded.issueId, companyId: seeded.companyId };
|
|
const interaction = await service.create(issue, kind === "questions" ? {
|
|
kind: "ask_user_questions", resolverPolicy: "board_or_agents", payload: { version: 1,
|
|
questions: [{ id: "scope", prompt: "Choose scope", selectionMode: "single", options: [{ id: "first", label: "First" }] }],
|
|
},
|
|
} : {
|
|
kind: "request_item_verdicts", resolverPolicy: "board_or_agents", payload: { version: 1,
|
|
prompt: "Review the work",
|
|
items: [{ id: "work", label: "Work" }],
|
|
},
|
|
}, { userId: "queue-owner" });
|
|
await db.update(heartbeatRuns).set({ resultJson: { executionCancellation: { state: "requested" } } })
|
|
.where(eq(heartbeatRuns.id, seeded.runId));
|
|
const actor = { agentId: seeded.agentId, runId: seeded.runId };
|
|
const responding = kind === "questions"
|
|
? service.answerQuestions(issue, interaction.id, { answers: [{ questionId: "scope", optionIds: ["first"] }] }, actor)
|
|
: service.submitItemVerdicts(issue, interaction.id, { verdicts: [{ id: "work", verdict: "approve" }] }, actor);
|
|
await expect(responding).rejects.toMatchObject({ status: 403, details: { code: "agent_run_cancelled" } });
|
|
expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, interaction.id)))[0].status)
|
|
.toBe("pending");
|
|
});
|
|
|
|
it.each((["request_confirmation", "request_checkbox_confirmation", "ask_user_questions"] as const)
|
|
.flatMap(kind => (["legacy", "native"] as const).map(runtime => ({ kind, runtime }))))(
|
|
"queues $kind resolution while its $runtime source run is still running",
|
|
async ({ kind, runtime }) => {
|
|
const seeded = await seedQueue();
|
|
await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
if (runtime === "legacy") {
|
|
await db.update(agents).set({ adapterType: "codex_local" }).where(eq(agents.id, seeded.agentId));
|
|
await db.update(heartbeatRuns).set({ runtimeMode: "legacy" }).where(eq(heartbeatRuns.id, seeded.runId));
|
|
}
|
|
const child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { stdio: "ignore" });
|
|
testProcesses.set(seeded.runId, child);
|
|
runningProcesses.set(seeded.runId, { child, graceSec: 1, processGroupId: null });
|
|
const interactionId = randomUUID();
|
|
await db.insert(issueThreadInteractions).values({
|
|
id: interactionId, companyId: seeded.companyId, issueId: seeded.issueId,
|
|
kind, status: "pending", createdByAgentId: seeded.agentId, sourceRunId: seeded.runId,
|
|
continuationPolicy: "wake_assignee", requestedResolverPolicy: "human_only",
|
|
effectiveResolverPolicy: "human_only", resolverPolicyProvenance: "explicit",
|
|
payload: kind === "ask_user_questions"
|
|
? { version: 1, questions: [{ id: "runtime", prompt: "Which runtime?", selectionMode: "single", required: true,
|
|
options: [{ id: "node", label: "Node.js" }, { id: "python", label: "Python" }] }] }
|
|
: kind === "request_checkbox_confirmation"
|
|
? { version: 1, prompt: "Accept scope", options: [{ id: "scope", label: "Build the app" }] }
|
|
: { version: 1, prompt: "Build the app?", detailsMarkdown: "Create exactly one child task." },
|
|
});
|
|
const client = app(seeded.companyId);
|
|
await request(client).post(`/api/issues/${seeded.issueId}/interactions/${interactionId}/${kind === "ask_user_questions" ? "respond" : "accept"}`)
|
|
.send(kind === "ask_user_questions" ? { answers: [{ questionId: "runtime", optionIds: ["node"] }] }
|
|
: kind === "request_checkbox_confirmation" ? { selectedOptionIds: ["scope"] } : {}).expect(200);
|
|
await vi.waitFor(async () => {
|
|
const receipts = await db.select().from(agentWakeupRequests);
|
|
expect(receipts.some(row => row.status === "deferred_issue_execution")).toBe(true);
|
|
});
|
|
const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
|
expect(queue.body.entries).toHaveLength(1);
|
|
expect(queue.body.entries[0]).toMatchObject({
|
|
source: { kind: "interaction", interactionId, interactionKind: kind },
|
|
canEdit: false, canDiscard: false,
|
|
});
|
|
expect(queue.body.targetRunId).toBe(seeded.runId);
|
|
expect(queue.body.entries[0].comment.body).toContain(kind === "ask_user_questions" ? "Node.js" : "Accepted");
|
|
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)))[0]!.status).toBe("running");
|
|
const wakes = await db.select().from(agentWakeupRequests);
|
|
expect(wakes).toHaveLength(1);
|
|
expect(wakes[0]).toMatchObject({ status: "deferred_issue_execution", runId: null });
|
|
},
|
|
);
|
|
|
|
async function seedResponseQueue(native = false) {
|
|
const seeded = await seedQueue();
|
|
const interactionId = randomUUID();
|
|
await db.insert(issueThreadInteractions).values({
|
|
id: interactionId, companyId: seeded.companyId, issueId: seeded.issueId,
|
|
kind: "request_confirmation", status: "accepted", createdByAgentId: seeded.agentId,
|
|
sourceRunId: seeded.runId, resolvedByUserId: "queue-owner", resolvedAt: new Date(),
|
|
continuationPolicy: "wake_assignee", requestedResolverPolicy: "human_only",
|
|
effectiveResolverPolicy: "human_only", resolverPolicyProvenance: "explicit",
|
|
title: "Build the app", payload: { version: 1, prompt: "Build the app?" },
|
|
result: { version: 1, outcome: "accepted" },
|
|
});
|
|
const context = { issueId: seeded.issueId, interactionId, interactionKind: "request_confirmation",
|
|
interactionStatus: "accepted", planReviewInteraction: { id: interactionId,
|
|
acceptedTargetRevision: { documentId: "plan-1", revisionId: "revision-1" } } };
|
|
await db.update(agentWakeupRequests).set({ payload: { issueId: seeded.issueId,
|
|
mutation: "interaction", interactionId, interactionStatus: "accepted", _paperclipWakeContext: context },
|
|
}).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
if (!native) {
|
|
await db.update(agents).set({ adapterType: "claude_local",
|
|
runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } },
|
|
}).where(eq(agents.id, seeded.agentId));
|
|
await db.update(heartbeatRuns).set({ runtimeMode: "legacy" }).where(eq(heartbeatRuns.id, seeded.runId));
|
|
}
|
|
return { ...seeded, interactionId, context };
|
|
}
|
|
|
|
it("keeps messages and approvals in separate durable queues in either arrival order", async () => {
|
|
const seeded = await seedResponseQueue();
|
|
const child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { stdio: "ignore" });
|
|
testProcesses.set(seeded.runId, child);
|
|
runningProcesses.set(seeded.runId, { child, graceSec: 1, processGroupId: null });
|
|
const client = app(seeded.companyId);
|
|
await request(client).post(`/api/issues/${seeded.issueId}/comments`)
|
|
.send({ body: "Keep this message too" }).expect(201);
|
|
await vi.waitFor(async () => expect(await db.select().from(agentWakeupRequests)).toHaveLength(2));
|
|
let wakes = await db.select().from(agentWakeupRequests);
|
|
expect(wakes.find(w => w.id === seeded.wakeId)?.payload).not.toHaveProperty("commentId");
|
|
// A second resolved card must not overwrite either existing receipt.
|
|
const nextId = randomUUID();
|
|
await db.insert(issueThreadInteractions).values({
|
|
id: nextId, companyId: seeded.companyId, issueId: seeded.issueId,
|
|
kind: "request_confirmation", status: "pending", createdByAgentId: seeded.agentId,
|
|
sourceRunId: seeded.runId, continuationPolicy: "wake_assignee",
|
|
requestedResolverPolicy: "human_only", effectiveResolverPolicy: "human_only",
|
|
resolverPolicyProvenance: "explicit", payload: { version: 1, prompt: "Second approval?" },
|
|
});
|
|
await request(client).post(`/api/issues/${seeded.issueId}/interactions/${nextId}/accept`).send({}).expect(200);
|
|
wakes = await db.select().from(agentWakeupRequests);
|
|
expect(wakes).toHaveLength(3);
|
|
expect(wakes.filter(w => w.payload?.mutation === "interaction").map(w => w.payload?.interactionId).sort())
|
|
.toEqual([seeded.interactionId, nextId].sort());
|
|
expect(wakes.find(w => w.payload?.commentId)?.payload?._paperclipWakeContext).not.toHaveProperty("interactionId");
|
|
});
|
|
|
|
it("recovers an approval acknowledged before the steering transaction failed without redelivery", async () => {
|
|
const seeded = await seedResponseQueue(true);
|
|
await seedDispatchIdentity(seeded);
|
|
// Reconstruct the durable state after acknowledgement and HTTP rollback.
|
|
const [pending] = await db.insert(runIdentityContexts).values({
|
|
companyId: seeded.companyId, runId: seeded.runId, revision: 2,
|
|
cause: "steering", correlationId: `interaction:${seeded.interactionId}`,
|
|
messageId: seeded.interactionId, responsibleUserId: "queue-owner", status: "pending",
|
|
}).returning();
|
|
await db.insert(heartbeatRunEvents).values({
|
|
companyId: seeded.companyId, runId: seeded.runId, agentId: seeded.agentId,
|
|
seq: 1, eventType: "item.completed", sourceEventId: randomUUID(),
|
|
payload: { prpEvent: { turnId: "approval-ack", itemId: `approval-ack:steer:${seeded.interactionId}`,
|
|
payload: { kind: "steering_acknowledgement" } } },
|
|
});
|
|
const client = app(seeded.companyId);
|
|
const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
|
const body = { queueId: seeded.wakeId, targetRunId: seeded.runId, revision: queue.body.revision };
|
|
const endpoint = `/api/issues/${seeded.issueId}/queued-comments/${seeded.interactionId}/steer`;
|
|
const count = steerNativeSessionMock.mock.calls.length;
|
|
await request(client).post(endpoint).send(body).expect(200);
|
|
expect(steerNativeSessionMock.mock.calls).toHaveLength(count);
|
|
expect((await db.select().from(runIdentityContexts).where(eq(runIdentityContexts.id, pending.id)))[0].status).toBe("accepted");
|
|
});
|
|
|
|
it("interrupts an active legacy run and delivers the exact approval once", async () => {
|
|
const seeded = await seedResponseQueue();
|
|
const child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { stdio: "ignore" });
|
|
testProcesses.set(seeded.runId, child);
|
|
runningProcesses.set(seeded.runId, { child, graceSec: 1, processGroupId: null });
|
|
await db.update(heartbeatRuns).set({ processPid: child.pid }).where(eq(heartbeatRuns.id, seeded.runId));
|
|
// Occupy another task to inspect admission without invoking a paid provider.
|
|
await db.insert(heartbeatRuns).values({ companyId: seeded.companyId, agentId: seeded.agentId,
|
|
status: "running", contextSnapshot: { issueId: randomUUID() } });
|
|
const client = app(seeded.companyId);
|
|
const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
|
await request(client).post(`/api/issues/${seeded.issueId}/queued-comments/interrupt`)
|
|
.send({ queueId: seeded.wakeId, targetRunId: seeded.runId, revision: queue.body.revision }).expect(200);
|
|
runningProcesses.delete(seeded.runId);
|
|
await heartbeatService(db).resumeQueuedCommentInterrupt(seeded.companyId, seeded.wakeId);
|
|
const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
expect(["queued", "coalesced"]).toContain(wake.status);
|
|
const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, wake.runId!));
|
|
expect(successor.status).toBe("queued");
|
|
expect(successor.contextSnapshot).toMatchObject(seeded.context);
|
|
await heartbeatService(db).resumeQueuedCommentInterrupt(seeded.companyId, seeded.wakeId);
|
|
expect(await db.select().from(heartbeatRuns)).toHaveLength(3);
|
|
});
|
|
|
|
it("steers a saved approval only on click and acknowledges retries once", async () => {
|
|
const seeded = await seedResponseQueue(true);
|
|
await seedDispatchIdentity(seeded);
|
|
steerNativeSessionMock.mockResolvedValueOnce({ turnId: "approval-turn" });
|
|
const client = app(seeded.companyId);
|
|
const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
|
const body = { queueId: seeded.wakeId, targetRunId: seeded.runId, revision: queue.body.revision };
|
|
const endpoint = `/api/issues/${seeded.issueId}/queued-comments/${seeded.interactionId}/steer`;
|
|
await request(client).post(endpoint).send(body).expect(200);
|
|
const delivered = steerNativeSessionMock.mock.calls.at(-1)![0];
|
|
expect(delivered.message).toContain(seeded.interactionId);
|
|
expect(delivered.message).toContain('"status": "accepted"');
|
|
expect((await db.select().from(runIdentityContexts).where(eq(runIdentityContexts.messageId, seeded.interactionId)))[0])
|
|
.toMatchObject({ status: "accepted", responsibleUserId: "queue-owner", cause: "steering" });
|
|
const callCount = steerNativeSessionMock.mock.calls.length;
|
|
await request(client).post(endpoint).send(body).expect(200);
|
|
expect(steerNativeSessionMock.mock.calls).toHaveLength(callCount);
|
|
expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId)))[0].status).toBe("cancelled");
|
|
expect(await db.select().from(issueComments)).toHaveLength(2); // no forged approval comment
|
|
});
|
|
|
|
it.each([false, true])("lets the source run finish review after acceptance, before wake persistence: %s", async (beforeWake) => {
|
|
const seeded = await seedResponseQueue();
|
|
if (beforeWake) await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
const client = app(seeded.companyId, "queue-owner", { agentId: seeded.agentId, runId: seeded.runId });
|
|
const result = await request(client).patch(`/api/issues/${seeded.issueId}`).send({ status: "in_review", reviewInteractionId: seeded.interactionId });
|
|
expect(result.status, JSON.stringify(result.body)).toBe(200);
|
|
const [task] = await db.select().from(issues).where(eq(issues.id, seeded.issueId));
|
|
expect(task.assigneeAgentId).toBe(seeded.agentId);
|
|
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)))[0].status).toBe("running");
|
|
});
|
|
|
|
it.each(["legacy", "native"] as const)("resumes a queued plan approval across a stopped %s run only after cleanup", async (runtime) => {
|
|
const seeded = await seedResponseQueue(runtime === "native");
|
|
await db.update(agents).set({ runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }).where(eq(agents.id, seeded.agentId));
|
|
const [receipt] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
await db.update(agentWakeupRequests).set({ payload: { ...receipt.payload,
|
|
_paperclipWakeContext: { ...seeded.context, forceFreshSession: true } },
|
|
}).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date(), errorCode: "process_lost", processPid: process.pid })
|
|
.where(eq(heartbeatRuns.id, seeded.runId));
|
|
await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, seeded.issueId));
|
|
await db.insert(issueRecoveryActions).values({ companyId: seeded.companyId, sourceIssueId: seeded.issueId,
|
|
kind: "active_run_watchdog", cause: runtime === "legacy" ? "legacy_execution_requires_reconciliation" : "native_continuation_requires_reconciliation",
|
|
fingerprint: seeded.runId, status: "resolved", outcome: "blocked", nextAction: "Inspect stopped execution",
|
|
evidence: { runId: seeded.runId, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } } });
|
|
await db.insert(heartbeatRuns).values({ companyId: seeded.companyId, agentId: seeded.agentId,
|
|
status: "running", contextSnapshot: { issueId: randomUUID() } });
|
|
const client = app(seeded.companyId);
|
|
const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
|
await request(client).post(`/api/issues/${seeded.issueId}/queued-comments/interrupt`)
|
|
.send({ queueId: seeded.wakeId, targetRunId: null, revision: queue.body.revision }).expect(200);
|
|
const [waiting] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
expect(waiting.status).toBe("deferred_issue_execution");
|
|
expect(waiting.payload?.executionWait).toMatchObject({ reason: "process_running" });
|
|
await db.update(heartbeatRuns).set({ processPid: 999999999 }).where(eq(heartbeatRuns.id, seeded.runId));
|
|
await heartbeatService(db).resumeQueuedCommentInterrupt(seeded.companyId, seeded.wakeId);
|
|
const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
expect(wake.status).toBe("coalesced");
|
|
const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, wake.runId!));
|
|
expect(successor.contextSnapshot).toMatchObject({ ...seeded.context, forceFreshSession: true, previousRunId: seeded.runId });
|
|
await heartbeatService(db).resumeQueuedCommentInterrupt(seeded.companyId, seeded.wakeId);
|
|
expect(await db.select().from(heartbeatRuns)).toHaveLength(3);
|
|
});
|
|
|
|
it("keeps queued approval content immutable through comment mutation endpoints", async () => {
|
|
const seeded = await seedResponseQueue();
|
|
const client = app(seeded.companyId);
|
|
const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
|
const body = { queueId: seeded.wakeId, revision: queue.body.revision };
|
|
await request(client).patch(`/api/issues/${seeded.issueId}/queued-comments/${seeded.interactionId}`)
|
|
.send({ ...body, body: "Reject instead" }).expect(409);
|
|
await request(client).delete(`/api/issues/${seeded.issueId}/queued-comments/${seeded.interactionId}`).send(body).expect(409);
|
|
await request(client).put(`/api/issues/${seeded.issueId}/queued-comments/order`)
|
|
.send({ ...body, orderedCommentIds: [seeded.interactionId] }).expect(409);
|
|
expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, seeded.interactionId)))[0].status).toBe("accepted");
|
|
expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId)))[0].status).toBe("deferred_issue_execution");
|
|
});
|
|
|
|
it("does not let a viewer steer someone else's approval", async () => {
|
|
const seeded = await seedResponseQueue(true);
|
|
const client = app(seeded.companyId);
|
|
const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
|
await db.update(companyMemberships).set({ membershipRole: "viewer" }).where(eq(companyMemberships.principalId, "queue-owner"));
|
|
const calls = steerNativeSessionMock.mock.calls.length;
|
|
await request(client).post(`/api/issues/${seeded.issueId}/queued-comments/${seeded.interactionId}/steer`)
|
|
.send({ queueId: seeded.wakeId, targetRunId: seeded.runId, revision: queue.body.revision }).expect(403);
|
|
expect(steerNativeSessionMock.mock.calls).toHaveLength(calls);
|
|
});
|
|
|
|
it("does not steer a plan approval that requires a fresh execution session", async () => {
|
|
const seeded = await seedResponseQueue(true);
|
|
const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
await db.update(agentWakeupRequests).set({ payload: { ...wake.payload,
|
|
_paperclipWakeContext: { ...seeded.context, forceFreshSession: true } } }).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
const client = app(seeded.companyId);
|
|
const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
|
expect(queue.body.entries[0].source.requiresFreshSession).toBe(true);
|
|
const calls = steerNativeSessionMock.mock.calls.length;
|
|
const result = await request(client).post(`/api/issues/${seeded.issueId}/queued-comments/${seeded.interactionId}/steer`)
|
|
.send({ queueId: seeded.wakeId, targetRunId: seeded.runId, revision: queue.body.revision }).expect(409);
|
|
expect(result.body.details.code).toBe("queued_response_requires_fresh_session");
|
|
expect(steerNativeSessionMock.mock.calls).toHaveLength(calls);
|
|
});
|
|
|
|
it("rejects stale agent handback without cancelling the accepted proposal's run", async () => {
|
|
const seeded = await seedResponseQueue();
|
|
await db.update(issues).set({ createdByUserId: "queue-owner" }).where(eq(issues.id, seeded.issueId));
|
|
const client = app(seeded.companyId, "queue-owner", { agentId: seeded.agentId, runId: seeded.runId });
|
|
const result = await request(client).patch(`/api/issues/${seeded.issueId}`)
|
|
.send({ status: "in_review", assigneeAgentId: null, assigneeUserId: "queue-owner" }).expect(409);
|
|
expect(result.body.details.code).toBe("interaction_response_queued");
|
|
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)))[0].status).toBe("running");
|
|
});
|
|
|
|
// N-01: a redundant continuation wake (for example a recovery sweep that raced
|
|
// the response's own wake) carries a receipt the agent already received. It
|
|
// must not sit in the composer queue as an immutable, unremovable item.
|
|
const deliveredReceiptCases = [
|
|
{ kind: "ask_user_questions", status: "answered", title: null, label: "Answered: Questions",
|
|
payload: { version: 1, questions: [{ id: "goal", prompt: "What are you building?", selectionMode: "single", required: true,
|
|
options: [{ id: "shop", label: "A shop" }] }] }, result: { version: 1, answers: [{ questionId: "goal", optionIds: ["shop"] }] } },
|
|
{ kind: "request_confirmation", status: "accepted", title: "Approve hiring 2 agents", label: "Accepted: Approve hiring 2 agents",
|
|
payload: { version: 1, prompt: "Hire 2 agents?" }, result: { version: 1, outcome: "accepted" } },
|
|
{ kind: "request_confirmation", status: "rejected", title: "Approve hiring 2 agents", label: "Rejected: Approve hiring 2 agents",
|
|
payload: { version: 1, prompt: "Hire 2 agents?" }, result: { version: 1, outcome: "rejected" } },
|
|
{ kind: "request_checkbox_confirmation", status: "accepted", title: "Approve storefront prototype plan",
|
|
label: "Accepted: Approve storefront prototype plan",
|
|
payload: { version: 1, prompt: "Accept scope", options: [{ id: "scope", label: "Build the app" }] },
|
|
result: { version: 1, outcome: "accepted", selectedOptionIds: ["scope"] } },
|
|
{ kind: "connection_intent", status: "accepted", title: "Connect GitHub", label: "Connected: GitHub connection",
|
|
payload: { version: 1, serviceName: "GitHub" }, result: { version: 1, outcome: "connected" } },
|
|
{ kind: "connection_intent", status: "rejected", title: "Connect Slack", label: "Declined: Slack connection",
|
|
payload: { version: 1, serviceName: "Slack" }, result: { version: 1, outcome: "declined" } },
|
|
] as const;
|
|
|
|
async function seedDeliveredReceipt(
|
|
c: (typeof deliveredReceiptCases)[number],
|
|
delivering: "succeeded" | "running" | "failed" | "cancelled" | null,
|
|
runContext: "response" | "card_creation" = "response",
|
|
) {
|
|
const seeded = await seedQueue();
|
|
await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
const interactionId = randomUUID();
|
|
const resolvedAt = new Date("2026-08-22T15:05:00.000Z");
|
|
await db.insert(issueThreadInteractions).values({
|
|
id: interactionId, companyId: seeded.companyId, issueId: seeded.issueId,
|
|
kind: c.kind, status: c.status, createdByAgentId: seeded.agentId, sourceRunId: seeded.runId,
|
|
resolvedByUserId: "queue-owner", resolvedAt,
|
|
continuationPolicy: "wake_assignee", requestedResolverPolicy: "human_only",
|
|
effectiveResolverPolicy: "human_only", resolverPolicyProvenance: "explicit",
|
|
// Fixture payloads cover several interaction kinds; the schema types each kind separately.
|
|
title: c.title, payload: c.payload as never, result: c.result as never,
|
|
});
|
|
const context = { issueId: seeded.issueId, interactionId, interactionKind: c.kind, interactionStatus: c.status };
|
|
// The response's own wake already started a run that carries the answer.
|
|
const deliveringRunId = randomUUID();
|
|
if (delivering) {
|
|
await db.insert(heartbeatRuns).values({
|
|
id: deliveringRunId, companyId: seeded.companyId, agentId: seeded.agentId,
|
|
invocationSource: "automation", triggerDetail: "system", status: delivering, runtimeMode: "native",
|
|
startedAt: new Date("2026-08-22T15:05:01.000Z"),
|
|
finishedAt: delivering === "running" ? null : new Date("2026-08-22T15:06:00.000Z"),
|
|
// The card-creation wake carries the same interaction id while the
|
|
// card is still pending, without a resolved status.
|
|
contextSnapshot: runContext === "card_creation"
|
|
? { issueId: seeded.issueId, interactionId, interactionKind: c.kind, wakeReason: "interaction_pending" }
|
|
: { ...context, source: "issue.interaction.respond", wakeReason: "issue_commented" },
|
|
});
|
|
}
|
|
// The redundant continuation, still deferred behind the issue's active run.
|
|
const staleWakeId = randomUUID();
|
|
await db.insert(agentWakeupRequests).values({
|
|
id: staleWakeId, companyId: seeded.companyId, agentId: seeded.agentId,
|
|
source: "automation", triggerDetail: "system", reason: "issue_continuation_needed",
|
|
status: "deferred_issue_execution", requestedByActorType: "system", requestedByActorId: "recovery",
|
|
requestedAt: new Date("2026-08-22T15:05:02.000Z"),
|
|
payload: { issueId: seeded.issueId, mutation: "interaction", interactionId, interactionStatus: c.status,
|
|
_paperclipWakeContext: { ...context, source: "issue.interaction_continuation_recovery" } },
|
|
});
|
|
return { ...seeded, interactionId, staleWakeId, deliveringRunId };
|
|
}
|
|
|
|
it.each(deliveredReceiptCases.flatMap(c => (["succeeded", "running"] as const).map(delivering => ({ ...c, delivering }))))(
|
|
"clears a $kind $status receipt once a $delivering run already delivered it",
|
|
async (c) => {
|
|
const seeded = await seedDeliveredReceipt(c, c.delivering);
|
|
const client = app(seeded.companyId);
|
|
const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
|
expect(queue.body.entries.map((entry: { source?: { summary?: string } }) => entry.source?.summary)).toEqual([]);
|
|
},
|
|
);
|
|
|
|
it.each(deliveredReceiptCases)("keeps a pending $kind $status receipt until a run delivers it", async (c) => {
|
|
const seeded = await seedDeliveredReceipt(c, null);
|
|
const queue = await request(app(seeded.companyId)).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
|
expect(queue.body.entries).toHaveLength(1);
|
|
expect(queue.body.entries[0]).toMatchObject({
|
|
source: { kind: "interaction", interactionId: seeded.interactionId, interactionKind: c.kind, summary: c.label },
|
|
canEdit: false, canDiscard: false,
|
|
});
|
|
});
|
|
|
|
it.each([
|
|
{ delivering: "failed", runContext: "response" },
|
|
{ delivering: "cancelled", runContext: "response" },
|
|
{ delivering: "succeeded", runContext: "card_creation" },
|
|
{ delivering: "running", runContext: "card_creation" },
|
|
] as const)("keeps a receipt owed after a $delivering $runContext run", async ({ delivering, runContext }) => {
|
|
const seeded = await seedDeliveredReceipt(deliveredReceiptCases[0], delivering, runContext);
|
|
const queue = await request(app(seeded.companyId)).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
|
expect(queue.body.entries.map((entry: { source?: { summary?: string } }) => entry.source?.summary))
|
|
.toEqual(["Answered: Questions"]);
|
|
expect(await hasQueuedInteractionResponse(db, seeded.companyId, seeded.issueId, seeded.agentId)).toBe(true);
|
|
});
|
|
|
|
it("shows a later undelivered receipt instead of an earlier delivered one", async () => {
|
|
const seeded = await seedDeliveredReceipt(deliveredReceiptCases[3], "succeeded");
|
|
const pendingId = randomUUID();
|
|
await db.insert(issueThreadInteractions).values({
|
|
id: pendingId, companyId: seeded.companyId, issueId: seeded.issueId,
|
|
kind: "connection_intent", status: "rejected", createdByAgentId: seeded.agentId,
|
|
resolvedByUserId: "queue-owner", resolvedAt: new Date("2026-08-22T15:10:00.000Z"),
|
|
continuationPolicy: "wake_assignee", requestedResolverPolicy: "human_only",
|
|
effectiveResolverPolicy: "human_only", resolverPolicyProvenance: "explicit",
|
|
title: "Connect Slack", payload: { version: 1, serviceName: "Slack" } as never, result: { version: 1, outcome: "declined" } as never,
|
|
});
|
|
await db.insert(agentWakeupRequests).values({
|
|
companyId: seeded.companyId, agentId: seeded.agentId, source: "automation", triggerDetail: "system",
|
|
reason: "issue_commented", status: "deferred_issue_execution", requestedAt: new Date("2026-08-22T16:00:00.000Z"),
|
|
payload: { issueId: seeded.issueId, mutation: "interaction", interactionId: pendingId, interactionStatus: "rejected" },
|
|
});
|
|
const queue = await request(app(seeded.companyId)).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
|
expect(queue.body.entries.map((entry: { source?: { summary?: string } }) => entry.source?.summary))
|
|
.toEqual(["Declined: Slack connection"]);
|
|
});
|
|
|
|
it("does not hold agent handback for a receipt the agent already received", async () => {
|
|
const pending = await seedDeliveredReceipt(deliveredReceiptCases[1], null);
|
|
expect(await hasQueuedInteractionResponse(db, pending.companyId, pending.issueId, pending.agentId)).toBe(true);
|
|
await db.execute(sql`TRUNCATE TABLE companies CASCADE`);
|
|
const delivered = await seedDeliveredReceipt(deliveredReceiptCases[1], "succeeded");
|
|
expect(await hasQueuedInteractionResponse(db, delivered.companyId, delivered.issueId, delivered.agentId)).toBe(false);
|
|
});
|
|
|
|
it("promotes an approval after normal completion without a click", async () => {
|
|
const seeded = await seedResponseQueue();
|
|
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, seeded.runId));
|
|
await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, seeded.issueId));
|
|
await db.insert(heartbeatRuns).values({ companyId: seeded.companyId, agentId: seeded.agentId,
|
|
status: "running", contextSnapshot: { issueId: randomUUID() } });
|
|
await heartbeatService(db).resumeQueuedRuns();
|
|
const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
expect(wake.status).toBe("queued");
|
|
const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, wake.runId!));
|
|
expect(successor.contextSnapshot).toMatchObject(seeded.context);
|
|
await heartbeatService(db).resumeQueuedRuns();
|
|
expect(await db.select().from(heartbeatRuns)).toHaveLength(3);
|
|
});
|
|
|
|
it.each(["stale revision", "different issue"] as const)(
|
|
"rejects queued interruption for a %s without stopping the run",
|
|
async (scenario) => {
|
|
const seeded = await seedQueue();
|
|
await db.update(agents).set({ adapterType: "codex_local" }).where(eq(agents.id, seeded.agentId));
|
|
await db.update(heartbeatRuns).set({ runtimeMode: "legacy" }).where(eq(heartbeatRuns.id, seeded.runId));
|
|
const client = app(seeded.companyId);
|
|
const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
|
if (scenario === "different issue") {
|
|
await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: randomUUID() } }).where(eq(heartbeatRuns.id, seeded.runId));
|
|
}
|
|
await request(client).post(`/api/issues/${seeded.issueId}/queued-comments/interrupt`).send({
|
|
queueId: seeded.wakeId, targetRunId: seeded.runId,
|
|
revision: scenario === "stale revision" ? "stale" : queue.body.revision,
|
|
}).expect(409);
|
|
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)))[0]!.status).toBe("running");
|
|
},
|
|
);
|
|
|
|
it.each(["preparing", "native"] as const)(
|
|
"accepts an explicit interrupt fallback for a %s runner and preserves queued messages",
|
|
async (runtime) => {
|
|
const seeded = await seedQueue();
|
|
if (runtime === "preparing") {
|
|
await db.update(heartbeatRuns).set({ runtimeMode: "legacy", runtimeModeResolvedAt: null,
|
|
executionStage: "preparing",
|
|
runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } },
|
|
}).where(eq(heartbeatRuns.id, seeded.runId));
|
|
} else {
|
|
await db.update(heartbeatRuns).set({ nativeIssueId: seeded.issueId })
|
|
.where(eq(heartbeatRuns.id, seeded.runId));
|
|
await db.insert(nativeRunFinalizations).values({ companyId: seeded.companyId,
|
|
issueId: seeded.issueId, runId: seeded.runId, phase: "observed" });
|
|
}
|
|
// Hold the successor in the ordinary queue without launching a provider.
|
|
await db.update(agents).set({ runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } })
|
|
.where(eq(agents.id, seeded.agentId));
|
|
await db.insert(heartbeatRuns).values({ companyId: seeded.companyId, agentId: seeded.agentId,
|
|
status: "running", contextSnapshot: { issueId: randomUUID() } });
|
|
const client = app(seeded.companyId);
|
|
const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
|
expect(queue.body).toMatchObject({ protocol: "paperclip_runner_v1", steeringDisposition: "temporarily_unavailable" });
|
|
const response = await request(client).post(`/api/issues/${seeded.issueId}/queued-comments/interrupt`).send({
|
|
queueId: seeded.wakeId, targetRunId: seeded.runId, revision: queue.body.revision,
|
|
});
|
|
expect(response.status, JSON.stringify(response.body)).toBe(200);
|
|
const [waiting] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
expect(waiting.payload?.queuedCommentInterrupt).toMatchObject({ actorId: "queue-owner", targetRunId: seeded.runId });
|
|
expect(await db.select().from(issueComments).where(eq(issueComments.issueId, seeded.issueId))).toHaveLength(2);
|
|
expect(waiting.status).toBe("deferred_issue_execution");
|
|
// The synthetic preparation owner now confirms that its unwind finished.
|
|
// No provider ran; until this receipt exists, cancellation must hold input.
|
|
const [stopped] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId));
|
|
expect(stopped.status).toBe("cancelled");
|
|
await db.update(heartbeatRuns).set({ resultJson: { ...stopped.resultJson,
|
|
startupPreparationSettledAt: new Date().toISOString() } }).where(eq(heartbeatRuns.id, seeded.runId));
|
|
await heartbeatService(db).resumeQueuedCommentInterrupt(seeded.companyId, seeded.wakeId);
|
|
const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
expect(wake.status, JSON.stringify(wake)).toBe("coalesced");
|
|
const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, wake.runId!));
|
|
expect(successor.status).toBe("queued");
|
|
expect(successor.contextSnapshot?.wakeCommentIds).toEqual(seeded.commentIds);
|
|
await heartbeatService(db).resumeQueuedCommentInterrupt(seeded.companyId, seeded.wakeId);
|
|
expect(await db.select().from(heartbeatRuns)).toHaveLength(3);
|
|
},
|
|
);
|
|
|
|
it("does not accept interruption authority from an agent wake payload", async () => {
|
|
const seeded = await seedQueue();
|
|
await db.update(agents).set({ adapterType: "claude_local",
|
|
runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } },
|
|
}).where(eq(agents.id, seeded.agentId));
|
|
await db.update(heartbeatRuns).set({ runtimeMode: "legacy" }).where(eq(heartbeatRuns.id, seeded.runId));
|
|
await heartbeatService(db).wakeup(seeded.agentId, {
|
|
source: "on_demand", reason: "issue_commented",
|
|
requestedByActorType: "agent", requestedByActorId: seeded.agentId,
|
|
payload: { issueId: seeded.issueId, commentId: seeded.commentIds[1],
|
|
queuedCommentInterrupt: { actorId: "other-operator", requestedAt: new Date().toISOString() } },
|
|
contextSnapshot: { issueId: seeded.issueId, wakeCommentId: seeded.commentIds[1] },
|
|
});
|
|
const wakes = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, seeded.companyId));
|
|
expect(wakes.length).toBeGreaterThan(0);
|
|
expect(wakes.every(wake => !wake.payload?.queuedCommentInterrupt)).toBe(true);
|
|
const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, seeded.companyId));
|
|
expect(runs.find(run => run.id === seeded.runId)?.status).toBe("running");
|
|
expect(runs.every(run => !run.contextSnapshot?.explicitUserContinuation)).toBe(true);
|
|
});
|
|
|
|
it("denies a viewer's interrupt before persisting intent or cancelling a run", async () => {
|
|
const seeded = await seedQueue();
|
|
await db.update(companyMemberships).set({ membershipRole: "viewer" })
|
|
.where(eq(companyMemberships.principalId, "other-operator"));
|
|
const client = app(seeded.companyId, "other-operator");
|
|
const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
|
await request(client).post(`/api/issues/${seeded.issueId}/queued-comments/interrupt`).send({
|
|
queueId: seeded.wakeId, revision: queue.body.revision, targetRunId: seeded.runId,
|
|
}).expect(403);
|
|
const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
expect(wake.payload?.queuedCommentInterrupt).toBeUndefined();
|
|
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)))[0].status).toBe("running");
|
|
});
|
|
|
|
it.each([
|
|
{ target: null, runtime: "legacy" }, { target: "stopped-target", runtime: "legacy" },
|
|
{ target: "system-receipt", runtime: "legacy" }, { target: null, runtime: "native" },
|
|
] as const)("sends a stopped $runtime queue once with target $target", async ({ target, runtime }) => {
|
|
const seeded = await seedQueue();
|
|
if (target === "system-receipt") await db.update(agentWakeupRequests).set({
|
|
requestedByActorType: "system", requestedByActorId: "heartbeat",
|
|
}).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
await db.update(agents).set({ adapterType: runtime === "native" ? "paperclip_runner" : "claude_local",
|
|
runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } },
|
|
}).where(eq(agents.id, seeded.agentId));
|
|
await db.update(heartbeatRuns).set({ runtimeMode: runtime, status: "succeeded",
|
|
finishedAt: new Date("2026-08-22T15:03:00.000Z"),
|
|
}).where(eq(heartbeatRuns.id, seeded.runId));
|
|
if (runtime === "native") await commitNativeSuccess(seeded);
|
|
await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, seeded.issueId));
|
|
// Occupy this agent on a different task so the actual successor remains
|
|
// queued and the test never launches a provider.
|
|
await db.insert(heartbeatRuns).values({ companyId: seeded.companyId, agentId: seeded.agentId,
|
|
status: "running", contextSnapshot: { issueId: randomUUID() },
|
|
});
|
|
const client = app(seeded.companyId, "other-operator");
|
|
const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
|
expect(queue.body.targetRunId).toBeNull();
|
|
const body = { queueId: seeded.wakeId, revision: queue.body.revision,
|
|
targetRunId: target === "stopped-target" ? seeded.runId : null };
|
|
await request(client).post(`/api/issues/${seeded.issueId}/queued-comments/interrupt`).send(body).expect(200);
|
|
const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
expect(wake.status).toBe("coalesced");
|
|
const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, wake.runId!));
|
|
expect(successor.status).toBe("queued");
|
|
expect(successor.responsibleUserId).toBe("other-operator");
|
|
const identity = await initializeRunIdentity(db, {
|
|
companyId: seeded.companyId, issueId: seeded.issueId,
|
|
runId: successor.id, messageIds: seeded.commentIds, responsibleUserId: "queue-owner", cause: "dispatch",
|
|
});
|
|
expect(identity.responsibleUserId).toBe("other-operator");
|
|
expect(identity.cause).toBe("queued_comment_interrupt");
|
|
expect(successor.contextSnapshot?.wakeCommentIds).toEqual(seeded.commentIds);
|
|
await heartbeatService(db).resumeQueuedCommentInterrupt(seeded.companyId, seeded.wakeId);
|
|
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, seeded.companyId))).toHaveLength(3);
|
|
await request(client).post(`/api/issues/${seeded.issueId}/queued-comments/interrupt`).send(body).expect(409);
|
|
});
|
|
|
|
it.each([
|
|
["other", "running"], ["other", "queued"], ["other", "scheduled_retry"],
|
|
["same", "running"], ["same", "queued"], ["same", "scheduled_retry"],
|
|
] as const)("scopes interrupted queue successors to its agent: %s agent %s", async (owner, status) => {
|
|
const seeded = await seedQueue();
|
|
await db.update(agents).set({ adapterType: "claude_local",
|
|
runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } },
|
|
}).where(eq(agents.id, seeded.agentId));
|
|
await db.update(heartbeatRuns).set({ runtimeMode: "legacy", status: "succeeded",
|
|
finishedAt: new Date("2026-08-22T15:03:00.000Z"),
|
|
}).where(eq(heartbeatRuns.id, seeded.runId));
|
|
await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, seeded.issueId));
|
|
const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
await db.update(agentWakeupRequests).set({ payload: { ...wake.payload,
|
|
queuedCommentInterrupt: { actorId: "other-operator", requestedAt: new Date().toISOString() },
|
|
} }).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
// Keep this agent at capacity so successful delivery queues a successor
|
|
// without launching a provider. The independent run shares only the task.
|
|
await db.insert(heartbeatRuns).values({ companyId: seeded.companyId, agentId: seeded.agentId,
|
|
status: "running", contextSnapshot: { issueId: randomUUID() },
|
|
});
|
|
const successorAgentId = owner === "same" ? seeded.agentId : randomUUID();
|
|
if (owner === "other") await db.insert(agents).values({ id: successorAgentId,
|
|
companyId: seeded.companyId, name: "Independent agent", role: "engineer",
|
|
status: "idle", adapterType: "claude_local",
|
|
});
|
|
const [existingRun] = await db.insert(heartbeatRuns).values({ companyId: seeded.companyId, agentId: successorAgentId,
|
|
status, contextSnapshot: { issueId: seeded.issueId },
|
|
}).returning();
|
|
|
|
await heartbeatService(db).resumeQueuedCommentInterrupt(seeded.companyId, seeded.wakeId, { retryCleanup: true });
|
|
if (owner === "other" && status !== "scheduled_retry") {
|
|
// Another agent is not this queue's successor, but ordinary admission
|
|
// must still preserve the task execution lock until its work stops.
|
|
const [waiting] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
const [task] = await db.select().from(issues).where(eq(issues.id, seeded.issueId));
|
|
expect(waiting.status).toBe("deferred_issue_execution");
|
|
expect(task.executionRunId).toBe(existingRun.id);
|
|
expect(waiting.payload?.queuedCommentInterrupt).toMatchObject({ actorId: "other-operator" });
|
|
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() })
|
|
.where(eq(heartbeatRuns.id, existingRun.id));
|
|
await heartbeatService(db).resumeQueuedCommentInterrupt(seeded.companyId, seeded.wakeId, { retryCleanup: true });
|
|
}
|
|
const [receipt] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
expect(receipt.status).toBe(owner === "same" ? "deferred_issue_execution" : "coalesced");
|
|
if (owner === "other") {
|
|
const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, receipt.runId!));
|
|
expect(successor).toMatchObject({ agentId: seeded.agentId, status: "queued", responsibleUserId: "other-operator" });
|
|
expect(successor.contextSnapshot?.wakeCommentIds).toEqual(seeded.commentIds);
|
|
}
|
|
await heartbeatService(db).resumeQueuedCommentInterrupt(seeded.companyId, seeded.wakeId, { retryCleanup: true });
|
|
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, seeded.companyId)))
|
|
.toHaveLength(owner === "same" ? 3 : 4);
|
|
});
|
|
|
|
it.each(["user", "system"])("keeps stopped-run interruption intent on a %s receipt across restart until the process stops, then delivers once", async (actorType) => {
|
|
const seeded = await seedQueue();
|
|
await db.update(agentWakeupRequests).set({ requestedByActorType: actorType })
|
|
.where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
await db.update(agents).set({ adapterType: "claude_local",
|
|
runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } },
|
|
}).where(eq(agents.id, seeded.agentId));
|
|
await db.update(heartbeatRuns).set({ runtimeMode: "legacy", status: "failed",
|
|
processPid: process.pid, errorCode: "process_lost",
|
|
finishedAt: new Date("2026-08-22T15:03:00.000Z"),
|
|
}).where(eq(heartbeatRuns.id, seeded.runId));
|
|
await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, seeded.issueId));
|
|
await db.insert(issueRecoveryActions).values({ companyId: seeded.companyId, sourceIssueId: seeded.issueId,
|
|
kind: "active_run_watchdog", cause: "legacy_execution_requires_reconciliation", fingerprint: seeded.runId,
|
|
status: "resolved", outcome: "blocked", nextAction: "Automatic recovery stopped.",
|
|
evidence: { runId: seeded.runId, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } },
|
|
});
|
|
await db.insert(heartbeatRuns).values({ companyId: seeded.companyId, agentId: seeded.agentId,
|
|
status: "running", contextSnapshot: { issueId: randomUUID() },
|
|
});
|
|
const client = app(seeded.companyId, "other-operator");
|
|
const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
|
await request(client).post(`/api/issues/${seeded.issueId}/queued-comments/interrupt`).send({
|
|
queueId: seeded.wakeId, revision: queue.body.revision, targetRunId: null,
|
|
}).expect(200);
|
|
const [waiting] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
expect(waiting.status).toBe("deferred_issue_execution");
|
|
expect(waiting.payload?.queuedCommentInterrupt).toMatchObject({ actorId: "other-operator" });
|
|
expect(waiting.payload?.executionWait).toMatchObject({ reason: "process_running" });
|
|
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, seeded.companyId))).toHaveLength(2);
|
|
await db.update(heartbeatRuns).set({ processPid: 999999999 }).where(eq(heartbeatRuns.id, seeded.runId));
|
|
await db.update(agentWakeupRequests).set({ updatedAt: new Date(0) }).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
// New service instances have no memory of the HTTP request. Concurrent
|
|
// periodic workers must consume its durable receipt exactly once.
|
|
await Promise.all([heartbeatService(db).resumeQueuedRuns(), heartbeatService(db).resumeQueuedRuns()]);
|
|
const [delivered] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
expect(delivered.status).toBe("coalesced");
|
|
const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, delivered.runId!));
|
|
expect(successor.contextSnapshot).toMatchObject({ wakeCommentIds: seeded.commentIds,
|
|
previousRunId: seeded.runId, forceFreshSession: true });
|
|
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, seeded.companyId))).toHaveLength(3);
|
|
});
|
|
|
|
it.each(["user", "system", "manual_receipt_other_actor", "rejected_admission", "consumed_first", "consumed_last", "deleted_first", "agent_first", "cancelled_queued"])("delivers saved user messages on a %s queue after automatic recovery stopped, without another click", async (actorType) => {
|
|
const seeded = await seedQueue();
|
|
await db.update(agentWakeupRequests).set({ requestedByActorType: actorType === "user" ? "user" : "system" })
|
|
.where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
if (actorType === "manual_receipt_other_actor") {
|
|
const [saved] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
await db.update(agentWakeupRequests).set({ requestedByActorType: "user", requestedByActorId: "other-operator",
|
|
payload: { ...saved.payload, manualUserWake: true },
|
|
}).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
}
|
|
await db.update(agents).set({ adapterType: "claude_local",
|
|
runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } },
|
|
}).where(eq(agents.id, seeded.agentId));
|
|
await db.update(heartbeatRuns).set({ runtimeMode: "legacy", status: "failed",
|
|
processPid: process.pid, errorCode: "process_lost", finishedAt: new Date("2026-08-22T15:03:00.000Z"),
|
|
}).where(eq(heartbeatRuns.id, seeded.runId));
|
|
await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, seeded.issueId));
|
|
await db.insert(issueRecoveryActions).values({ companyId: seeded.companyId, sourceIssueId: seeded.issueId,
|
|
kind: "active_run_watchdog", cause: "legacy_execution_requires_reconciliation", fingerprint: seeded.runId,
|
|
status: "resolved", outcome: "blocked", nextAction: "Automatic recovery stopped.",
|
|
evidence: { runId: seeded.runId, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } },
|
|
});
|
|
// Leave the agent's only slot occupied on another task so dispatch stays queued.
|
|
await db.insert(heartbeatRuns).values({ companyId: seeded.companyId, agentId: seeded.agentId,
|
|
status: "running", contextSnapshot: { issueId: randomUUID() },
|
|
});
|
|
await heartbeatService(db).resumeQueuedRuns();
|
|
expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId)))[0].status)
|
|
.toBe("deferred_issue_execution");
|
|
await db.update(heartbeatRuns).set({ processPid: 999999999 }).where(eq(heartbeatRuns.id, seeded.runId));
|
|
const excludedIndex = actorType === "consumed_last" ? 1 : 0;
|
|
const filtersInput = ["consumed_first", "consumed_last", "deleted_first", "agent_first"].includes(actorType);
|
|
if (actorType.startsWith("consumed_")) await db.update(heartbeatRuns).set({
|
|
contextSnapshot: { issueId: seeded.issueId, wakeCommentIds: [seeded.commentIds[excludedIndex]] },
|
|
}).where(eq(heartbeatRuns.id, seeded.runId));
|
|
if (actorType === "deleted_first") await db.update(issueComments).set({ deletedAt: new Date() })
|
|
.where(eq(issueComments.id, seeded.commentIds[0]));
|
|
if (actorType === "agent_first") await db.update(issueComments).set({
|
|
authorType: "agent", authorUserId: null, authorAgentId: seeded.agentId,
|
|
}).where(eq(issueComments.id, seeded.commentIds[0]));
|
|
const expectedIds = seeded.commentIds.filter((_, index) => !filtersInput || index !== excludedIndex);
|
|
if (actorType === "cancelled_queued") await db.insert(heartbeatRuns).values({
|
|
companyId: seeded.companyId, agentId: seeded.agentId, status: "cancelled", runtimeMode: "legacy",
|
|
errorCode: "agent_paused", createdAt: new Date(0), finishedAt: new Date(1),
|
|
contextSnapshot: { issueId: seeded.issueId, wakeCommentIds: [seeded.commentIds[0]] },
|
|
});
|
|
if (actorType === "rejected_admission") {
|
|
await db.insert(heartbeatRuns).values({ companyId: seeded.companyId, agentId: seeded.agentId,
|
|
status: "cancelled", runtimeMode: "legacy", errorCode: "execution_reconciliation_required",
|
|
contextSnapshot: { issueId: seeded.issueId }, finishedAt: new Date(),
|
|
});
|
|
}
|
|
await Promise.all([heartbeatService(db).resumeQueuedRuns(), heartbeatService(db).resumeQueuedRuns()]);
|
|
const [delivered] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
expect(delivered.status).toBe("coalesced");
|
|
const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, delivered.runId!));
|
|
expect(successor.contextSnapshot).toMatchObject({ wakeCommentIds: expectedIds,
|
|
previousRunId: seeded.runId, forceFreshSession: true });
|
|
if (actorType === "manual_receipt_other_actor") {
|
|
const [dispatch] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, successor.wakeupRequestId!));
|
|
expect(dispatch.requestedByActorId).toBe("queue-owner");
|
|
expect(dispatch.payload?.manualUserWake).toBeUndefined();
|
|
expect(successor.responsibleUserId).toBe("queue-owner");
|
|
}
|
|
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, seeded.companyId))).toHaveLength(["rejected_admission", "cancelled_queued"].includes(actorType) ? 4 : 3);
|
|
const [recovery] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, seeded.issueId));
|
|
expect(recovery.evidence.automaticRecovery).toMatchObject({ actionOutcome: "unknown" });
|
|
});
|
|
|
|
it("recovers a message deferred after legacy finalization released the task lock", async () => {
|
|
const seeded = await seedQueue();
|
|
await db.update(agents).set({ adapterType: "claude_local",
|
|
runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } },
|
|
}).where(eq(agents.id, seeded.agentId));
|
|
await db.update(heartbeatRuns).set({ runtimeMode: "legacy", status: "succeeded",
|
|
finishedAt: new Date("2026-08-22T15:03:00.000Z"),
|
|
}).where(eq(heartbeatRuns.id, seeded.runId));
|
|
await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, seeded.issueId));
|
|
await db.insert(heartbeatRuns).values({ companyId: seeded.companyId, agentId: seeded.agentId,
|
|
status: "running", contextSnapshot: { issueId: randomUUID() },
|
|
});
|
|
await heartbeatService(db).resumeQueuedRuns();
|
|
const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
expect(wake.status).toBe("queued");
|
|
const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, wake.runId!));
|
|
expect(successor.contextSnapshot?.wakeCommentIds).toEqual(seeded.commentIds);
|
|
await heartbeatService(db).resumeQueuedRuns();
|
|
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, seeded.companyId))).toHaveLength(3);
|
|
});
|
|
|
|
async function promoteQueue(seeded: Awaited<ReturnType<typeof seedQueue>>) {
|
|
const queueRunId = randomUUID();
|
|
const wake = await db
|
|
.select({ payload: agentWakeupRequests.payload })
|
|
.from(agentWakeupRequests)
|
|
.where(eq(agentWakeupRequests.id, seeded.wakeId))
|
|
.then((rows) => rows[0]);
|
|
const wakeContext = (wake?.payload as any)?._paperclipWakeContext ?? {};
|
|
await db
|
|
.update(heartbeatRuns)
|
|
.set({ status: "succeeded", finishedAt: new Date("2026-08-22T15:05:00.000Z") })
|
|
.where(eq(heartbeatRuns.id, seeded.runId));
|
|
await db.insert(heartbeatRuns).values({
|
|
id: queueRunId,
|
|
companyId: seeded.companyId,
|
|
agentId: seeded.agentId,
|
|
invocationSource: "issue_comment",
|
|
triggerDetail: "queue route promotion test",
|
|
status: "queued",
|
|
runtimeMode: "native",
|
|
wakeupRequestId: seeded.wakeId,
|
|
contextSnapshot: {
|
|
issueId: seeded.issueId,
|
|
wakeReason: "issue_reopened_via_comment",
|
|
...wakeContext,
|
|
},
|
|
});
|
|
await db
|
|
.update(agentWakeupRequests)
|
|
.set({ status: "queued", runId: queueRunId })
|
|
.where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
await db
|
|
.update(issues)
|
|
.set({ executionRunId: queueRunId })
|
|
.where(eq(issues.id, seeded.issueId));
|
|
return queueRunId;
|
|
}
|
|
|
|
it("projects the recovery wait reason only while the message is deferred", async () => {
|
|
const seeded = await seedQueue();
|
|
const executionWait = { reason: "remote_cleanup", message: "Waiting for the previous environment to stop." };
|
|
await db.update(agentWakeupRequests).set({
|
|
payload: sql`coalesce(${agentWakeupRequests.payload}, '{}'::jsonb) || ${JSON.stringify({ executionWait })}::jsonb`,
|
|
}).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
const waiting = await request(app(seeded.companyId)).get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
expect(waiting.status).toBe(200);
|
|
expect(waiting.body.executionWait).toEqual(executionWait);
|
|
await promoteQueue(seeded);
|
|
const admitted = await request(app(seeded.companyId)).get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
expect(admitted.status).toBe(200);
|
|
expect(admitted.body.executionWait).toBeUndefined();
|
|
});
|
|
|
|
it("returns the authoritative order, preserves full Markdown edits, and rejects stale revisions", async () => {
|
|
const seeded = await seedQueue();
|
|
const initial = await request(app(seeded.companyId))
|
|
.get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
|
|
expect(initial.status, JSON.stringify(initial.body)).toBe(200);
|
|
expect(initial.body).toMatchObject({
|
|
issueId: seeded.issueId,
|
|
queueId: seeded.wakeId,
|
|
state: "deferred",
|
|
targetRunId: seeded.runId,
|
|
protocol: "paperclip_runner_v1",
|
|
entries: [
|
|
{ position: 0, canEdit: true, canDiscard: true, comment: { id: seeded.commentIds[0] } },
|
|
{ position: 1, canEdit: true, canDiscard: true, comment: { id: seeded.commentIds[1] } },
|
|
],
|
|
});
|
|
|
|
const markdown = " Keep **all** Markdown. \n";
|
|
const edited = await request(app(seeded.companyId))
|
|
.patch(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}`)
|
|
.send({ queueId: seeded.wakeId, revision: initial.body.revision, body: markdown });
|
|
expect(edited.status, JSON.stringify(edited.body)).toBe(200);
|
|
expect(edited.body.entries[0].comment.body).toBe(markdown);
|
|
expect(edited.body.revision).not.toBe(initial.body.revision);
|
|
const stored = await db.select({ body: issueComments.body })
|
|
.from(issueComments)
|
|
.where(eq(issueComments.id, seeded.commentIds[0]))
|
|
.then((rows) => rows[0]);
|
|
expect(stored?.body).toBe(markdown);
|
|
|
|
const stale = await request(app(seeded.companyId))
|
|
.patch(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}`)
|
|
.send({ queueId: seeded.wakeId, revision: initial.body.revision, body: "stale" });
|
|
expect(stale.status).toBe(409);
|
|
expect(stale.body.details?.code).toBe("queued_comment_revision_conflict");
|
|
});
|
|
|
|
it("writes one activity log row for each successful queue mutation, and none for a rejected one", async () => {
|
|
const seeded = await seedQueue();
|
|
const initial = await request(app(seeded.companyId))
|
|
.get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
|
|
const edited = await request(app(seeded.companyId))
|
|
.patch(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}`)
|
|
.send({ queueId: seeded.wakeId, revision: initial.body.revision, body: "edited body" });
|
|
expect(edited.status, JSON.stringify(edited.body)).toBe(200);
|
|
|
|
const staleEdit = await request(app(seeded.companyId))
|
|
.patch(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}`)
|
|
.send({ queueId: seeded.wakeId, revision: initial.body.revision, body: "stale" });
|
|
expect(staleEdit.status).toBe(409);
|
|
|
|
const editRows = await db
|
|
.select({ details: activityLog.details })
|
|
.from(activityLog)
|
|
.where(eq(activityLog.action, "issue.queued_comment_edited"));
|
|
expect(editRows).toHaveLength(1);
|
|
expect(editRows[0]?.details).toMatchObject({
|
|
commentId: seeded.commentIds[0],
|
|
queueId: seeded.wakeId,
|
|
revision: edited.body.revision,
|
|
});
|
|
|
|
const reordered = await request(app(seeded.companyId))
|
|
.put(`/api/issues/${seeded.issueId}/queued-comments/order`)
|
|
.send({
|
|
queueId: seeded.wakeId,
|
|
revision: edited.body.revision,
|
|
orderedCommentIds: [...seeded.commentIds].reverse(),
|
|
});
|
|
expect(reordered.status, JSON.stringify(reordered.body)).toBe(200);
|
|
const reorderRow = await db
|
|
.select({ details: activityLog.details })
|
|
.from(activityLog)
|
|
.where(eq(activityLog.action, "issue.queued_comments_reordered"))
|
|
.then((rows) => rows[0]);
|
|
expect(reorderRow?.details).toMatchObject({
|
|
queueId: seeded.wakeId,
|
|
revision: reordered.body.revision,
|
|
orderedCommentIds: [...seeded.commentIds].reverse(),
|
|
});
|
|
|
|
const discarded = await request(app(seeded.companyId))
|
|
.delete(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[1]}`)
|
|
.send({ queueId: seeded.wakeId, revision: reordered.body.revision });
|
|
expect(discarded.status, JSON.stringify(discarded.body)).toBe(200);
|
|
const discardRow = await db
|
|
.select({ details: activityLog.details })
|
|
.from(activityLog)
|
|
.where(eq(activityLog.action, "issue.queued_comment_discarded"))
|
|
.then((rows) => rows[0]);
|
|
expect(discardRow?.details).toMatchObject({
|
|
commentId: seeded.commentIds[1],
|
|
queueId: seeded.wakeId,
|
|
revision: discarded.body.revision,
|
|
cancelledRunId: null,
|
|
});
|
|
});
|
|
|
|
it("preserves reordered messages across promotion and cancels the queued run after final trash", async () => {
|
|
const seeded = await seedQueue();
|
|
const initial = await request(app(seeded.companyId))
|
|
.get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
const reordered = await request(app(seeded.companyId))
|
|
.put(`/api/issues/${seeded.issueId}/queued-comments/order`)
|
|
.send({
|
|
queueId: seeded.wakeId,
|
|
revision: initial.body.revision,
|
|
orderedCommentIds: [...seeded.commentIds].reverse(),
|
|
});
|
|
expect(reordered.status, JSON.stringify(reordered.body)).toBe(200);
|
|
expect(reordered.body.entries.map((entry: any) => entry.comment.id)).toEqual([...seeded.commentIds].reverse());
|
|
|
|
const queueRunId = await promoteQueue(seeded);
|
|
const promoted = await request(app(seeded.companyId))
|
|
.get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
expect(promoted.body).toMatchObject({
|
|
queueId: seeded.wakeId,
|
|
state: "queued",
|
|
targetRunId: null,
|
|
revision: reordered.body.revision,
|
|
});
|
|
|
|
const afterFirstTrash = await request(app(seeded.companyId))
|
|
.delete(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[1]}`)
|
|
.send({ queueId: seeded.wakeId, revision: promoted.body.revision });
|
|
expect(afterFirstTrash.status, JSON.stringify(afterFirstTrash.body)).toBe(200);
|
|
expect(afterFirstTrash.body.entries.map((entry: any) => entry.comment.id)).toEqual([seeded.commentIds[0]]);
|
|
const queuedRunAfterFirstTrash = await db
|
|
.select({ contextSnapshot: heartbeatRuns.contextSnapshot })
|
|
.from(heartbeatRuns)
|
|
.where(eq(heartbeatRuns.id, queueRunId))
|
|
.then((rows) => rows[0]);
|
|
expect((queuedRunAfterFirstTrash?.contextSnapshot as any)?.wakeCommentIds).toEqual([
|
|
seeded.commentIds[0],
|
|
]);
|
|
|
|
const emptied = await request(app(seeded.companyId))
|
|
.delete(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}`)
|
|
.send({ queueId: seeded.wakeId, revision: afterFirstTrash.body.revision });
|
|
expect(emptied.status, JSON.stringify(emptied.body)).toBe(200);
|
|
expect(emptied.body.entries).toEqual([]);
|
|
const wake = await db.select({ status: agentWakeupRequests.status })
|
|
.from(agentWakeupRequests)
|
|
.where(eq(agentWakeupRequests.id, seeded.wakeId))
|
|
.then((rows) => rows[0]);
|
|
expect(wake?.status).toBe("cancelled");
|
|
const [queueRun, storedIssue] = await Promise.all([
|
|
db.select({ status: heartbeatRuns.status })
|
|
.from(heartbeatRuns)
|
|
.where(eq(heartbeatRuns.id, queueRunId))
|
|
.then((rows) => rows[0]),
|
|
db.select({ executionRunId: issues.executionRunId })
|
|
.from(issues)
|
|
.where(eq(issues.id, seeded.issueId))
|
|
.then((rows) => rows[0]),
|
|
]);
|
|
expect(queueRun?.status).toBe("cancelled");
|
|
expect(storedIssue?.executionRunId).toBeNull();
|
|
|
|
// Two discards happen in this scenario (the first trash, then the final
|
|
// one that empties the queue), so match the row by its own commentId
|
|
// instead of assuming insertion order.
|
|
const discardRows = await db
|
|
.select({ details: activityLog.details })
|
|
.from(activityLog)
|
|
.where(eq(activityLog.action, "issue.queued_comment_discarded"));
|
|
const finalDiscardRow = discardRows.find(
|
|
(row) => (row.details as { commentId?: string } | null)?.commentId === seeded.commentIds[0],
|
|
);
|
|
expect(finalDiscardRow?.details).toMatchObject({
|
|
commentId: seeded.commentIds[0],
|
|
cancelledRunId: queueRunId,
|
|
});
|
|
});
|
|
|
|
it("keeps a mutation response's steering disposition in step with a fresh GET after promotion", async () => {
|
|
const seeded = await seedQueue();
|
|
await promoteQueue(seeded);
|
|
const initial = await request(app(seeded.companyId))
|
|
.get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
expect(initial.body.steeringDisposition).toBe("temporarily_unavailable");
|
|
|
|
const edited = await request(app(seeded.companyId))
|
|
.patch(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}`)
|
|
.send({ queueId: seeded.wakeId, revision: initial.body.revision, body: "edited during promotion" });
|
|
expect(edited.status, JSON.stringify(edited.body)).toBe(200);
|
|
const afterEdit = await request(app(seeded.companyId))
|
|
.get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
expect(edited.body.steeringDisposition).toBe(afterEdit.body.steeringDisposition);
|
|
|
|
const reordered = await request(app(seeded.companyId))
|
|
.put(`/api/issues/${seeded.issueId}/queued-comments/order`)
|
|
.send({
|
|
queueId: seeded.wakeId,
|
|
revision: edited.body.revision,
|
|
orderedCommentIds: [...seeded.commentIds].reverse(),
|
|
});
|
|
expect(reordered.status, JSON.stringify(reordered.body)).toBe(200);
|
|
const afterReorder = await request(app(seeded.companyId))
|
|
.get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
expect(reordered.body.steeringDisposition).toBe(afterReorder.body.steeringDisposition);
|
|
});
|
|
|
|
it("returns entry objects with the same keys as the GET endpoint", async () => {
|
|
const seeded = await seedQueue();
|
|
const initial = await request(app(seeded.companyId))
|
|
.get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
const edited = await request(app(seeded.companyId))
|
|
.patch(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}`)
|
|
.send({ queueId: seeded.wakeId, revision: initial.body.revision, body: "edited body" });
|
|
expect(edited.status, JSON.stringify(edited.body)).toBe(200);
|
|
expect(Object.keys(edited.body.entries[0]).sort()).toEqual(Object.keys(initial.body.entries[0]).sort());
|
|
});
|
|
|
|
it("cancels the deferred wake when the final message is discarded before promotion", async () => {
|
|
const seeded = await seedQueue();
|
|
await db.delete(issueComments).where(eq(issueComments.id, seeded.commentIds[1]));
|
|
await db
|
|
.update(agentWakeupRequests)
|
|
.set({
|
|
payload: {
|
|
issueId: seeded.issueId,
|
|
commentId: seeded.commentIds[0],
|
|
_paperclipWakeContext: {
|
|
commentId: seeded.commentIds[0],
|
|
wakeCommentId: seeded.commentIds[0],
|
|
wakeCommentIds: [seeded.commentIds[0]],
|
|
},
|
|
},
|
|
})
|
|
.where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
const initial = await request(app(seeded.companyId))
|
|
.get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
const discarded = await request(app(seeded.companyId))
|
|
.delete(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}`)
|
|
.send({ queueId: seeded.wakeId, revision: initial.body.revision });
|
|
expect(discarded.status, JSON.stringify(discarded.body)).toBe(200);
|
|
const [wake, runs] = await Promise.all([
|
|
db.select({ status: agentWakeupRequests.status })
|
|
.from(agentWakeupRequests)
|
|
.where(eq(agentWakeupRequests.id, seeded.wakeId))
|
|
.then((rows) => rows[0]),
|
|
db.select({ id: heartbeatRuns.id }).from(heartbeatRuns),
|
|
]);
|
|
expect(wake?.status).toBe("cancelled");
|
|
expect(runs.map((run) => run.id)).toEqual([seeded.runId]);
|
|
});
|
|
|
|
it("routes legacy queued-comment cancellation through the promoted queue", async () => {
|
|
const seeded = await seedQueue();
|
|
const queueRunId = await promoteQueue(seeded);
|
|
const cancelled = await request(app(seeded.companyId))
|
|
.delete(`/api/issues/${seeded.issueId}/comments/${seeded.commentIds[0]}?mode=cancel`);
|
|
expect(cancelled.status, JSON.stringify(cancelled.body)).toBe(200);
|
|
expect(cancelled.body.id).toBe(seeded.commentIds[0]);
|
|
const [wake, queueRun] = await Promise.all([
|
|
db.select({ payload: agentWakeupRequests.payload })
|
|
.from(agentWakeupRequests)
|
|
.where(eq(agentWakeupRequests.id, seeded.wakeId))
|
|
.then((rows) => rows[0]),
|
|
db.select({ contextSnapshot: heartbeatRuns.contextSnapshot })
|
|
.from(heartbeatRuns)
|
|
.where(eq(heartbeatRuns.id, queueRunId))
|
|
.then((rows) => rows[0]),
|
|
]);
|
|
expect((wake?.payload as any)?._paperclipWakeContext?.wakeCommentIds).toEqual([
|
|
seeded.commentIds[1],
|
|
]);
|
|
expect((queueRun?.contextSnapshot as any)?.wakeCommentIds).toEqual([
|
|
seeded.commentIds[1],
|
|
]);
|
|
});
|
|
|
|
it("reports an explicit conflict after queued-run dispatch has begun", async () => {
|
|
const seeded = await seedQueue();
|
|
const initial = await request(app(seeded.companyId))
|
|
.get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
const queueRunId = await promoteQueue(seeded);
|
|
await db
|
|
.update(agentWakeupRequests)
|
|
.set({ status: "claimed", claimedAt: new Date() })
|
|
.where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
await db
|
|
.update(heartbeatRuns)
|
|
.set({ status: "running", startedAt: new Date() })
|
|
.where(eq(heartbeatRuns.id, queueRunId));
|
|
|
|
const discard = await request(app(seeded.companyId))
|
|
.delete(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}`)
|
|
.send({ queueId: seeded.wakeId, revision: initial.body.revision });
|
|
expect(discard.status).toBe(409);
|
|
expect(discard.body.details?.code).toBe("queued_comment_already_dispatching");
|
|
const comment = await db
|
|
.select({ id: issueComments.id })
|
|
.from(issueComments)
|
|
.where(eq(issueComments.id, seeded.commentIds[0]))
|
|
.then((rows) => rows[0]);
|
|
expect(comment?.id).toBe(seeded.commentIds[0]);
|
|
});
|
|
|
|
it("limits edit and trash to the comment owner", async () => {
|
|
const seeded = await seedQueue();
|
|
const initial = await request(app(seeded.companyId, "other-operator"))
|
|
.get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
expect(initial.status).toBe(200);
|
|
expect(initial.body.entries[0]).toMatchObject({ canEdit: false, canDiscard: false });
|
|
|
|
const edit = await request(app(seeded.companyId, "other-operator"))
|
|
.patch(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}`)
|
|
.send({ queueId: seeded.wakeId, revision: initial.body.revision, body: "not mine" });
|
|
expect(edit.status).toBe(403);
|
|
const discard = await request(app(seeded.companyId, "other-operator"))
|
|
.delete(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}`)
|
|
.send({ queueId: seeded.wakeId, revision: initial.body.revision });
|
|
expect(discard.status).toBe(403);
|
|
});
|
|
|
|
it("leaves the selected row queued when no native steering session is attached", async () => {
|
|
const seeded = await seedQueue();
|
|
const initial = await request(app(seeded.companyId))
|
|
.get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
expect(initial.body.steeringDisposition).toBe("temporarily_unavailable");
|
|
|
|
const steered = await request(app(seeded.companyId))
|
|
.post(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}/steer`)
|
|
.send({ queueId: seeded.wakeId, targetRunId: seeded.runId, revision: initial.body.revision });
|
|
|
|
expect(steered.status).toBe(409);
|
|
expect(steered.body.details).toMatchObject({
|
|
code: "steering_temporarily_unavailable",
|
|
retryable: true,
|
|
});
|
|
const queueAfterFailure = await request(app(seeded.companyId))
|
|
.get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
expect(queueAfterFailure.body.entries.map((entry: any) => entry.comment.id)).toEqual(seeded.commentIds);
|
|
});
|
|
|
|
it("returns the persisted acknowledgement when the final steering response is retried", async () => {
|
|
const seeded = await seedQueue();
|
|
await db.delete(issueComments).where(eq(issueComments.id, seeded.commentIds[1]));
|
|
await db
|
|
.update(agentWakeupRequests)
|
|
.set({
|
|
status: "cancelled",
|
|
finishedAt: new Date("2026-08-22T15:04:00.000Z"),
|
|
payload: {
|
|
issueId: seeded.issueId,
|
|
commentId: seeded.commentIds[0],
|
|
_paperclipWakeContext: {
|
|
commentId: seeded.commentIds[0],
|
|
wakeCommentId: seeded.commentIds[0],
|
|
wakeCommentIds: [seeded.commentIds[0]],
|
|
},
|
|
},
|
|
})
|
|
.where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
await db
|
|
.update(heartbeatRuns)
|
|
.set({
|
|
resultJson: {
|
|
queuedSteeringAcknowledgements: {
|
|
[seeded.commentIds[0]]: {
|
|
status: "acknowledged",
|
|
queueId: seeded.wakeId,
|
|
turnId: "turn-acknowledged",
|
|
acknowledgedAt: "2026-08-22T15:04:00.000Z",
|
|
},
|
|
},
|
|
},
|
|
})
|
|
.where(eq(heartbeatRuns.id, seeded.runId));
|
|
|
|
const retried = await request(app(seeded.companyId))
|
|
.post(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}/steer`)
|
|
.send({
|
|
queueId: seeded.wakeId,
|
|
targetRunId: seeded.runId,
|
|
revision: "response-was-lost-before-the-client-stored-the-revision",
|
|
});
|
|
|
|
expect(retried.status, JSON.stringify(retried.body)).toBe(200);
|
|
expect(retried.body).toMatchObject({
|
|
issueId: seeded.issueId,
|
|
queueId: null,
|
|
state: null,
|
|
entries: [],
|
|
});
|
|
const activity = await db
|
|
.select({ details: activityLog.details })
|
|
.from(activityLog)
|
|
.where(eq(activityLog.action, "issue.queued_comment_steered"))
|
|
.then((rows) => rows[0]);
|
|
expect(activity?.details).toMatchObject({
|
|
commentId: seeded.commentIds[0],
|
|
targetRunId: seeded.runId,
|
|
turnId: "turn-acknowledged",
|
|
duplicate: true,
|
|
});
|
|
});
|
|
|
|
it("does not reuse an acknowledgement from a different queue", async () => {
|
|
const seeded = await seedQueue();
|
|
const initial = await request(app(seeded.companyId))
|
|
.get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
await db
|
|
.update(heartbeatRuns)
|
|
.set({
|
|
resultJson: {
|
|
queuedSteeringAcknowledgements: {
|
|
[seeded.commentIds[0]]: {
|
|
status: "acknowledged",
|
|
queueId: randomUUID(),
|
|
turnId: "turn-from-another-queue",
|
|
acknowledgedAt: "2026-08-22T15:04:00.000Z",
|
|
},
|
|
},
|
|
},
|
|
})
|
|
.where(eq(heartbeatRuns.id, seeded.runId));
|
|
|
|
const steered = await request(app(seeded.companyId))
|
|
.post(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}/steer`)
|
|
.send({
|
|
queueId: seeded.wakeId,
|
|
targetRunId: seeded.runId,
|
|
revision: initial.body.revision,
|
|
});
|
|
|
|
expect(steered.status).toBe(409);
|
|
expect(steered.body.details).toMatchObject({
|
|
code: "steering_temporarily_unavailable",
|
|
retryable: true,
|
|
});
|
|
const queueAfterFailure = await request(app(seeded.companyId))
|
|
.get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
expect(queueAfterFailure.body.entries.map((entry: any) => entry.comment.id))
|
|
.toEqual(seeded.commentIds);
|
|
});
|
|
|
|
it("keeps queue edits available during handoff but rejects stale same-turn steering", async () => {
|
|
const seeded = await seedQueue();
|
|
const initial = await request(app(seeded.companyId))
|
|
.get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
await db
|
|
.update(heartbeatRuns)
|
|
.set({ status: "succeeded", finishedAt: new Date("2026-08-22T15:05:00.000Z") })
|
|
.where(eq(heartbeatRuns.id, seeded.runId));
|
|
|
|
const edit = await request(app(seeded.companyId))
|
|
.patch(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}`)
|
|
.send({
|
|
queueId: seeded.wakeId,
|
|
revision: initial.body.revision,
|
|
body: "edited during handoff",
|
|
});
|
|
|
|
expect(edit.status, JSON.stringify(edit.body)).toBe(200);
|
|
const stored = await db
|
|
.select({ body: issueComments.body })
|
|
.from(issueComments)
|
|
.where(eq(issueComments.id, seeded.commentIds[0]))
|
|
.then((rows) => rows[0]);
|
|
expect(stored?.body).toBe("edited during handoff");
|
|
const steer = await request(app(seeded.companyId))
|
|
.post(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}/steer`)
|
|
.send({
|
|
queueId: seeded.wakeId,
|
|
targetRunId: seeded.runId,
|
|
revision: edit.body.revision,
|
|
});
|
|
expect(steer.status).toBe(409);
|
|
expect(steer.body.details?.code).toBe("queued_comment_stale_target");
|
|
const wake = await db
|
|
.select({ status: agentWakeupRequests.status, payload: agentWakeupRequests.payload })
|
|
.from(agentWakeupRequests)
|
|
.where(eq(agentWakeupRequests.id, seeded.wakeId))
|
|
.then((rows) => rows[0]);
|
|
expect(wake?.status).toBe("deferred_issue_execution");
|
|
expect((wake?.payload as any)?._paperclipWakeContext?.wakeCommentIds).toEqual(seeded.commentIds);
|
|
});
|
|
|
|
it("cancels a queued continuation whose comments disappeared before claim", async () => {
|
|
const seeded = await seedQueue();
|
|
const queueRunId = await promoteQueue(seeded);
|
|
await db.delete(issueComments).where(eq(issueComments.issueId, seeded.issueId));
|
|
const heartbeat = heartbeatService(db, { runtimeEnv: {} });
|
|
|
|
await heartbeat.resumeQueuedRuns();
|
|
|
|
const [queueRun, wake] = await Promise.all([
|
|
db.select({ status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode })
|
|
.from(heartbeatRuns)
|
|
.where(eq(heartbeatRuns.id, queueRunId))
|
|
.then((rows) => rows[0]),
|
|
db.select({ status: agentWakeupRequests.status })
|
|
.from(agentWakeupRequests)
|
|
.where(eq(agentWakeupRequests.id, seeded.wakeId))
|
|
.then((rows) => rows[0]),
|
|
]);
|
|
expect(queueRun).toMatchObject({
|
|
status: "cancelled",
|
|
errorCode: "queued_comment_discarded",
|
|
});
|
|
expect(wake?.status).toBe("cancelled");
|
|
});
|
|
|
|
it("keeps a persisted legacy queue on the legacy protocol after the agent changes adapters", async () => {
|
|
const seeded = await seedQueue();
|
|
const queueRunId = await promoteQueue(seeded);
|
|
await db
|
|
.update(heartbeatRuns)
|
|
.set({ runtimeMode: "legacy" })
|
|
.where(eq(heartbeatRuns.id, queueRunId));
|
|
|
|
const queued = await request(app(seeded.companyId))
|
|
.get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
|
|
expect(queued.status, JSON.stringify(queued.body)).toBe(200);
|
|
expect(queued.body).toMatchObject({
|
|
queueId: seeded.wakeId,
|
|
state: "queued",
|
|
targetRunId: null,
|
|
protocol: "legacy",
|
|
steeringDisposition: "unsupported",
|
|
});
|
|
});
|
|
|
|
it.each(["paperclip_runner", "codex_local"])("keeps Steer during native preparation even after agent settings change to %s", async adapterType => {
|
|
const seeded = await seedQueue();
|
|
const steeringCalls = steerNativeSessionMock.mock.calls.length;
|
|
await db.update(heartbeatRuns).set({ runtimeMode: "legacy", runtimeModeResolvedAt: null,
|
|
runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" } },
|
|
}).where(eq(heartbeatRuns.id, seeded.runId));
|
|
await db.update(agents).set({ adapterType }).where(eq(agents.id, seeded.agentId));
|
|
const client = app(seeded.companyId);
|
|
const initial = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
|
expect(initial.body).toMatchObject({ protocol: "paperclip_runner_v1", steeringDisposition: "temporarily_unavailable" });
|
|
const edited = await request(client).patch(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}`)
|
|
.send({ queueId: seeded.wakeId, revision: initial.body.revision, body: "Updated direction" }).expect(200);
|
|
expect(edited.body).toMatchObject({ protocol: "paperclip_runner_v1", steeringDisposition: "temporarily_unavailable" });
|
|
expect(steerNativeSessionMock.mock.calls).toHaveLength(steeringCalls);
|
|
});
|
|
|
|
it("serializes discard against queued-run claim", async () => {
|
|
const seeded = await seedQueue();
|
|
await db.delete(issueComments).where(eq(issueComments.id, seeded.commentIds[1]));
|
|
await db
|
|
.update(agentWakeupRequests)
|
|
.set({
|
|
payload: {
|
|
issueId: seeded.issueId,
|
|
commentId: seeded.commentIds[0],
|
|
_paperclipWakeContext: {
|
|
commentId: seeded.commentIds[0],
|
|
wakeCommentId: seeded.commentIds[0],
|
|
wakeCommentIds: [seeded.commentIds[0]],
|
|
},
|
|
},
|
|
})
|
|
.where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
const initial = await request(app(seeded.companyId))
|
|
.get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
const queueRunId = await promoteQueue(seeded);
|
|
await db
|
|
.update(agents)
|
|
.set({
|
|
adapterType: "process",
|
|
adapterConfig: {
|
|
command: process.execPath,
|
|
args: ["-e", "process.exit(0)"],
|
|
},
|
|
})
|
|
.where(eq(agents.id, seeded.agentId));
|
|
await db
|
|
.update(heartbeatRuns)
|
|
.set({ runtimeMode: "legacy" })
|
|
.where(eq(heartbeatRuns.id, queueRunId));
|
|
const heartbeat = heartbeatService(db, { runtimeEnv: {} });
|
|
|
|
const [discard] = await Promise.all([
|
|
request(app(seeded.companyId))
|
|
.delete(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}`)
|
|
.send({ queueId: seeded.wakeId, revision: initial.body.revision }),
|
|
heartbeat.resumeQueuedRuns(),
|
|
]);
|
|
|
|
const queueRun = await db
|
|
.select({ status: heartbeatRuns.status })
|
|
.from(heartbeatRuns)
|
|
.where(eq(heartbeatRuns.id, queueRunId))
|
|
.then((rows) => rows[0]);
|
|
if (discard.status === 200) {
|
|
expect(queueRun?.status).toBe("cancelled");
|
|
} else {
|
|
expect(discard.status, JSON.stringify(discard.body)).toBe(409);
|
|
expect(discard.body.details?.code).toBe("queued_comment_already_dispatching");
|
|
expect(queueRun?.status).not.toBe("queued");
|
|
}
|
|
await heartbeat.drainActiveRunExecutions();
|
|
}, 30_000);
|
|
|
|
async function seedDispatchIdentity(seeded: Awaited<ReturnType<typeof seedQueue>>) {
|
|
const [identity] = await db.insert(runIdentityContexts).values({
|
|
companyId: seeded.companyId,
|
|
runId: seeded.runId,
|
|
revision: 1,
|
|
cause: "dispatch",
|
|
correlationId: "dispatch",
|
|
status: "accepted",
|
|
acceptedAt: new Date("2026-08-22T15:00:00.000Z"),
|
|
}).returning();
|
|
await db.update(heartbeatRuns)
|
|
.set({ activeIdentityContextId: identity!.id })
|
|
.where(eq(heartbeatRuns.id, seeded.runId));
|
|
return identity!;
|
|
}
|
|
|
|
it("writes the identity acceptance, the wake payload, the run acknowledgement, and the activity row on a successful steering transaction", async () => {
|
|
const seeded = await seedQueue();
|
|
await seedDispatchIdentity(seeded);
|
|
steerNativeSessionMock.mockResolvedValueOnce({ turnId: "turn-1" });
|
|
|
|
const initial = await request(app(seeded.companyId))
|
|
.get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
const steered = await request(app(seeded.companyId))
|
|
.post(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}/steer`)
|
|
.send({ queueId: seeded.wakeId, targetRunId: seeded.runId, revision: initial.body.revision });
|
|
|
|
expect(steered.status, JSON.stringify(steered.body)).toBe(200);
|
|
|
|
const steeringIdentity = await db
|
|
.select()
|
|
.from(runIdentityContexts)
|
|
.where(eq(runIdentityContexts.messageId, seeded.commentIds[0]))
|
|
.then((rows) => rows[0]);
|
|
expect(steeringIdentity).toMatchObject({
|
|
status: "accepted",
|
|
responsibleUserId: "queue-owner",
|
|
cause: "steering",
|
|
});
|
|
|
|
const wake = await db
|
|
.select({ payload: agentWakeupRequests.payload })
|
|
.from(agentWakeupRequests)
|
|
.where(eq(agentWakeupRequests.id, seeded.wakeId))
|
|
.then((rows) => rows[0]);
|
|
expect((wake?.payload as any)?._paperclipWakeContext?.wakeCommentIds).toEqual([seeded.commentIds[1]]);
|
|
|
|
const run = await db
|
|
.select({ resultJson: heartbeatRuns.resultJson, activeIdentityContextId: heartbeatRuns.activeIdentityContextId })
|
|
.from(heartbeatRuns)
|
|
.where(eq(heartbeatRuns.id, seeded.runId))
|
|
.then((rows) => rows[0]);
|
|
expect((run?.resultJson as any)?.queuedSteeringAcknowledgements?.[seeded.commentIds[0]]).toMatchObject({
|
|
status: "acknowledged",
|
|
queueId: seeded.wakeId,
|
|
turnId: "turn-1",
|
|
});
|
|
expect(run?.activeIdentityContextId).toBe(steeringIdentity!.id);
|
|
|
|
const activity = await db
|
|
.select({ details: activityLog.details })
|
|
.from(activityLog)
|
|
.where(eq(activityLog.action, "issue.queued_comment_steered"))
|
|
.then((rows) => rows[0]);
|
|
expect(activity?.details).toMatchObject({
|
|
commentId: seeded.commentIds[0],
|
|
targetRunId: seeded.runId,
|
|
turnId: "turn-1",
|
|
duplicate: false,
|
|
});
|
|
});
|
|
|
|
it("allows the runner to persist its acknowledgement while steering, preserving concurrent run updates", async () => {
|
|
const seeded = await seedQueue();
|
|
await seedDispatchIdentity(seeded);
|
|
const originalProfile = { adapterDispatch: { adapterType: "paperclip_runner" },
|
|
nativeExecutionInput: { provider: { kind: "codex", model: "gpt-5.6-sol", reasoningEffort: "medium" } } };
|
|
await db.update(heartbeatRuns).set({ runnerProfileJson: originalProfile }).where(eq(heartbeatRuns.id, seeded.runId));
|
|
await db.update(issues).set({ assigneeAdapterOverrides: { model: "gpt-6-sol", reasoningEffort: "xhigh" } })
|
|
.where(eq(issues.id, seeded.issueId));
|
|
steerNativeSessionMock.mockImplementationOnce(async () => {
|
|
await db.transaction(async tx => {
|
|
await tx.execute(sql`set local lock_timeout = '1000ms'`);
|
|
await appendHeartbeatRunEvent(tx as unknown as typeof db, {
|
|
companyId: seeded.companyId, agentId: seeded.agentId, runId: seeded.runId,
|
|
eventType: "item.completed", payload: { kind: "steering_acknowledgement" },
|
|
});
|
|
await tx.update(heartbeatRuns).set({ resultJson: { concurrentProviderReceipt: "preserve" } })
|
|
.where(eq(heartbeatRuns.id, seeded.runId));
|
|
});
|
|
return { turnId: "turn-before-model-change" };
|
|
});
|
|
const initial = await request(app(seeded.companyId)).get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
const steered = await request(app(seeded.companyId))
|
|
.post(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}/steer`)
|
|
.send({ queueId: seeded.wakeId, targetRunId: seeded.runId, revision: initial.body.revision });
|
|
expect(steered.status, JSON.stringify(steered.body)).toBe(200);
|
|
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId));
|
|
expect(run.status).toBe("running");
|
|
expect(run.runnerProfileJson).toEqual(originalProfile);
|
|
expect(steerNativeSessionMock).toHaveBeenCalledWith(expect.objectContaining({ runId: seeded.runId,
|
|
message: "First queued message", correlationId: seeded.commentIds[0] }));
|
|
expect(run.resultJson).toMatchObject({ concurrentProviderReceipt: "preserve", queuedSteeringAcknowledgements: {
|
|
[seeded.commentIds[0]]: { status: "acknowledged", turnId: "turn-before-model-change" },
|
|
} });
|
|
expect(steered.body.entries.map((entry: any) => entry.comment.id)).toEqual([seeded.commentIds[1]]);
|
|
});
|
|
|
|
it.each(["cancelled", "failed", "succeeded"] as const)("preserves queued input when the run stops during acknowledgement (%s)", async status => {
|
|
const seeded = await seedQueue();
|
|
await seedDispatchIdentity(seeded);
|
|
steerNativeSessionMock.mockImplementationOnce(async () => {
|
|
await db.update(heartbeatRuns).set({ status, finishedAt: new Date(), resultJson: { terminalReceipt: "preserve" } })
|
|
.where(eq(heartbeatRuns.id, seeded.runId));
|
|
return { turnId: "turn-that-stopped" };
|
|
});
|
|
const initial = await request(app(seeded.companyId)).get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
const response = await request(app(seeded.companyId))
|
|
.post(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}/steer`)
|
|
.send({ queueId: seeded.wakeId, targetRunId: seeded.runId, revision: initial.body.revision });
|
|
expect(response.status).toBe(409);
|
|
expect(response.body.details).toMatchObject({ code: "queued_comment_stale_target" });
|
|
const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
expect(wake.status).toBe("deferred_issue_execution");
|
|
expect((wake.payload as any)._paperclipWakeContext.wakeCommentIds).toEqual(seeded.commentIds);
|
|
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId));
|
|
expect(run.resultJson).toEqual({ terminalReceipt: "preserve" });
|
|
const activity = await db.select().from(activityLog).where(eq(activityLog.action, "issue.queued_comment_steered"));
|
|
expect(activity).toHaveLength(0);
|
|
});
|
|
|
|
it("keeps the identity pending after a steering timeout, then reconciles it on a later acknowledgement", async () => {
|
|
const seeded = await seedQueue();
|
|
await seedDispatchIdentity(seeded);
|
|
steerNativeSessionMock.mockRejectedValueOnce(
|
|
new NativeSessionSteeringError("steering_timeout", "The provider did not acknowledge steering in time."),
|
|
);
|
|
|
|
const initial = await request(app(seeded.companyId))
|
|
.get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
const steered = await request(app(seeded.companyId))
|
|
.post(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}/steer`)
|
|
.send({ queueId: seeded.wakeId, targetRunId: seeded.runId, revision: initial.body.revision });
|
|
|
|
expect(steered.status).toBe(409);
|
|
expect(steered.body.details).toMatchObject({ code: "steering_timeout", retryable: true });
|
|
|
|
const pending = await db
|
|
.select()
|
|
.from(runIdentityContexts)
|
|
.where(eq(runIdentityContexts.messageId, seeded.commentIds[0]))
|
|
.then((rows) => rows[0]);
|
|
expect(pending?.status).toBe("pending");
|
|
|
|
await reconcileSteeredIdentity(db, pending!);
|
|
|
|
const reconciled = await db
|
|
.select({ status: runIdentityContexts.status })
|
|
.from(runIdentityContexts)
|
|
.where(eq(runIdentityContexts.id, pending!.id))
|
|
.then((rows) => rows[0]);
|
|
expect(reconciled?.status).toBe("accepted");
|
|
const run = await db
|
|
.select({ activeIdentityContextId: heartbeatRuns.activeIdentityContextId })
|
|
.from(heartbeatRuns)
|
|
.where(eq(heartbeatRuns.id, seeded.runId))
|
|
.then((rows) => rows[0]);
|
|
expect(run?.activeIdentityContextId).toBe(pending!.id);
|
|
});
|
|
|
|
it("sets the identity to rejected on a definite steering rejection", async () => {
|
|
const seeded = await seedQueue();
|
|
await seedDispatchIdentity(seeded);
|
|
steerNativeSessionMock.mockRejectedValueOnce(
|
|
new NativeSessionSteeringError("steering_rejected", "The provider rejected the steering message."),
|
|
);
|
|
|
|
const initial = await request(app(seeded.companyId))
|
|
.get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
const steered = await request(app(seeded.companyId))
|
|
.post(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}/steer`)
|
|
.send({ queueId: seeded.wakeId, targetRunId: seeded.runId, revision: initial.body.revision });
|
|
|
|
expect(steered.status).toBe(409);
|
|
expect(steered.body.details).toMatchObject({ code: "steering_rejected", retryable: true });
|
|
|
|
const rejected = await db
|
|
.select({ status: runIdentityContexts.status })
|
|
.from(runIdentityContexts)
|
|
.where(eq(runIdentityContexts.messageId, seeded.commentIds[0]))
|
|
.then((rows) => rows[0]);
|
|
expect(rejected?.status).toBe("rejected");
|
|
});
|
|
|
|
it("throws queued_comment_order_mismatch and changes no row for an invalid reorder set", async () => {
|
|
const seeded = await seedQueue();
|
|
const initial = await request(app(seeded.companyId))
|
|
.get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
|
|
const invalid = await request(app(seeded.companyId))
|
|
.put(`/api/issues/${seeded.issueId}/queued-comments/order`)
|
|
.send({
|
|
queueId: seeded.wakeId,
|
|
revision: initial.body.revision,
|
|
orderedCommentIds: [seeded.commentIds[0], randomUUID()],
|
|
});
|
|
|
|
expect(invalid.status, JSON.stringify(invalid.body)).toBe(409);
|
|
expect(invalid.body.details?.code).toBe("queued_comment_order_mismatch");
|
|
|
|
const wake = await db
|
|
.select({ payload: agentWakeupRequests.payload })
|
|
.from(agentWakeupRequests)
|
|
.where(eq(agentWakeupRequests.id, seeded.wakeId))
|
|
.then((rows) => rows[0]);
|
|
expect((wake?.payload as any)?._paperclipWakeContext?.wakeCommentIds).toEqual(seeded.commentIds);
|
|
const after = await request(app(seeded.companyId))
|
|
.get(`/api/issues/${seeded.issueId}/queued-comments`);
|
|
expect(after.body.entries.map((entry: any) => entry.comment.id)).toEqual(seeded.commentIds);
|
|
});
|
|
// BUG-08: a declined connection request (Slack) queued its read-only receipt
|
|
// ahead of the user's next message. The receipt showed raw JSON, could not be
|
|
// edited or discarded, and the next message never appeared in the queue.
|
|
async function seedDeclinedConnectionReceipt(runState: "running" | "legacy" | "stopped") {
|
|
const seeded = await seedQueue();
|
|
const interactionId = randomUUID();
|
|
await db.insert(issueThreadInteractions).values({
|
|
id: interactionId, companyId: seeded.companyId, issueId: seeded.issueId,
|
|
kind: "connection_intent", status: "rejected", createdByAgentId: seeded.agentId,
|
|
sourceRunId: seeded.runId, resolvedByUserId: "queue-owner", resolvedAt: new Date(),
|
|
continuationPolicy: "wake_assignee", requestedResolverPolicy: "human_only",
|
|
effectiveResolverPolicy: "human_only", resolverPolicyProvenance: "explicit",
|
|
title: "Connect Slack", summary: "The agent needs this connection to continue.",
|
|
payload: { version: 1, serviceSlug: "slack", serviceName: "Slack", phase: "requested",
|
|
requestingAgentId: seeded.agentId, requestingAgentName: "Paperclip Runner" },
|
|
result: { version: 1, outcome: "declined", reason: null },
|
|
});
|
|
// Same receipt shape as wakeConnectionIntentAfterResolution.
|
|
await db.update(agentWakeupRequests).set({ payload: { issueId: seeded.issueId,
|
|
mutation: "interaction", interactionId, interactionKind: "connection_intent", interactionStatus: "rejected",
|
|
_paperclipWakeContext: { issueId: seeded.issueId, interactionId, interactionKind: "connection_intent",
|
|
interactionStatus: "rejected", mutation: "interaction", refreshTools: true } },
|
|
}).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
if (runState === "legacy") {
|
|
// A busy legacy run defers each new message behind the issue execution.
|
|
await db.update(agents).set({ adapterType: "claude_local",
|
|
runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }).where(eq(agents.id, seeded.agentId));
|
|
await db.update(heartbeatRuns).set({ runtimeMode: "legacy" }).where(eq(heartbeatRuns.id, seeded.runId));
|
|
const child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { stdio: "ignore" });
|
|
testProcesses.set(seeded.runId, child);
|
|
runningProcesses.set(seeded.runId, { child, graceSec: 1, processGroupId: null });
|
|
}
|
|
if (runState === "stopped") {
|
|
await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date(), errorCode: "workspace_busy" })
|
|
.where(eq(heartbeatRuns.id, seeded.runId));
|
|
await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, seeded.issueId));
|
|
// Occupy the agent on another task so admission holds new work without
|
|
// invoking a provider (the staging agent was busy on ONB-2/ONB-3).
|
|
await db.update(agents).set({ runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } })
|
|
.where(eq(agents.id, seeded.agentId));
|
|
await db.insert(heartbeatRuns).values({ companyId: seeded.companyId, agentId: seeded.agentId,
|
|
status: "running", contextSnapshot: { issueId: randomUUID() } });
|
|
}
|
|
return { ...seeded, interactionId };
|
|
}
|
|
|
|
it("shows a declined connection receipt as a readable status, not its JSON", async () => {
|
|
const seeded = await seedDeclinedConnectionReceipt("running");
|
|
const queue = await request(app(seeded.companyId)).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
|
expect(queue.body.entries).toHaveLength(1);
|
|
const [entry] = queue.body.entries;
|
|
expect(entry.source).toMatchObject({ kind: "interaction", interactionKind: "connection_intent",
|
|
summary: "Declined: Slack connection" });
|
|
// The agent still receives the typed receipt in the steering body.
|
|
expect(entry.comment.body).toContain('"outcome": "declined"');
|
|
});
|
|
|
|
it.each(["running", "stopped"] as const)(
|
|
"keeps the next message visible and actionable after a declined connection (%s run)",
|
|
async (runState) => {
|
|
const seeded = await seedDeclinedConnectionReceipt(runState);
|
|
const client = app(seeded.companyId);
|
|
const sent = await request(client).post(`/api/issues/${seeded.issueId}/comments`)
|
|
.send({ body: "Skip Slack for now and continue" }).expect(201);
|
|
await vi.waitFor(async () => expect(await db.select().from(agentWakeupRequests)).toHaveLength(2));
|
|
const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
|
const next = queue.body.entries.find((entry: any) => entry.comment.id === sent.body.id);
|
|
expect(next, JSON.stringify(queue.body.entries.map((entry: any) => entry.comment.body.split("\n")[0])))
|
|
.toMatchObject({ canEdit: true, canDiscard: true });
|
|
expect(next.source).toBeUndefined();
|
|
expect(queue.body.queueId).not.toBe(seeded.wakeId);
|
|
// The message is the user's own queue entry, so its controls work.
|
|
await request(client).delete(`/api/issues/${seeded.issueId}/queued-comments/${sent.body.id}`)
|
|
.send({ queueId: queue.body.queueId, revision: queue.body.revision }).expect(200);
|
|
// Declining the connection still reaches the agent on its own receipt.
|
|
const [receipt] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
expect(receipt.status).toBe("deferred_issue_execution");
|
|
await new Promise(resolve => setTimeout(resolve, 250));
|
|
},
|
|
);
|
|
|
|
it.each(["legacy", "stopped"] as const)(
|
|
"queues every later message together after a declined connection (%s run)",
|
|
async (runState) => {
|
|
const seeded = await seedDeclinedConnectionReceipt(runState);
|
|
const client = app(seeded.companyId);
|
|
const first = await request(client).post(`/api/issues/${seeded.issueId}/comments`)
|
|
.send({ body: "Skip Slack for now" }).expect(201);
|
|
await vi.waitFor(async () => expect(await db.select().from(agentWakeupRequests)).toHaveLength(2));
|
|
const second = await request(client).post(`/api/issues/${seeded.issueId}/comments`)
|
|
.send({ body: "Use email updates instead" }).expect(201);
|
|
await vi.waitFor(async () => {
|
|
const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
|
expect(queue.body.entries.map((entry: any) => entry.comment.id)).toEqual([first.body.id, second.body.id]);
|
|
});
|
|
const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
|
expect(queue.body.entries.every((entry: any) => entry.canEdit && entry.canDiscard)).toBe(true);
|
|
// The receipt is never merged with user input; it keeps its own wake.
|
|
const [receipt] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
expect(receipt.status).toBe("deferred_issue_execution");
|
|
expect(receipt.payload?._paperclipWakeContext).not.toHaveProperty("wakeCommentIds");
|
|
await new Promise(resolve => setTimeout(resolve, 250));
|
|
},
|
|
);
|
|
|
|
// N-06: the composer offers "Send" for a deferred queue with no running
|
|
// turn. The issue's own run can still be waiting to start (queued) or to
|
|
// retry (scheduled_retry); that Send must be accepted and delivered next.
|
|
it.each([
|
|
{ runtime: "native", status: "queued" }, { runtime: "native", status: "scheduled_retry" },
|
|
{ runtime: "legacy", status: "queued" }, { runtime: "legacy", status: "scheduled_retry" },
|
|
] as const)("accepts Send while the previous $runtime run is $status and delivers it once that run stops", async ({ runtime, status }) => {
|
|
const seeded = await seedQueue();
|
|
await db.update(agents).set({ adapterType: runtime === "native" ? "paperclip_runner" : "codex_local",
|
|
runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } },
|
|
}).where(eq(agents.id, seeded.agentId));
|
|
await db.update(heartbeatRuns).set({ runtimeMode: runtime, status, startedAt: null })
|
|
.where(eq(heartbeatRuns.id, seeded.runId));
|
|
const client = app(seeded.companyId);
|
|
const queue = await request(client).get(`/api/issues/${seeded.issueId}/queued-comments`).expect(200);
|
|
// This is the state in which the UI renders the "Send" button.
|
|
expect(queue.body).toMatchObject({ state: "deferred", targetRunId: null, queueId: seeded.wakeId });
|
|
const sent = await request(client).post(`/api/issues/${seeded.issueId}/queued-comments/interrupt`).send({
|
|
queueId: seeded.wakeId, revision: queue.body.revision, targetRunId: null,
|
|
});
|
|
expect(sent.status, JSON.stringify(sent.body)).toBe(200);
|
|
expect(sent.body).toMatchObject({ state: "deferred", queueId: seeded.wakeId });
|
|
expect(sent.body.entries.map((entry: { comment: { id: string } }) => entry.comment.id)).toEqual(seeded.commentIds);
|
|
// The previous run is not cancelled: it has not started any work to stop.
|
|
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)))[0]!.status).toBe(status);
|
|
const [saved] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
expect(saved.status).toBe("deferred_issue_execution");
|
|
expect(saved.payload?.queuedCommentInterrupt).toMatchObject({ actorId: "queue-owner" });
|
|
|
|
// The previous run then finishes. Occupy the agent on another task so the
|
|
// successor stays queued and the test never launches a provider.
|
|
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() })
|
|
.where(eq(heartbeatRuns.id, seeded.runId));
|
|
if (runtime === "native") await commitNativeSuccess(seeded);
|
|
await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, seeded.issueId));
|
|
await db.insert(heartbeatRuns).values({ companyId: seeded.companyId, agentId: seeded.agentId,
|
|
status: "running", contextSnapshot: { issueId: randomUUID() },
|
|
});
|
|
await heartbeatService(db).resumeQueuedCommentInterrupt(seeded.companyId, seeded.wakeId);
|
|
const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, seeded.wakeId));
|
|
expect(wake.status).toBe("coalesced");
|
|
const [successor] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, wake.runId!));
|
|
expect(successor.status).toBe("queued");
|
|
expect(successor.contextSnapshot?.wakeCommentIds).toEqual(seeded.commentIds);
|
|
});
|
|
});
|