mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-03 17:51:58 +01:00
fix(server): deliver chat completions after nested workspace saves (#18665)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Agent Chat returns completed worker tasks to the original conversation. > - Delivery must retain its wake receipt when execution gates refuse admission. > - A workspace save can contain a nested snapshot operation. > - The old checks lost the receipt and treated that saved snapshot as unfinished work. > - This pull request preserves the receipt and recognizes a successful owning save. > - The benefit is that completed work can reach the chat through the normal queue. ## Linked Issues or Issue Description Refs #18652. Related: #17206 covers failed finalize recovery. This repair handles a successful finalize with an enclosed snapshot and does not change failed-save recovery policy. **What happened?** A delegated worker finished and saved its result. The original chat received no update. The dependency gate compared operation start times. A nested snapshot started after its enclosing finalize and therefore left a permanent hold. The completion wake also lost its exact idempotency receipt when the queue merged a refused wake into a generic dependency wait. The outbox reported `Completion wake returned no run or durable receipt`. **Expected behavior** A successful finalize covers a completed operation enclosed by that save when company, workspace, issue, and run match. Completion wakes that disable coalescing keep their own durable receipt even when admission waits. Dependency, budget, Stop, and pause gates still apply. **Steps to reproduce** 1. Hand a task from a chat to a worker and make the chat wait on that task. 2. Finish a workspace finalize that starts before its nested snapshot and ends after the successful snapshot. 3. Mark the worker Done and deliver its completion outbox entry. 4. On the parent revision, the dependency stays blocked and the completion receipt is absent. **Paperclip version or commit** Reproduced against `53f480d1b3` with real disposable Postgres rows and the normal queue/outbox services. ## What Changed - Honor `allowRunCoalescing: false` when recording execution waits. - Accept an enclosed successful operation only when the successful finalize has the same company, workspace, issue, and run and covers its whole recorded interval. - Preserve the existing rule that a strictly later successful workspace finalize clears earlier work. - Add failure controls for foreign runs, unfinished snapshots, failed saves, snapshots that finish too late, and later operations. - Exercise the real outbox and queue together. Hold provider dispatch at the existing drain boundary and verify server-recorded Done facts reach the admitted turn. - Document receipt identity and workspace-save coverage. ## Verification - Before repair: the two causal regressions fail and five safety controls pass. The missing exact receipt and empty wakeable-dependent list match the observed failure. - After repair: those seven cases pass without assertion changes. - The full affected set passed 288 tests and failed one existing scheduling test. Its unchanged Node24 baseline also failed. A closed-phase probe showed the fake adapter finished successfully at 3.453 seconds, after the incidental 3-second poll. The test now joins the existing execution promise and retains its status and dependency assertions under the unchanged Vitest timeout. The synchronized full dependency suite passes all 13 cases. - `pnpm -r typecheck` and `pnpm build` pass on pinned Node24/pnpm9. `pnpm test:run` is running. The combined real outbox/queue case passes. Fresh Greptile is 5/5 on `d7cee24b7032c7fdf171bfd1246cd90d41db1cba`. Exact-head normal CI and required `ci / verify` and `ci / e2e` pass. - The original scheduling failures on Node26 and Node24, the unchanged-source baseline failure, and the closed-phase probe are retained in the validation record. No model, deadline, production assertion, or paid attempt was retried to change the failure grade. - Current-head CI attempt1 passed all product and workspace coverage but failed the unchanged serialized retry-worker case at `issue-scheduled-retry-routes.test.ts:606` (first batch not observed before the poll deadline;410 other shard tests passed). The exact unchanged case passes locally on Node24 with real Postgres. The one failed-jobs-only retry passed on the same head; the original grade is retained. - The full local command completed all1828 chat integration tests. Its general stage remains running and has retained failures in unchanged `workspace-runtime.test.ts` (30-second local port-conflict fixture) and `muse-control.test.ts` (current binding authentication fixture). These are not counted as passing local full-suite evidence. Their current-head hosted coverage passed. - The separate advisory Paperclip Review app is paused and has no code finding. Its in-progress state is preserved and is not counted as a pass; it is not a required repository status check. - The live Cloud completion journey remains failed and unqualified. It must pass at the deployed fixed revision before daily-plan admission. No QA oracle, model, budget, or run limit changes are included. ## Risks - The workspace coverage query must not clear a save from another owner or new work after finalization. The guard checks the full interval and exact owner identity, with negative controls. - Refused noncoalescible automatic wakes retain separate receipts. Replaceable automatic signals still merge. No new polling or provider retry path is added. - No migration or API change. Deployment and live acceptance are separate remaining gates. ## Model Used - OpenAI Codex, GPT-6, with reasoning, tool use, and code execution. The exact deployment variant and context-window size 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 #` / `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>
This commit is contained in:
1 parent
cd12432947
commit
b4a66027b7
6 files changed
+138
-11
No files matched your search
@@ -202,6 +202,14 @@ and review run. The server reads these records from saved review state; it does
|
||||
not depend on the parent session remembering a separate review session. These
|
||||
records are evidence and do not grant permission to resolve another review.
|
||||
|
||||
An automatic chat-completion wake retains its exact delivery receipt when
|
||||
admission waits on a dependency, budget, or pause. It does not merge into a
|
||||
replaceable dependency signal. The same gates still prevent execution.
|
||||
Workspace readiness accepts a completed operation enclosed by a successful
|
||||
finalize from the same company, workspace, issue, and run. A nested snapshot
|
||||
that finishes before that finalize finishes does not leave a permanent wait.
|
||||
Unfinished operations and work after the finalize still require a later save.
|
||||
|
||||
### Answered Slack conversations
|
||||
|
||||
An accepted scheduled-monitor yield remains unfinished task work. Its committed
|
||||
|
||||
@@ -4,7 +4,7 @@ import { randomUUID } from "node:crypto";
|
||||
import { and, asc, eq } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { agents, agentWakeupRequests, chatCompletionDeliveries as deliveries, chatTaskHandoffs as handoffs,
|
||||
companies, createDb, documents, environmentLeases, heartbeatRuns, issueComments, issueRecoveryActions, issueThreadInteractions, issues } from "@paperclipai/db";
|
||||
companies, createDb, documents, environmentLeases, executionWorkspaces, heartbeatRuns, issueComments, issueRecoveryActions, issueThreadInteractions, issues, workspaceOperations } from "@paperclipai/db";
|
||||
import { recoveryPostgresSupport as support, startRecoveryPostgres } from "./helpers/recovery-postgres.js";
|
||||
import { settleInterruptedNativeBootstrap, terminalizeLegacyExecution } from "../services/legacy-execution-recovery.js";
|
||||
import { getExecutionBlocker } from "../services/execution-blocker.js";
|
||||
@@ -15,6 +15,8 @@ import { documentService } from "../services/documents.js";
|
||||
import { instanceSettingsService } from "../services/instance-settings.js";
|
||||
import { chatCompletionDeliveryService, isCompletedOnboardingHandoffWake, prepareChatCompletionTurn, recordChatCompletion, recordChatHandoff } from "../services/chat-completion-delivery.js";
|
||||
import { heartbeatService, shouldQueueFollowupForRunningIssueWake } from "../services/heartbeat.js";
|
||||
import { createHeartbeatQueue } from "../services/heartbeat/queue.js";
|
||||
import { createHeartbeatQueueTestDependencies } from "./helpers/heartbeat-queue.js";
|
||||
|
||||
(support.supported ? describe : describe.skip)("chat completion delivery", () => {
|
||||
let db: ReturnType<typeof createDb>;
|
||||
@@ -68,6 +70,75 @@ import { heartbeatService, shouldQueueFollowupForRunningIssueWake } from "../ser
|
||||
})]);
|
||||
expect(JSON.stringify(run.contextSnapshot?.chatCompletionUpdates)).not.toMatch(/Injected|credentials|Plan accepted|Agent approved/);
|
||||
});
|
||||
it("retains a distinct completion receipt behind a real dependency gate", async () => {
|
||||
const f = await seed();
|
||||
const blocker = await f.create();
|
||||
await issueService(db).update(f.sourceId, { status: "blocked", blockedByIssueIds: [blocker.id] });
|
||||
await f.finish();
|
||||
const heartbeat = heartbeatService(db);
|
||||
// The generic dependency signal is replaceable. The outbox's exact wake
|
||||
// receipt is not: it needs that identity to reconcile a refused admission.
|
||||
await heartbeat.wakeup(f.agentId, {
|
||||
source: "automation", triggerDetail: "system", reason: "issue_continuation_needed",
|
||||
requestedByActorType: "system", requestedByActorId: "dependency-test",
|
||||
payload: { issueId: f.sourceId }, contextSnapshot: { issueId: f.sourceId },
|
||||
});
|
||||
const [delivery] = await f.rows();
|
||||
const service = chatCompletionDeliveryService(db, heartbeat);
|
||||
await service.deliver(delivery.id);
|
||||
const receipts = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId));
|
||||
expect(receipts).toEqual(expect.arrayContaining([
|
||||
expect.objectContaining({ idempotencyKey: `chat-completion:${delivery.id}:0`,
|
||||
reason: "issue_dependencies_blocked", status: "skipped", runId: null,
|
||||
payload: expect.objectContaining({ chatCompletionDeliveryIds: [delivery.id] }) }),
|
||||
]));
|
||||
expect(receipts).toHaveLength(2);
|
||||
expect(receipts.find(receipt => receipt.idempotencyKey?.startsWith("execution-wait:")))
|
||||
.toMatchObject({ coalescedCount: 0 });
|
||||
expect((await f.rows())[0]).toMatchObject({ attempts: 0, status: "pending", targetRunId: null, error: null });
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(1);
|
||||
expect(await issueService(db).getDependencyReadiness(f.sourceId)).toMatchObject({
|
||||
isDependencyReady: false, unresolvedBlockerIssueIds: [blocker.id],
|
||||
});
|
||||
});
|
||||
it("admits a saved worker completion after its owning finalize encloses the snapshot", async () => {
|
||||
const f = await seed(), workspaceId = randomUUID(), workerRunId = randomUUID();
|
||||
await db.update(issues).set({ responsibleUserId: "operator" }).where(eq(issues.id, f.sourceId));
|
||||
await db.insert(executionWorkspaces).values({ id: workspaceId, companyId: f.companyId,
|
||||
mode: "isolated_workspace", strategyType: "managed_remote", name: "Saved worker", providerType: "daytona" });
|
||||
await db.insert(heartbeatRuns).values({ id: workerRunId, companyId: f.companyId, agentId: f.agentId,
|
||||
status: "succeeded", contextSnapshot: { issueId: f.task.id } });
|
||||
await db.update(issues).set({ executionWorkspaceId: workspaceId }).where(eq(issues.id, f.task.id));
|
||||
await issueService(db).update(f.sourceId, { status: "blocked", blockedByIssueIds: [f.task.id] });
|
||||
await db.insert(workspaceOperations).values([
|
||||
{ companyId: f.companyId, issueId: f.task.id, executionWorkspaceId: workspaceId, heartbeatRunId: workerRunId,
|
||||
phase: "workspace_finalize", status: "succeeded", startedAt: new Date(10_000), finishedAt: new Date(12_000) },
|
||||
{ companyId: f.companyId, issueId: f.task.id, executionWorkspaceId: workspaceId, heartbeatRunId: workerRunId,
|
||||
phase: "remote_workspace_snapshot", status: "succeeded", startedAt: new Date(10_100), finishedAt: new Date(11_900) },
|
||||
]);
|
||||
await f.finish();
|
||||
const deps = createHeartbeatQueueTestDependencies(db), issueSvc = issueService(db);
|
||||
deps.getAgent.mockImplementation(id => db.select().from(agents).where(eq(agents.id, id)).then(rows => rows[0] ?? null));
|
||||
deps.instanceSettings.getExperimental.mockImplementation(() => instanceSettingsService(db).getExperimental());
|
||||
deps.issuesSvc.getDependencyReadiness.mockImplementation(issueSvc.getDependencyReadiness);
|
||||
deps.issuesSvc.listDependencyReadiness.mockImplementation(issueSvc.listDependencyReadiness);
|
||||
// Admit the real durable run, then hold provider dispatch at the normal drain boundary.
|
||||
deps.getSchedulingSuppression.mockResolvedValue({ suppressed: true, reason: "task_drain" });
|
||||
const queue = createHeartbeatQueue(db, deps);
|
||||
const service = chatCompletionDeliveryService(db, { wakeup: queue.trackWakeup });
|
||||
const [delivery] = await f.rows();
|
||||
await service.deliver(delivery.id);
|
||||
const [queued] = await f.rows();
|
||||
expect(queued).toMatchObject({ status: "queued", attempts: 0, error: null, targetRunId: expect.any(String) });
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, queued.targetRunId!));
|
||||
expect(run).toMatchObject({ status: "queued", contextSnapshot: { issueId: f.sourceId,
|
||||
wakeReason: "chat_task_completed", chatCompletionDeliveryIds: [delivery.id] } });
|
||||
const prepared = await prepareChatCompletionTurn(db, run);
|
||||
expect(prepared.contextSnapshot?.chatCompletionUpdates).toEqual([
|
||||
expect.objectContaining({ id: f.task.id, status: "done", completedAt: expect.any(Date), url: `/issues/${f.task.identifier ?? f.task.id}` }),
|
||||
]);
|
||||
expect(deps.executeRun).not.toHaveBeenCalled();
|
||||
});
|
||||
it("dispatches a committed completion without activity publication and recovers after restart", async () => {
|
||||
const f = await seed();
|
||||
const run = vi.fn(() => f.service.sweepPending());
|
||||
|
||||
@@ -689,14 +689,9 @@ describeEmbeddedPostgres("heartbeat dependency-aware queued run selection", () =
|
||||
});
|
||||
expect(interactionWake).not.toBeNull();
|
||||
|
||||
await waitForCondition(async () => {
|
||||
const run = await db
|
||||
.select({ status: heartbeatRuns.status })
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, interactionWake!.id))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
return run?.status === "succeeded";
|
||||
});
|
||||
// This checks dependency interaction semantics, not provider startup speed.
|
||||
// Join the admitted execution before inspecting its persisted outcome.
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
|
||||
const interactionRun = await db
|
||||
.select({
|
||||
|
||||
@@ -4166,6 +4166,7 @@ const dependencyTestDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_ISSU
|
||||
await db.delete(executionWorkspaces);
|
||||
await db.delete(projectWorkspaces);
|
||||
await db.delete(projects);
|
||||
await db.delete(heartbeatRuns);
|
||||
await db.delete(agents);
|
||||
await db.delete(instanceSettings);
|
||||
await db.delete(companies);
|
||||
@@ -4720,6 +4721,42 @@ const dependencyTestDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_ISSU
|
||||
});
|
||||
});
|
||||
|
||||
it.each(["covered", "foreign-run", "unfinished", "failed-finalize", "late-finish", "later-operation"])(
|
||||
"requires a successful owning finalize to cover a nested snapshot: %s", async kind => {
|
||||
const f = await seedSharedWorkspaceDependency();
|
||||
const runId = randomUUID(), foreignRunId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values([runId, foreignRunId].map(id => ({
|
||||
id, companyId: f.companyId, agentId: f.assigneeAgentId, status: "succeeded",
|
||||
})));
|
||||
const started = new Date("2026-05-23T22:00:00.000Z");
|
||||
const finished = new Date("2026-05-23T22:00:02.000Z");
|
||||
await db.insert(workspaceOperations).values([
|
||||
{ companyId: f.companyId, executionWorkspaceId: f.executionWorkspaceId, issueId: f.blockerId,
|
||||
heartbeatRunId: runId, phase: "workspace_finalize",
|
||||
status: kind === "failed-finalize" ? "failed" : "succeeded", startedAt: started, finishedAt: finished },
|
||||
{ companyId: f.companyId, executionWorkspaceId: f.executionWorkspaceId, issueId: f.blockerId,
|
||||
heartbeatRunId: kind === "foreign-run" ? foreignRunId : runId, phase: "remote_workspace_snapshot",
|
||||
status: kind === "unfinished" ? "running" : "succeeded",
|
||||
startedAt: new Date(started.getTime() + 100),
|
||||
finishedAt: kind === "unfinished" ? null : new Date(finished.getTime() + (kind === "late-finish" ? 100 : -100)) },
|
||||
...(kind === "later-operation" ? [{
|
||||
companyId: f.companyId, executionWorkspaceId: f.executionWorkspaceId, issueId: f.blockerId,
|
||||
heartbeatRunId: runId, phase: "worktree_prepare", status: "succeeded",
|
||||
startedAt: new Date(finished.getTime() + 100), finishedAt: new Date(finished.getTime() + 200),
|
||||
}] : []),
|
||||
]);
|
||||
const ready = kind === "covered";
|
||||
expect((await svc.listWakeableBlockedDependents(f.blockerId)).map(issue => issue.id))
|
||||
.toEqual(ready ? [f.dependentId] : []);
|
||||
await expect(svc.getDependencyReadiness(f.dependentId)).resolves.toMatchObject({
|
||||
isDependencyReady: ready,
|
||||
pendingFinalizeBlockerIssueIds: ready ? [] : [f.blockerId],
|
||||
});
|
||||
const listed = (await svc.list(f.companyId, { status: "blocked" })).find(issue => issue.id === f.dependentId);
|
||||
expect(listed?.blockerAttention?.pendingFinalizeBlockerIssueIds ?? []).toEqual(ready ? [] : [f.blockerId]);
|
||||
},
|
||||
);
|
||||
|
||||
it("reports dependency readiness for blocked issue chains", async () => {
|
||||
const companyId = randomUUID();
|
||||
await seedIssueCompany(db, companyId);
|
||||
|
||||
@@ -2518,6 +2518,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
|
||||
// retain distinct durable receipts even when the same gate blocks them.
|
||||
const coalesceExecutionWait =
|
||||
opts.requestedByActorType === "system" &&
|
||||
opts.allowRunCoalescing !== false &&
|
||||
!receiptRequest &&
|
||||
!wakeCommentId &&
|
||||
queuedCommentIdsFromRunContext(enrichedContextSnapshot).length === 0 &&
|
||||
|
||||
@@ -2562,12 +2562,13 @@ function blockerWorkspaceFacts(companyId: string, workspaceScope: SQL) {
|
||||
}
|
||||
|
||||
/** An issue's own operation takes precedence over the legacy NULL fallback.
|
||||
* Only a strictly later successful finalize clears either barrier. */
|
||||
* A later successful finalize clears either barrier. A successful operation
|
||||
* enclosed by its owning finalize is also saved, although it started later. */
|
||||
function pendingFinalizeBlockerCondition(companyId: string, issueId: SQL, workspaceId: SQL) {
|
||||
return sql`${workspaceId} is not null and exists (
|
||||
select 1 from (values (1)) seed(value)
|
||||
left join lateral (
|
||||
select phase, status, started_at from workspace_operations blocker_operation
|
||||
select phase, status, started_at, finished_at, heartbeat_run_id from workspace_operations blocker_operation
|
||||
where blocker_operation.company_id = ${companyId}::uuid
|
||||
and blocker_operation.execution_workspace_id = ${workspaceId}
|
||||
and blocker_operation.issue_id = ${issueId}
|
||||
@@ -2578,6 +2579,20 @@ function pendingFinalizeBlockerCondition(companyId: string, issueId: SQL, worksp
|
||||
and coalesce(latest_blocker_operation.status, blocker_workspace_facts.status) = 'succeeded')
|
||||
and (blocker_workspace_facts.finalized_at is null
|
||||
or blocker_workspace_facts.finalized_at <= coalesce(latest_blocker_operation.started_at, blocker_workspace_facts.started_at))
|
||||
and not (
|
||||
latest_blocker_operation.status = 'succeeded'
|
||||
and latest_blocker_operation.finished_at is not null
|
||||
and exists (
|
||||
select 1 from workspace_operations owning_finalize
|
||||
where owning_finalize.company_id = ${companyId}::uuid
|
||||
and owning_finalize.execution_workspace_id = ${workspaceId}
|
||||
and owning_finalize.issue_id = ${issueId}
|
||||
and owning_finalize.heartbeat_run_id = latest_blocker_operation.heartbeat_run_id
|
||||
and owning_finalize.phase = 'workspace_finalize' and owning_finalize.status = 'succeeded'
|
||||
and owning_finalize.started_at <= latest_blocker_operation.started_at
|
||||
and owning_finalize.finished_at >= latest_blocker_operation.finished_at
|
||||
)
|
||||
)
|
||||
)`;
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user