Files
PaperClipAI/server/src/__tests__/activity-service.test.ts
T
DottaandPaperclip 5c7a025032 fix(ui): clear consumed messages from the pending queue (#16317)
## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Task conversations keep follow-up messages pending while an agent
works.
> - The screen combines optimistic sends, queue snapshots, and run input
records.
> - A late send response can restore a local queue flag after a run
takes the message.
> - A stale queue snapshot can then keep the same message in pending.
> - This pull request matches the input record to its source queue and
clears stale pending flags.
> - The benefit is that consumed messages leave pending while unsent
messages stay visible.

## Linked Issues or Issue Description

**What happened?**

A follow-up can remain in the task composer's pending row after a run
takes it as input. Local queue state can outlive the send request. The
old projection cleared that state for steering, but not for ordinary
run-start input. A stale queue read can also retain the consumed entry.

**Expected behavior**

A message that a started run takes as input appears once in history and
leaves pending. Other pending messages remain available. A message
assigned to a run that has not started remains pending.

**Steps to reproduce**

1. Send a follow-up while a task has a live run.
2. Let a started run record that message as input before or after the
send response arrives.
3. Let a queue read still include the consumed message.
4. Observe the message remain pending on the base revision.

**Paperclip version or commit**

Reproduced against master at `33d24e48ad` with controlled request
ordering.

**Deployment mode**

Source checkout. Browser verification used the production task screen
with a simulated native runner and delayed API responses. It did not run
a live model or change a deployed instance.

Related: [PR
#15030](https://github.com/paperclipai/paperclip/pull/15030) addresses
acknowledgement matching in the runner console. This change addresses
the task-detail composer and does not modify that console.

## What Changed

- Clear local pending flags when a started run records the comment as
input.
- Use live-run input IDs while the run-history query catches up.
- Match stale queue entries by their primary or coalesced source wake
ID, even when the snapshot targets the previous run or has no target.
- Retain coalesced source queue IDs in run context and include receipt
IDs in compact task run reads.
- Keep unstarted run inputs and authoritative retries pending. Cover
send ordering, same-run and successor delivery, queued snapshots,
coalesced inputs, and retry consumption.
- Add the delivery-race checks to the steering feature map.

## Verification

- Before the fix, the first four request-order regressions failed with
the consumed message still in pending. Coverage now includes 16
combinations of receipt source, send-response order, and stale snapshot
kind.
- Focused UI checks: 5 files and 233 tests passed. These cover task
detail, queue projection, optimistic comments, queued-message controls,
and the task-chat adapter.
- `pnpm -r typecheck` passed.
- Server receipt checks: 5 files and 139 tests passed with
`PAPERCLIP_TEST_POSTGRES_TEMPLATE=0`. These exercise real database
reads, coalesced heartbeat input, and polling routes. The default
cached-template path timed out during setup on this host; the
fresh-cluster path passed without longer timeouts.
- Token gates and `node scripts/check-maps.mjs` passed.
- Browser: sent a follow-up with a delayed response in the production
task screen. The consumed message moved into history. A second unsent
follow-up stayed queued. Reload preserved both states. The API and
runner were simulated.
- `pnpm build` passed.
- The full local repository test run was started, then stopped to
isolate database setup timeouts. All 55 GitHub CI checks passed on
`4ec6820cf1f89880b801fd5f938d72f54591c5e2`, including the full test
matrix. One serialized server shard passed on its single rerun after a
15-second route-test timeout; the same 97-test route suite also passed
locally.
- Greptile reviewed the final head at 5/5. All review threads are
resolved.

## Risks

- Queue display depends on matching comment IDs, source queue IDs, and a
recorded start time. Rows without start evidence stay pending. Newer
queues remain visible until their own input receipt arrives.
- Run context gains additive receipt metadata. No schema migration is
required. Dispatch order, stored queue state, company access, and
steering authority stay unchanged.
- Browser evidence covers controlled response ordering. It does not
establish deployment or live-provider behavior.

## Model Used

- OpenAI Codex, GPT-6, with reasoning, tool use, and code execution. The
exact serving model ID and context-window limit are not exposed in this
session.

## 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
#` 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>
2026-10-14 16:11:29 -05:00

657 lines
20 KiB
TypeScript

import { randomUUID } from "node:crypto";
import { eq } from "drizzle-orm";
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
import {
activityLog,
agentWakeupRequests,
agents,
companies,
createDb,
documentRevisions,
documents,
heartbeatRuns,
issueComments,
issueDocuments,
issues,
} from "@paperclipai/db";
import { ISSUE_CONTINUATION_SUMMARY_DOCUMENT_KEY } from "@paperclipai/shared";
import {
getEmbeddedPostgresTestSupport,
startEmbeddedPostgresTestDatabase,
} from "./helpers/embedded-postgres.js";
import { activityService } from "../services/activity.ts";
import { issueReadSqlCondition } from "../services/authorization.ts";
import { issueService } from "../services/issues.ts";
import { documentService } from "../services/documents.ts";
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
type ActivityService = ReturnType<typeof activityService>;
type IssueRun = Awaited<ReturnType<ActivityService["runsForIssue"]>>[number];
if (!embeddedPostgresSupport.supported) {
console.warn(
`Skipping embedded Postgres activity service tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`,
);
}
async function waitForIssueRun(
service: ActivityService,
companyId: string,
issueId: string,
predicate: (run: IssueRun) => boolean,
) {
const deadline = Date.now() + 2_000;
let latestRuns: IssueRun[] = [];
while (Date.now() < deadline) {
latestRuns = await service.runsForIssue(companyId, issueId);
const run = latestRuns.find(predicate);
if (run) return { run, runs: latestRuns };
await new Promise((resolve) => setTimeout(resolve, 25));
}
throw new Error(`Timed out waiting for issue run. Latest run count: ${latestRuns.length}`);
}
describeEmbeddedPostgres("activity service", () => {
let db!: ReturnType<typeof createDb>;
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
beforeAll(async () => {
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-activity-service-");
db = createDb(tempDb.connectionString);
}, 20_000);
afterEach(async () => {
await db.delete(activityLog);
await db.delete(issueComments);
await db.delete(issueDocuments);
await db.delete(documentRevisions);
await db.delete(documents);
await db.delete(issues);
await db.delete(heartbeatRuns);
await db.delete(agentWakeupRequests);
await db.delete(agents);
await db.delete(companies);
});
afterAll(async () => {
await tempDb?.cleanup();
});
it.each(["standard", "planning", "ask"] as const)("LCA-05 persists explicit %s mode across wording edits and uses it in ledger backfill", async (workMode) => {
const companyId = randomUUID();
const agentId = randomUUID();
await db.insert(companies).values({ id: companyId, name: "Mode fixture", issuePrefix: `M${companyId.slice(0, 6)}` });
await db.insert(agents).values({ id: agentId, companyId, name: "Writer", role: "engineer", status: "idle", adapterType: "codex_local" });
const service = issueService(db);
const task = await service.create(companyId, { title: "Making a plan", description: "Create a report and research proposal.", status: "in_progress", assigneeAgentId: agentId, ...(workMode === "standard" ? {} : { workMode }) });
expect(task.workMode).toBe(workMode);
for (const prose of [
{ title: "Implement exporter", description: "Change the code now." },
{ title: "Making a plan", description: "Create a plan for the report exporter." },
]) {
await service.update(task.id, prose);
expect((await service.getById(task.id))?.workMode).toBe(workMode);
}
const runId = randomUUID();
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "succeeded", contextSnapshot: { issueId: task.id }, resultJson: { summary: "I will inspect the repository next." } });
const expected = workMode === "planning" ? "advanced" : "plan_only";
await waitForIssueRun(activityService(db), companyId, task.id, run => run.runId === runId && run.livenessState === expected);
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0].livenessState).toBe(expected);
if (workMode === "standard") {
await documentService(db).upsertIssueDocument({ issueId: task.id, key: "plan", title: "Requested plan", format: "markdown", body: "1. Inspect files.\n2. Implement the exporter.", createdByAgentId: agentId });
expect((await service.getById(task.id))?.workMode).toBe("standard");
expect(await documentService(db).getIssueDocumentByKey(task.id, "plan")).toBeTruthy();
}
const nextMode = workMode === "planning" ? "standard" : "planning";
await service.update(task.id, { workMode: nextMode });
expect((await service.getById(task.id))?.workMode).toBe(nextMode);
});
it("limits company activity lists", async () => {
const companyId = randomUUID();
await db.insert(companies).values({
id: companyId,
name: "Paperclip",
issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
requireBoardApprovalForNewAgents: false,
});
await db.insert(activityLog).values([
{
companyId,
actorType: "system",
actorId: "system",
action: "test.oldest",
entityType: "company",
entityId: companyId,
createdAt: new Date("2026-04-21T10:00:00.000Z"),
},
{
companyId,
actorType: "system",
actorId: "system",
action: "test.middle",
entityType: "company",
entityId: companyId,
createdAt: new Date("2026-04-21T11:00:00.000Z"),
},
{
companyId,
actorType: "system",
actorId: "system",
action: "test.newest",
entityType: "company",
entityId: companyId,
createdAt: new Date("2026-04-21T12:00:00.000Z"),
},
]);
const result = await activityService(db).list({ companyId, limit: 2 });
expect(result.map((event) => event.action)).toEqual(["test.newest", "test.middle"]);
});
it("filters company activity rows whose entity is a private issue", async () => {
const companyId = randomUUID();
const privateIssueId = randomUUID();
const openIssueId = randomUUID();
await db.insert(companies).values({
id: companyId,
name: "Private activity",
issuePrefix: `A${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
});
await db.insert(issues).values([
{
id: privateIssueId,
companyId,
title: "Confidential activity",
status: "todo",
priority: "medium",
visibility: "private",
privacyRootIssueId: privateIssueId,
responsibleUserId: "private-owner",
},
{
id: openIssueId,
companyId,
title: "Open activity",
status: "todo",
priority: "medium",
},
]);
await db.insert(activityLog).values([
{ companyId, actorType: "system", actorId: "system", action: "private.changed", entityType: "issue", entityId: privateIssueId },
{ companyId, actorType: "system", actorId: "system", action: "open.changed", entityType: "issue", entityId: openIssueId },
]);
const actor = {
type: "board" as const,
userId: "non-member",
companyIds: [companyId],
source: "session" as const,
isInstanceAdmin: false,
};
const rows = await activityService(db).list({
companyId,
readCondition: await issueReadSqlCondition(db, actor),
});
expect(rows.map((row) => row.action)).toContain("open.changed");
expect(rows.map((row) => row.action)).not.toContain("private.changed");
});
it.each([null, "native_provider_model_rejected", "adapter_failed"])("returns compact issue runs with bounded model rejection details: %s", async (errorCode) => {
const companyId = randomUUID();
const agentId = randomUUID();
const issueId = randomUUID();
const runId = randomUUID();
const wakeupRequestId = randomUUID();
await db.insert(companies).values({
id: companyId,
name: "Paperclip",
issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
requireBoardApprovalForNewAgents: false,
});
await db.insert(agents).values({
id: agentId,
companyId,
name: "CodexCoder",
role: "engineer",
status: "running",
adapterType: "codex_local",
adapterConfig: {},
runtimeConfig: {},
permissions: {},
});
await db.insert(issues).values({
id: issueId,
companyId,
title: "Summarize a completed run",
status: "done",
priority: "medium",
assigneeAgentId: agentId,
});
await db.insert(agentWakeupRequests).values({
id: wakeupRequestId, companyId, agentId, source: "issue", status: "claimed",
});
await db.insert(heartbeatRuns).values({
id: runId,
wakeupRequestId,
companyId,
agentId,
scopeKind: "issue",
issueId,
invocationSource: "assignment",
status: errorCode ? "failed" : "succeeded",
errorCode,
error: "provider-error".repeat(200),
contextSnapshot: { issueId, queuedCommentSourceWakeIds: ["coalesced-queue"] },
usageJson: {
inputTokens: 11,
output_tokens: 7,
cache_read_input_tokens: 3,
billingType: "metered",
costUsd: 0.42,
enormousBlob: "x".repeat(256_000),
},
resultJson: {
conversationReset: true,
configurationIncomplete: { reason: "workspace_base_ref_unresolved", requestedRef: "main", fetchError: "private diagnostic" },
billing_type: "metered",
total_cost_usd: 0.42,
stopReason: "timeout",
effectiveTimeoutSec: 30,
timeoutFired: true,
summary: "done",
nestedHuge: { payload: "y".repeat(256_000) },
},
livenessState: "advanced",
livenessReason: "Run produced concrete action evidence: 1 issue comment(s)",
continuationAttempt: 2,
lastUsefulActionAt: new Date("2026-04-18T19:59:00.000Z"),
nextAction: "Review the completed output.",
});
const runs = await activityService(db).runsForIssue(companyId, issueId);
expect(runs).toHaveLength(1);
expect(runs[0]).toMatchObject({
runId,
wakeupRequestId,
agentId,
queuedCommentSourceWakeIds: ["coalesced-queue"],
invocationSource: "assignment",
contextIssueId: issueId,
error: errorCode === "native_provider_model_rejected" ? "provider-error".repeat(200).slice(0, 2000) : null,
});
expect(runs[0]?.usageJson).toEqual({
inputTokens: 11,
input_tokens: 11,
outputTokens: 7,
output_tokens: 7,
cachedInputTokens: 3,
cached_input_tokens: 3,
cache_read_input_tokens: 3,
billingType: "metered",
billing_type: "metered",
costUsd: 0.42,
cost_usd: 0.42,
total_cost_usd: 0.42,
});
expect(runs[0]?.resultJson).toEqual({
conversationReset: true,
configurationIncomplete: { reason: "workspace_base_ref_unresolved" },
billingType: "metered",
billing_type: "metered",
costUsd: 0.42,
cost_usd: 0.42,
total_cost_usd: 0.42,
stopReason: "timeout",
effectiveTimeoutSec: 30,
timeoutFired: true,
});
expect(runs[0]).toMatchObject({
livenessState: "advanced",
livenessReason: "Run produced concrete action evidence: 1 issue comment(s)",
continuationAttempt: 2,
lastUsefulActionAt: new Date("2026-04-18T19:59:00.000Z"),
nextAction: "Review the completed output.",
});
expect(runs[0]).not.toHaveProperty("contextSnapshot");
});
it("backfills missing liveness for completed issue runs before returning the ledger", async () => {
const companyId = randomUUID();
const agentId = randomUUID();
const issueId = randomUUID();
const runId = randomUUID();
const completedAt = new Date("2026-04-18T20:04:00.000Z");
await db.insert(companies).values({
id: companyId,
name: "Paperclip",
issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
requireBoardApprovalForNewAgents: false,
});
await db.insert(agents).values({
id: agentId,
companyId,
name: "CodexCoder",
role: "engineer",
status: "idle",
adapterType: "codex_local",
adapterConfig: {},
runtimeConfig: {},
permissions: {},
});
await db.insert(issues).values({
id: issueId,
companyId,
title: "Fix run ledger",
description: "Make the run ledger answer whether a run advanced.",
status: "done",
priority: "medium",
assigneeAgentId: agentId,
completedAt,
});
await db.insert(heartbeatRuns).values({
id: runId,
companyId,
agentId,
scopeKind: "issue",
issueId,
invocationSource: "assignment",
status: "succeeded",
startedAt: new Date("2026-04-18T20:00:00.000Z"),
finishedAt: completedAt,
contextSnapshot: { issueId },
resultJson: {
summary: "Finished the implementation.",
},
livenessState: null,
livenessReason: null,
lastUsefulActionAt: null,
nextAction: null,
});
await db.insert(issueComments).values({
companyId,
issueId,
authorAgentId: agentId,
createdByRunId: runId,
body: "Done",
createdAt: completedAt,
});
const service = activityService(db);
const { run, runs } = await waitForIssueRun(
service,
companyId,
issueId,
(entry) => entry.runId === runId && entry.livenessState === "completed",
);
expect(runs).toHaveLength(1);
expect(run).toMatchObject({
runId,
livenessState: "completed",
livenessReason: "Issue is done",
continuationAttempt: 0,
lastUsefulActionAt: completedAt,
});
const [persisted] = await db.select().from(heartbeatRuns);
expect(persisted).toMatchObject({
id: runId,
livenessState: "completed",
livenessReason: "Issue is done",
continuationAttempt: 0,
lastUsefulActionAt: completedAt,
});
});
it("does not backfill document evidence from a different run", async () => {
const companyId = randomUUID();
const agentId = randomUUID();
const issueId = randomUUID();
const runId = randomUUID();
const otherRunId = randomUUID();
const documentId = randomUUID();
const revisionId = randomUUID();
const createdAt = new Date("2026-04-18T20:08:00.000Z");
await db.insert(companies).values({
id: companyId,
name: "Paperclip",
issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
requireBoardApprovalForNewAgents: false,
});
await db.insert(agents).values({
id: agentId,
companyId,
name: "CodexCoder",
role: "engineer",
status: "idle",
adapterType: "codex_local",
adapterConfig: {},
runtimeConfig: {},
permissions: {},
});
await db.insert(issues).values({
id: issueId,
companyId,
title: "Fix run ledger",
description: "Make the run ledger answer whether a run advanced.",
status: "in_progress",
priority: "medium",
assigneeAgentId: agentId,
});
await db.insert(heartbeatRuns).values([
{
id: runId,
companyId,
agentId,
scopeKind: "issue",
issueId,
invocationSource: "assignment",
status: "succeeded",
startedAt: new Date("2026-04-18T20:00:00.000Z"),
finishedAt: new Date("2026-04-18T20:02:00.000Z"),
contextSnapshot: { issueId },
resultJson: {
summary: "Next steps:\n- inspect files",
},
livenessState: null,
livenessReason: null,
},
{
id: otherRunId,
companyId,
agentId,
scopeKind: "issue",
issueId,
invocationSource: "assignment",
status: "succeeded",
startedAt: new Date("2026-04-18T20:05:00.000Z"),
finishedAt: createdAt,
contextSnapshot: { issueId },
resultJson: {
summary: "Updated the plan document.",
},
livenessState: "advanced",
livenessReason: "Run produced concrete action evidence: 1 document revision(s)",
},
]);
await db.insert(documents).values({
id: documentId,
companyId,
title: "Plan",
format: "markdown",
latestBody: "# Plan\n\n- Inspect files",
latestRevisionId: revisionId,
latestRevisionNumber: 1,
createdByAgentId: agentId,
updatedByAgentId: agentId,
createdAt,
updatedAt: createdAt,
});
await db.insert(documentRevisions).values({
id: revisionId,
companyId,
documentId,
revisionNumber: 1,
title: "Plan",
format: "markdown",
body: "# Plan\n\n- Inspect files",
createdByAgentId: agentId,
createdByRunId: otherRunId,
createdAt,
});
await db.insert(issueDocuments).values({
companyId,
issueId,
documentId,
key: "plan",
createdAt,
updatedAt: createdAt,
});
const service = activityService(db);
const { run: backfilledRun } = await waitForIssueRun(
service,
companyId,
issueId,
(entry) => entry.runId === runId && entry.livenessState === "plan_only",
);
expect(backfilledRun).toMatchObject({
runId,
livenessState: "plan_only",
livenessReason: "Run described runnable future work without concrete action evidence",
lastUsefulActionAt: null,
});
});
it("does not treat continuation summary revisions as concrete backfill evidence", async () => {
const companyId = randomUUID();
const agentId = randomUUID();
const issueId = randomUUID();
const runId = randomUUID();
const documentId = randomUUID();
const revisionId = randomUUID();
const createdAt = new Date("2026-04-18T20:12:00.000Z");
await db.insert(companies).values({
id: companyId,
name: "Paperclip",
issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
requireBoardApprovalForNewAgents: false,
});
await db.insert(agents).values({
id: agentId,
companyId,
name: "CodexCoder",
role: "engineer",
status: "idle",
adapterType: "codex_local",
adapterConfig: {},
runtimeConfig: {},
permissions: {},
});
await db.insert(issues).values({
id: issueId,
companyId,
title: "Fix run ledger",
description: "Make the run ledger answer whether a run advanced.",
status: "in_progress",
priority: "medium",
assigneeAgentId: agentId,
});
await db.insert(heartbeatRuns).values({
id: runId,
companyId,
agentId,
scopeKind: "issue",
issueId,
invocationSource: "assignment",
status: "succeeded",
startedAt: new Date("2026-04-18T20:10:00.000Z"),
finishedAt: createdAt,
contextSnapshot: { issueId },
resultJson: {
summary: "Next steps:\n- inspect files",
},
livenessState: null,
livenessReason: null,
});
await db.insert(documents).values({
id: documentId,
companyId,
title: "Continuation Summary",
format: "markdown",
latestBody: "# Continuation Summary",
latestRevisionId: revisionId,
latestRevisionNumber: 1,
createdByAgentId: agentId,
updatedByAgentId: agentId,
createdAt,
updatedAt: createdAt,
});
await db.insert(documentRevisions).values({
id: revisionId,
companyId,
documentId,
revisionNumber: 1,
title: "Continuation Summary",
format: "markdown",
body: "# Continuation Summary",
createdByAgentId: agentId,
createdByRunId: runId,
createdAt,
});
await db.insert(issueDocuments).values({
companyId,
issueId,
documentId,
key: ISSUE_CONTINUATION_SUMMARY_DOCUMENT_KEY,
createdAt,
updatedAt: createdAt,
});
const service = activityService(db);
const { run: backfilledRun } = await waitForIssueRun(
service,
companyId,
issueId,
(entry) => entry.runId === runId && entry.livenessState === "plan_only",
);
expect(backfilledRun).toMatchObject({
runId,
livenessState: "plan_only",
livenessReason: "Run described runnable future work without concrete action evidence",
lastUsefulActionAt: null,
});
});
});