Files
PaperClipAI/server/src/__tests__/slack-conversation-lifecycle.test.ts
T
DottaandPaperclip f211d44c00 fix(heartbeat): preserve queued coalesced comments at claim (#18913)
## Thinking Path

> - Paperclip manages agents and their work.
> - The wake queue records task comments before an agent starts.
> - A queued run can accept another comment through coalescing.
> - Claim rebuilds that run's input from its primary wake.
> - Coalescing updated the run and its receipt but left the primary
batch unchanged.
> - This change saves the accepted batch atomically in the owning
writer.
> - The benefit is complete input at claim without repeating old work.

## Linked Issues or Issue Description

Related: #18887 preserves offline queued turns. #13948 addresses
equivalent automatic wake deduplication. This change repairs the
accepted comment batch at the existing claim boundary.

**What happened?**

A queued run started with comment A. Admission accepted comment B, saved
`[A, B]` in the run context, and finished B's coalesced receipt. The
primary wake still held `[A]`. Actual claim rebuilt input from that wake
and dropped B.

**Expected behavior**

Claim must retain every accepted live comment in order. Stop, changed
source authority, deleted comments, and company boundaries must still
apply. Already claimed wakes must retain their existing recovery source.

**Steps to reproduce**

1. Queue a task run whose primary wake has an authoritative comment
batch `[A]`.
2. Admit comment B through `createHeartbeatQueue.enqueueWakeup` and the
real Postgres admission writer.
3. Confirm that admission accepts `[A, B]` and records B's coalesced
receipt.
4. Claim the original queued run. Before this fix, the claimed input
contains only A.

**Paperclip version or commit**

The untouched producer regression fails on
`d7283e875772ca8c684dc86e471e6b48d543b7b1` with both process and native
Runner/Codex adapter selection. Provider execution is disabled in the
fixture.

## What Changed

- Save the accepted comment batch in the unclaimed primary wake, run
context, and coalesced receipt in one transaction.
- Lock the primary wake before the run. Recheck the expected run status
and same-company, same-agent and run bindings; apply the existing actor
equality fence only to durable receipts before mutation.
- Preserve incoming input as normal deferred work when the locked source
changed. Preserve receipt identity and actor fields.
- Preserve the original primary actor and trigger comment while updating
its accepted batch. Each incoming receipt retains its actual actor. Keep
claimed source payloads and legacy envelopes on their existing recovery
paths.
- Safely defer a same-named different agent without rewriting the
original agent's run, wake or receipt. Missing and foreign-company
targets remain rejected. Do not rebuild input from historical receipts.
- Add actual process/native producer, real database rollback, Stop,
deletion, foreign-source, and fallback controls. Update three direct
writer fixtures for the required expected-status contract.

## Verification

- The untouched process and native producer red on
`d7283e875772ca8c684dc86e471e6b48d543b7b1` loses accepted comment B at
actual claim. The initial incomplete mock is retained separately and
excluded from behavioral proof.
- Current process/native matrix: 22 passed, including mixed authors,
system primary, same-named different agents, changed status, claimed
sources, Stop, deletion and foreign-source boundaries. Queue, Postgres
admission and use-case suites passed 234 cases before the
source-preserving serializer follow-up; its 22 actual producer cases
passed again afterward.
- CI on the previous head found a real ordinary mixed-author
workspace-copyback regression. The earlier 13-case selection omitted
this test; that coverage gap is preserved. The first correction exposed
a second real source-provenance/copyback failure. After preserving the
primary trigger and actor while updating only its batch, the complete
workspace recovery file passed all 188 cases, including ordinary
mixed-author input and native copyback.
- Real HTTP queued-comment consumption, discard/reorder and Stop
boundaries: 101 passed before these review corrections. Existing chat
and Slack consumers each passed their selected real writer case.
- Previous-head CI also timed out in the CLI child-server startup
fixture and a ZIP transfer case. The full untouched transfer file passed
17 cases locally. Startup remains unclassified; its intentionally
reserved port is part of the fixture and is not diagnosed as the cause.
No CI retry was issued.
- An earlier queue run passed its 231 assertions but timed out during
teardown; its preserved replay passed all 60 queue cases. This is
distinct from the behavioral reds.
- Exact `ab59d944895774d89f950248e356e93af00f4832` server type layers
and production server/Runner build passed. The unchanged-default
serializer file passed all 11 cases. Current-head CI and both
substantive reviews remain required; previous grades and checks do not
qualify this head.
- No hosted task was retried or changed. Local tests do not establish
any hosted deferred event's cause, provider execution, native
acknowledgement, physical cleanup, or serving revision.

## Risks

- Admission now performs bounded row reads and locks for each coalesced
wake. It adds no idle scan or remote request. The existing task
transaction owns the operation; lock order follows queued claim.
- Claimed-wake recovery uses source identity. Its payload remains
unchanged, and the existing actual native copyback tests cover that
boundary.
- A source that changes before the lock can defer the new input. This
preserves work instead of consuming it into a stale run. No migration or
historical backfill is required.

## Model Used

OpenAI `gpt-6.1-sol`, with reasoning, repository inspection, shell
execution, and test tools. The runtime does not report the
context-window size. A separate OpenAI coordinator reviewed the source.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
2026-11-03 22:09:58 -06:00

808 lines
54 KiB
TypeScript
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
import { createHeartbeatQueue } from "../services/heartbeat/queue.js";
import { createHeartbeatQueueTestDependencies } from "./helpers/heartbeat-queue.js";
import { createHeartbeatScheduling, type HeartbeatSchedulingDependencies } from "../services/heartbeat/scheduling.js";
import { createHeartbeatRunState } from "../services/heartbeat/run-state.js";
import { createHeartbeatRunPreparation } from "../services/heartbeat/run-preparation.js";
import { heartbeatService } from "../services/heartbeat.js";
import { instanceSettingsService } from "../services/instance-settings.js";
import { reconcileTaskChatRetryReceipts } from "../services/task-retry.js";
import { chatActions } from "@paperclipai/db";
import { createChatPublicationResolution } from "../services/chat-channels/publications/resolution.js";
import { createChatPublicationTaskControl } from "../services/chat-channels/publications/task-control.js";
import { chatCredentialMutationLease } from "../services/chat-credential-mutation-lease.js";
import { createHeartbeatLifecycle, type HeartbeatLifecycleDependencies } from "../services/heartbeat/run-lifecycle.js";
import { createAdmissionTransactionScope, createPostgresWakeQueueAdapter, createWakeAdmissionWriter } from "../modules/wake-queue/adapters/postgres.js";
import { applyRunnerGoalPrpEvent } from "../services/runner-goals.js";
import { idleWorkSnapshot } from "../services/task-admission.js";
import { setTimeout as delay } from "node:timers/promises";
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
import { externalConversationSettlementWork } from "../services/external-conversation-settlement-work.js";
import { notifyNativeFinalizationWork } from "../services/native-finalization-notifications.js";
import { notifyChatRunMilestones } from "../services/chat-run-milestone-notifications.js";
import { notifyChatInboundWakeNotices } from "../services/chat-inbound-wake-notice-notifications.js";
import { notifyChatResponseAvailability } from "../services/chat-response-availability.js";
import { issueApprovalService } from "../services/issue-approvals.js";
import express from "express";
import request from "supertest";
import { issueRoutes } from "../routes/issues.js";
import { randomUUID } from "node:crypto";
import { and, eq } from "drizzle-orm";
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
import {
activityLog, agents, agentWakeupRequests, approvals, chatConversations, chatEndpoints,
chatMessageLinks, chatPublications, companies, createDb, heartbeatRuns, issueApprovals,
issueComments, issueRelations, issueThreadInteractions, issues, nativeRunFinalizations,
toolApplications, toolConnections, subscribeDatabaseWork, type Db,
} from "@paperclipai/db";
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
import { pendingExternalConversationSettlement, externalConversationSettlementCandidates, settleSlackConversation } from "../services/slack-conversation-lifecycle.js";
import { externalConversationStateSql, resumeExternalConversation } from "../services/slack-conversation-state.js";
import { executionIssueCondition } from "../services/issue-visibility.js";
import { dashboardService } from "../services/dashboard.js";
import { attentionService } from "../services/attention.js";
import { companySearchQuerySchema } from "@paperclipai/shared";
import { companySearchService } from "../services/company-search.js";
import { recoveryService } from "../services/recovery/service.js";
import { issueService } from "../services/issues.js";
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_EXTERNAL_CONVERSATION_TEST_DATABASE_URL");
const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
(support.supported ? describe : describe.skip)("Slack conversation idle lifecycle", () => {
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
let db: ReturnType<typeof createDb>;
beforeAll(async () => {
if (!external) database = await startEmbeddedPostgresTestDatabase("paperclip-slack-idle-");
db = createDb(external ?? database!.connectionString);
}, 90000);
afterAll(async () => {
await db?.$client.end({ timeout: 0 });
await database?.cleanup();
});
async function fixture(native = false, issueId = randomUUID()) {
const companyId = randomUUID();
const agentId = randomUUID();
const endpointId = randomUUID();
const conversationId = randomUUID();
const runId = randomUUID();
const wakeId = randomUUID();
const responseId = randomUUID();
const publicationId = randomUUID();
const applicationId = randomUUID();
const connectionId = randomUUID();
const start = new Date(Date.now() - 10000);
await db.insert(companies).values({ id: companyId, name: "Slack", issuePrefix: companyId.slice(0, 8) });
await db.insert(agents).values({ id: agentId, companyId, name: "Carl", role: "general", adapterType: "codex_local" });
await db.insert(issues).values({ id: issueId, companyId, title: "you there?", status: "in_progress", assigneeAgentId: agentId, originKind: "chat_channel" });
await db.insert(toolApplications).values({ id: applicationId, companyId, name: "Slack", type: "chat" });
await db.insert(toolConnections).values({ id: connectionId, companyId, applicationId, name: "Slack", uid: connectionId, transport: "chat_sdk", connectionPurpose: "channel" });
await db.insert(chatEndpoints).values({ id: endpointId, companyId, connectionId, provider: "slack", publicId: endpointId, assignedAgentId: agentId, status: "active" });
await db.insert(chatConversations).values({ id: conversationId, companyId, endpointId, issueId, externalConversationId: "channel", externalThreadId: runId, externalLabel: "#test" });
await db.insert(issueComments).values({ id: wakeId, companyId, issueId, authorType: "user", authorUserId: "local-board", body: "you there?", createdAt: start });
await db.insert(chatMessageLinks).values({ companyId, endpointId, conversationId, commentId: wakeId, providerMessageId: wakeId, direction: "inbound" });
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "succeeded", runtimeMode: native ? "native" : "legacy", nativeIssueId: native ? issueId : null,
resultJson: native ? { finalizationReasonCode: "external_chat_response_waiting" } : null,
createdAt: new Date(start.getTime() + 1), startedAt: start, finishedAt: new Date(),
contextSnapshot: { issueId, source: "chat:slack", wakeCommentId: wakeId, wakeCommentIds: [wakeId] } });
if (native) await db.insert(nativeRunFinalizations).values({ companyId, issueId, runId, phase: "committed" });
await db.insert(issueComments).values({ id: responseId, companyId, issueId, authorType: "agent", authorAgentId: agentId, createdByRunId: runId, body: "Yes, I’m here.",
metadata: { version: 1, authorizationReason: "allow_chat_run_presentation", sections: [] } });
await db.insert(chatPublications).values({ id: publicationId, companyId, endpointId, conversationId, issueId, commentId: responseId,
idempotencyKey: `comment:${responseId}:${endpointId}`, payload: { text: "Yes, I’m here." }, state: "published", providerMessageId: responseId, publishedAt: new Date() });
return { companyId, agentId, issueId, endpointId, conversationId, runId, wakeId, responseId, publicationId, start };
}
async function state(f: Awaited<ReturnType<typeof fixture>>) {
return (await db.select({ status: issues.status, externalConversationState: externalConversationStateSql() })
.from(issues).where(eq(issues.id, f.issueId)))[0];
}
const settle = (f: Awaited<ReturnType<typeof fixture>>) => settleSlackConversation(db, f.companyId, f.issueId);
async function exhaustedMonitor(f: Awaited<ReturnType<typeof fixture>>,
enqueueWakeup: HeartbeatSchedulingDependencies["enqueueWakeup"],
recoveryPolicy: "wake_owner" | "escalate_to_board" = "wake_owner") {
const now = new Date(), due = new Date(now.getTime() - 60_000);
await db.update(issues).set({ monitorNextCheckAt: due, monitorAttemptCount: 1,
executionPolicy: { mode: "normal", commentRequired: true, stages: [], monitor: {
nextCheckAt: due.toISOString(), maxAttempts: 1, recoveryPolicy,
} },
}).where(eq(issues.id, f.issueId));
const state = createHeartbeatRunState(db), preparation = createHeartbeatRunPreparation(db);
return createHeartbeatScheduling(db, {
getAgent: state.getAgent, getRun: state.getRun,
toAgentOrgRow: preparation.toAgentOrgRow, groupAgentOrgRowsByCompany: preparation.groupAgentOrgRowsByCompany,
issuesSvc: issueService(db), enqueueWakeup,
getSchedulingSuppression: async () => ({ suppressed: false }), getWorktreeExecutionCutoff: async () => null,
scheduleBoundedRetryForRun: vi.fn(), claimDueTimerHeartbeat: vi.fn(), parseHeartbeatPolicy: vi.fn(), nextTimerHeartbeatAt: vi.fn(),
});
}
it("notifies timer eligibility at outer commit when a published conversation becomes idle", async () => {
const f = await fixture();
const [agent] = await db.update(agents).set({ runtimeConfig: { heartbeat: {
enabled: true, intervalSec: 600, skipTimerWhenNoActionableWork: true,
} } }).where(eq(agents.id, f.agentId)).returning();
const queue = createHeartbeatQueue(db, createHeartbeatQueueTestDependencies(db));
expect(await queue.nextTimerHeartbeatAt(agent)).toBe(agent.createdAt.getTime() + 600_000);
const events: string[] = [];
const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.agentTimers, event => events.push(event));
try {
await db.transaction(async tx => {
expect(await settleSlackConversation(tx as unknown as Db, f.companyId, f.issueId)).toBe(true);
expect(events).toContain("intent");
expect(events).not.toContain("settled");
expect((await state(f)).status).toBe("in_progress");
});
expect(events).toContain("settled");
expect((await state(f)).status).toBe("in_review");
expect(await queue.nextTimerHeartbeatAt(agent)).toBeNull();
} finally { unsubscribe(); }
});
it("keeps monitor recovery pending until its wake handoff finishes", async () => {
const f = await fixture();
let release!: () => void, entered!: () => void;
const gate = new Promise<void>(resolve => { release = resolve; });
const started = new Promise<void>(resolve => { entered = resolve; });
const scheduler = await exhaustedMonitor(f, async () => {
entered(); await gate;
await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId,
source: "automation", status: "queued", payload: { issueId: f.issueId } });
return db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)).then(rows => rows[0]!);
});
const dispatch = scheduler.triggerIssueMonitor(f.issueId);
try {
await started;
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(false);
expect(await settle(f)).toBe(false);
} finally { release(); await dispatch; }
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(false);
});
it.each(["failure", "deferred"])("retains the monitor obligation after a %s recovery handoff", async mode => {
const f = await fixture();
const scheduler = await exhaustedMonitor(f, async () => {
if (mode === "failure") throw new Error("recovery handoff unavailable");
return null;
});
if (mode === "failure") await expect(scheduler.triggerIssueMonitor(f.issueId)).rejects.toThrow("recovery handoff unavailable");
else await scheduler.triggerIssueMonitor(f.issueId);
const [issue] = await db.select().from(issues).where(eq(issues.id, f.issueId));
expect(issue!.monitorNextCheckAt).not.toBeNull();
expect(issue!.monitorWakeRequestedAt).toBeNull();
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(false);
});
it.each(["original limit", "later timeout", "stable key"])("consumes accepted monitor recovery after restart (%s)", async scenario => {
const f = await fixture();
const enqueue = vi.fn<HeartbeatSchedulingDependencies["enqueueWakeup"]>(async () => null);
const scheduler = await exhaustedMonitor(f, enqueue);
const [issue] = await db.select().from(issues).where(eq(issues.id, f.issueId));
await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId,
source: "automation", status: "completed", runId: f.runId, reason: "issue_monitor_recovery",
idempotencyKey: scenario === "stable key"
? `issue-monitor-recovery:${f.issueId}:${issue!.monitorNextCheckAt!.toISOString()}`
: `issue-monitor-recovery:${f.issueId}:max_attempts_exhausted:${issue!.monitorNextCheckAt!.toISOString()}`,
payload: { issueId: f.issueId } });
const now = new Date();
if (scenario === "later timeout") {
const policy = issue!.executionPolicy!;
await db.update(issues).set({ executionPolicy: { ...policy,
monitor: { ...policy.monitor as object, timeoutAt: new Date(now.getTime() - 1).toISOString() },
} }).where(eq(issues.id, f.issueId));
}
await scheduler.triggerIssueMonitor(f.issueId, { now });
expect(enqueue).not.toHaveBeenCalled();
expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]!.monitorNextCheckAt).toBeNull();
});
it("consumes monitor recovery coalesced by the real queue writer after the joined run finishes", async () => {
const f = await fixture();
const enqueue = vi.fn<HeartbeatSchedulingDependencies["enqueueWakeup"]>(async () => null);
const scheduler = await exhaustedMonitor(f, enqueue);
const [issue] = await db.select().from(issues).where(eq(issues.id, f.issueId));
const key = `issue-monitor-recovery:${f.issueId}:${issue!.monitorNextCheckAt!.toISOString()}`;
await db.update(heartbeatRuns).set({ status: "running", finishedAt: null }).where(eq(heartbeatRuns.id, f.runId));
await db.transaction(async tx => {
await createWakeAdmissionWriter().coalesceIntoActiveExecutionRun(createAdmissionTransactionScope(f.companyId, tx as unknown as typeof db), {
companyId: f.companyId, agentId: f.agentId, activeExecutionRunId: f.runId, activeExecutionRunStatus: "running",
mergedContextSnapshot: { issueId: f.issueId }, source: "automation", triggerDetail: "system",
payload: { issueId: f.issueId }, requestedByActorType: "system", requestedByActorId: null,
idempotencyKey: key,
});
});
const [receipt] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.idempotencyKey, key));
expect(receipt).toMatchObject({ reason: "issue_execution_same_name", status: "coalesced", runId: f.runId });
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.runId));
await scheduler.triggerIssueMonitor(f.issueId);
expect(enqueue).not.toHaveBeenCalled();
expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]!.monitorNextCheckAt).toBeNull();
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.idempotencyKey, key))).toHaveLength(1);
});
it.each(["issue_monitor_recovery", "issue_execution_same_name"])("does not consume a %s receipt for a different monitor schedule", async reason => {
const f = await fixture();
const enqueue = vi.fn<HeartbeatSchedulingDependencies["enqueueWakeup"]>(async () => null);
const scheduler = await exhaustedMonitor(f, enqueue);
const [issue] = await db.select().from(issues).where(eq(issues.id, f.issueId));
const previousSchedule = new Date(issue!.monitorNextCheckAt!.getTime() - 60_000).toISOString();
await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId,
source: "automation", status: "completed", runId: f.runId, reason,
idempotencyKey: `issue-monitor-recovery:${f.issueId}:${previousSchedule}`,
payload: { issueId: f.issueId } });
await scheduler.triggerIssueMonitor(f.issueId);
expect(enqueue).toHaveBeenCalledOnce();
expect(enqueue.mock.calls[0]![1]?.idempotencyKey).toBe(
`issue-monitor-recovery:${f.issueId}:${issue!.monitorNextCheckAt!.toISOString()}`);
expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]!.monitorNextCheckAt).not.toBeNull();
});
describe("settlement worker", () => {
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
const errors: unknown[] = [];
let admitted = true;
afterEach(async () => { await coordinator?.stop(); coordinator = undefined; admitted = true; vi.useRealTimers(); expect(errors.splice(0)).toEqual([]); });
async function start(settle = settleSlackConversation) {
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, canReconcile: () => admitted,
onError: error => errors.push(error) });
await externalConversationSettlementWork(db, settle).register(coordinator, () => admitted).ready;
}
async function flush() {
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(idleWorkSnapshot().active).toBe(0); expect(coordinator!.nextWakeAt()).toBeNull(); }, { timeout: 5000 });
}
async function quiet() {
const prior = db.$client.options.debug, queries: string[] = [];
db.$client.options.debug = (_id, query) => { queries.push(query); };
try { await vi.advanceTimersByTimeAsync(120_000); await delay(10); expect(queries).toEqual([]); }
finally { db.$client.options.debug = prior; }
}
it.each(["native finalization", "chat milestone"] as const)("ignores routine company-only settlement hints from %s and resumes on execution release", async producer => {
const f = await fixture(true);
await db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId));
await start(); await flush(); await quiet();
const previous = db.$client.options.debug;
const queries: string[] = [];
db.$client.options.debug = (_id, query) => {
if (query.includes('"issues"') && query.includes("chat_conversations")) queries.push(query);
};
try {
for (let index = 0; index < 3; index += 1) {
await db.transaction(async tx => {
if (producer === "native finalization") await notifyNativeFinalizationWork(tx, { companyId: f.companyId });
else await notifyChatRunMilestones(tx, { companyId: f.companyId });
});
await flush();
}
expect(queries).toEqual([]);
} finally { db.$client.options.debug = previous; }
await issueService(db).update(f.issueId, { executionRunId: null });
await flush();
expect(await state(f)).toMatchObject({ status: "in_review", externalConversationState: "waiting" });
await quiet();
});
it.each(["inbound notice", "response availability"] as const)("keeps real company-wide %s settlement notifications enabled", async producer => {
const f = await fixture(true);
await db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId));
await start(); await flush(); await quiet();
await db.transaction(async tx => {
if (producer === "inbound notice") await notifyChatInboundWakeNotices(tx, { companyId: f.companyId });
else await notifyChatResponseAvailability(tx, { companyId: f.companyId });
await tx.update(issues).set({ executionRunId: null }).where(eq(issues.id, f.issueId));
});
await flush();
expect(await state(f)).toMatchObject({ status: "in_review", externalConversationState: "waiting" });
await quiet();
});
it("reconstructs answered turns and becomes quiet with retained history", async () => {
const f = await fixture(true); await start(); await flush();
expect(await state(f)).toMatchObject({ status: "in_review", externalConversationState: "waiting" });
await quiet();
});
it("finds ready work beyond a full page of ineligible active conversations", async () => {
const ids = Array.from({ length: 102 }, () => randomUUID()).sort();
const f = await fixture(false, ids.pop()!);
await db.insert(issues).values(ids.map(id => ({ id, companyId: f.companyId,
title: "Waiting for input", status: "in_progress", assigneeAgentId: f.agentId, originKind: "chat_channel" })));
await db.insert(chatConversations).values(ids.map(issueId => ({ companyId: f.companyId,
endpointId: f.endpointId, issueId, externalConversationId: issueId, externalLabel: "#waiting", state: "active" })));
const entries = await externalConversationSettlementWork(db).load({ companyId: f.companyId });
expect(entries.map(entry => entry.id)).toEqual([f.issueId]);
await start(); await flush();
expect(await state(f)).toMatchObject({ status: "in_review", externalConversationState: "waiting" });
await quiet();
});
it("leaves an active conversation waiting for real work without recurring SQL", async () => {
const f = await fixture();
await db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId));
await start(); await flush();
expect(await state(f)).toMatchObject({ status: "in_progress" }); await quiet();
});
it("reactivates after an issue producer releases execution", async () => {
const f = await fixture();
await db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId));
await start(); await flush(); await quiet();
await issueService(db).update(f.issueId, { executionRunId: null });
await flush(); expect(await state(f)).toMatchObject({ status: "in_review", externalConversationState: "waiting" });
});
it.each(["restore", "worktree"])("preserves %s suppression and activation", async mode => {
const f = await fixture();
const runtimeEnv = mode === "restore" ? { PAPERCLIP_DATABASE_RESTORE_IN_PROGRESS: "true" }
: { PAPERCLIP_IN_WORKTREE: "true", PAPERCLIP_INSTANCE_ID: "conversation-settlement-test" };
const settings = instanceSettingsService(db, { runtimeEnv });
if (mode === "worktree") await settings.updateExperimental({ enableWorktreeRunExecution: false });
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, canReconcile: () => admitted,
onError: error => errors.push(error) });
await heartbeatService(db, { runtimeEnv }).registerExternalConversationSettlementWork(coordinator, () => admitted).ready;
await flush(); await quiet(); expect(await state(f)).toMatchObject({ status: "in_progress" });
if (mode === "worktree") {
try { await settings.updateExperimental({ enableWorktreeRunExecution: true }); await flush();
expect(await state(f)).toMatchObject({ status: "in_review" }); await quiet();
} finally { await settings.updateExperimental({ enableWorktreeRunExecution: false }); }
}
});
it("joins in-flight settlement and reconstructs a later committed wake after restart", async () => {
const first = await fixture(), second = await fixture();
for (const f of [first, second]) await db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId));
let release!: () => void;
const gate = new Promise<void>(resolve => { release = resolve; });
const entered = vi.fn();
await start(async (...args) => { entered(); await gate; return settleSlackConversation(...args); });
await flush(); await issueService(db).update(first.issueId, { executionRunId: null });
await vi.waitFor(async () => { await vi.advanceTimersByTimeAsync(1000); expect(entered).toHaveBeenCalledTimes(1); });
await issueService(db).update(second.issueId, { executionRunId: null });
let stopped = false;
const stopping = coordinator!.stop().then(() => { stopped = true; });
try { await delay(20); expect(stopped).toBe(false); }
finally { release(); await stopping; }
expect(await state(first)).toMatchObject({ status: "in_review" });
expect(await state(second)).toMatchObject({ status: "in_progress" });
await start(); await flush(); expect(await state(second)).toMatchObject({ status: "in_review" }); await quiet();
});
it.each(["wake_owner", "escalate_to_board"] as const)("reactivates settlement checks after monitor recovery (%s)", async policy => {
const f = await fixture();
const enqueue = vi.fn<HeartbeatSchedulingDependencies["enqueueWakeup"]>(async () =>
db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)).then(rows => rows[0]!));
const scheduler = await exhaustedMonitor(f, enqueue, policy);
await start(); await flush(); await quiet();
expect(await state(f)).toMatchObject({ status: "in_progress" });
await scheduler.triggerIssueMonitor(f.issueId);
await flush();
if (policy === "wake_owner") {
expect(enqueue).toHaveBeenCalledOnce();
expect(await state(f)).toMatchObject({ status: "in_review", externalConversationState: "waiting" });
} else {
const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, f.issueId));
expect(comments.some(comment => comment.body.includes("Recovery policy: escalate_to_board"))).toBe(true);
expect(await state(f)).toMatchObject({ status: "in_progress", externalConversationState: "active" });
expect(enqueue).not.toHaveBeenCalled();
}
await quiet();
});
it("reactivates when a failed chat retry releases its task receipt", async () => {
const f = await fixture(), wakeId = randomUUID(), actionId = randomUUID();
await db.insert(chatActions).values({ id: actionId, companyId: f.companyId, endpointId: f.endpointId,
kind: "failed_run_retry", providerActionId: actionId, status: "failed", payload: { issueId: f.issueId } });
await db.insert(agentWakeupRequests).values({ id: wakeId, companyId: f.companyId, agentId: f.agentId, source: "on_demand",
reason: "task_retry", status: "deferred_issue_execution", payload: { issueId: f.issueId, taskRetry: {
schema: "paperclip.task-retry.v1", generationId: wakeId, sourceRunId: f.runId, actorId: "local-board",
strategy: "evaluate", chatActionId: actionId } } });
await start(); await flush(); await quiet();
await reconcileTaskChatRetryReceipts(db);
await flush(); expect(await state(f)).toMatchObject({ status: "in_review" }); await quiet();
});
it.each(["run", "publication"])("wakes from real run/publication owners when %s finishes first", async first => {
const f = await fixture();
await db.update(heartbeatRuns).set({ status: "running", finishedAt: null }).where(eq(heartbeatRuns.id, f.runId));
await db.update(chatPublications).set({ state: "delivery_unknown" }).where(eq(chatPublications.id, f.publicationId));
const resolution = createChatPublicationResolution(db, {
endpointRecord: async id => {
const [endpoint] = await db.select().from(chatEndpoints).where(eq(chatEndpoints.id, id));
return { endpoint, assignedAgentName: "Carl", connectionName: "Slack", applicationId: randomUUID(), credentialSecretRefs: [] };
},
withCredentialMutationLease: chatCredentialMutationLease(db),
teamsTransferProtocol: vi.fn(() => { throw new Error("Slack does not use Teams file resolution"); }),
commitTaskControlCompletion: createChatPublicationTaskControl(db).commitTaskControlCompletion,
processPendingPublications: vi.fn(async () => 0),
processProviderEffect: vi.fn(async () => "processed" as const),
});
const lifecycle = createHeartbeatLifecycle(db, { emitTerminalAgentTaskRun: vi.fn() } as unknown as HeartbeatLifecycleDependencies);
const finishRun = () => lifecycle.setRunStatus(f.runId, "succeeded", { finishedAt: new Date() });
const publish = () => resolution.resolvePublication(f.endpointId, f.publicationId, "mark_delivered", "local-board");
await start(); await flush(); await quiet();
await (first === "run" ? finishRun() : publish()); await flush();
expect(await state(f)).toMatchObject({ status: "in_progress" }); await quiet();
await (first === "run" ? publish() : finishRun()); await flush();
expect(await state(f)).toMatchObject({ status: "in_review" }); await quiet();
});
it("reactivates after administrative execution release", async () => {
const f = await fixture();
await db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId));
await start(); await flush(); await quiet();
await issueService(db).adminForceRelease(f.issueId);
await flush(); expect(await state(f)).toMatchObject({ status: "in_review" }); await quiet();
});
it("reactivates after a blocking issue is removed", async () => {
const f = await fixture(), blockerId = randomUUID();
await db.insert(issues).values({ id: blockerId, companyId: f.companyId, title: "Blocker", status: "todo" });
await issueService(db).update(f.issueId, { blockedByIssueIds: [blockerId] });
await start(); await flush(); await quiet();
await issueService(db).remove(blockerId);
await flush(); expect(await state(f)).toMatchObject({ status: "in_review" }); await quiet();
});
it.each(["release", "cancel_wake", "fail_wake"])("reactivates through the wake queue's %s transaction", async operation => {
const f = await fixture(true), wakeId = randomUUID();
if (operation === "release") await db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId));
else await db.insert(agentWakeupRequests).values({ id: wakeId, companyId: f.companyId, agentId: f.agentId,
source: "on_demand", status: "deferred_issue_execution", payload: { issueId: f.issueId } });
await start(); await flush(); await quiet();
const adapter = createPostgresWakeQueueAdapter(db, {
resolveResponsibleUserId: async () => "local-board",
getRoutineEnv: async () => ({ routineId: null, env: null, responsibleUserId: null }),
resolveSessionBeforeForWakeup: async () => null,
});
let entered = false;
await adapter.withIssueExecutionLock({ companyId: f.companyId, runId: f.runId, now: new Date() }, async (_locked, ports) => {
entered = true;
if (operation === "cancel_wake") await ports.transaction.cancelDeferredWake({ companyId: f.companyId, wakeId, agentId: f.agentId, reason: "cancelled", now: new Date() });
if (operation === "fail_wake") await ports.transaction.failDeferredWake({ companyId: f.companyId, wakeId, agentId: f.agentId, now: new Date() });
return { outcome: { kind: "released" as const }, postCommitEffects: [] };
});
expect(entered).toBe(true);
await flush(); expect(await state(f)).toMatchObject({ status: "in_review" }); await quiet();
});
it.each(["clear", "complete"])("reactivates after the goal owner reports %s", async outcome => {
const f = await fixture();
const binding = { companyId: f.companyId, issueId: f.issueId, agentId: f.agentId, adapterType: "codex_local" };
await applyRunnerGoalPrpEvent(db, binding, { eventType: "session.goal.updated", sourceSeq: 1,
payload: { goal: { objective: "Finish the conversation", status: "active" } } });
await start(); await flush(); await quiet();
expect(await state(f)).toMatchObject({ status: "in_progress" });
await applyRunnerGoalPrpEvent(db, binding, { eventType: outcome === "clear" ? "session.goal.cleared" : "session.goal.updated",
sourceSeq: 2, payload: outcome === "clear" ? {} : { goal: { objective: "Finish the conversation", status: "complete" } } });
await flush(); expect(await state(f)).toMatchObject({ status: "in_review" }); await quiet();
});
it.each(["remove", "tombstone"])("reactivates after the comment owner performs %s", async operation => {
const f = await fixture(), newer = randomUUID();
await db.insert(issueComments).values({ id: newer, companyId: f.companyId, issueId: f.issueId,
authorType: "user", authorUserId: "local-board", body: "another request" });
await start(); await flush(); await quiet();
if (operation === "remove") await issueService(db).removeComment(newer);
else await issueService(db).tombstoneComment(newer, { actorType: "user", userId: "local-board" });
await flush(); expect(await state(f)).toMatchObject({ status: "in_review" }); await quiet();
});
it("reactivates dependents when a different issue completes", async () => {
const f = await fixture(), blockerId = randomUUID();
await db.insert(issues).values({ id: blockerId, companyId: f.companyId, title: "Blocker", status: "todo" });
await issueService(db).update(f.issueId, { blockedByIssueIds: [blockerId] });
await start(); await flush(); await quiet();
await issueService(db).update(blockerId, { status: "done" });
await flush(); expect(await state(f)).toMatchObject({ status: "in_review" }); await quiet();
});
it.each(["commit", "rollback"])("honors the issue writer's outer %s", async outcome => {
const f = await fixture();
await db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId));
await start(); await flush();
const transaction = db.transaction(async tx => {
await issueService(db).update(f.issueId, { executionRunId: null }, tx);
await vi.advanceTimersByTimeAsync(5000);
expect(await state(f)).toMatchObject({ status: "in_progress" });
if (outcome === "rollback") throw new Error("rollback execution release");
});
if (outcome === "rollback") await expect(transaction).rejects.toThrow("rollback execution release");
else await transaction;
await flush();
expect(await state(f)).toMatchObject({ status: outcome === "rollback" ? "in_progress" : "in_review" });
await quiet();
});
it("preserves a release committed during drain until resume", async () => {
const f = await fixture();
await db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId));
await start(); await flush(); admitted = false;
await issueService(db).update(f.issueId, { executionRunId: null });
await vi.advanceTimersByTimeAsync(120_000); await quiet();
expect(await state(f)).toMatchObject({ status: "in_progress" });
admitted = true; await vi.advanceTimersByTimeAsync(30_000); await flush();
expect(await state(f)).toMatchObject({ status: "in_review" }); await quiet();
});
it("reconstructs failed settlement after drain and restart", async () => {
const f = await fixture();
await start(async () => { throw new Error("settlement unavailable"); });
expect(errors.length).toBeGreaterThan(0); errors.splice(0);
expect(coordinator!.nextWakeAt()).not.toBeNull(); admitted = false;
await vi.advanceTimersByTimeAsync(120_000); await quiet();
expect(await state(f)).toMatchObject({ status: "in_progress" });
await coordinator!.stop(); admitted = true; await start(); await flush();
expect(await state(f)).toMatchObject({ status: "in_review" }); await quiet();
});
it("reactivates when the approval owner removes the last blocker", async () => {
const f = await fixture(), approvalId = randomUUID();
await db.insert(approvals).values({ id: approvalId, companyId: f.companyId, type: "approve_ceo_strategy", status: "pending", payload: {} });
await issueApprovalService(db).link(f.issueId, approvalId);
await start(); await flush(); await quiet();
await issueApprovalService(db).unlink(f.issueId, approvalId);
await flush(); expect(await state(f)).toMatchObject({ status: "in_review", externalConversationState: "waiting" });
});
});
it("discovers active candidates without treating unready or retained conversations as work", async () => {
const f = await fixture();
await db.update(heartbeatRuns).set({ status: "running", finishedAt: null }).where(eq(heartbeatRuns.id, f.runId));
expect(await externalConversationSettlementCandidates(db, { companyId: f.companyId })).toEqual([{ id: f.issueId, companyId: f.companyId }]);
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(false);
expect(await externalConversationSettlementCandidates(db, { companyId: f.companyId }, f.issueId)).toEqual([]);
expect(await externalConversationSettlementCandidates(db, { companyId: randomUUID(), issueId: f.issueId })).toEqual([]);
await db.update(chatConversations).set({ state: "waiting" }).where(eq(chatConversations.id, f.conversationId));
expect(await externalConversationSettlementCandidates(db, { companyId: f.companyId })).toEqual([]);
});
it.each([false, true])("settles a published answered turn once (native=%s)", async (native) => {
const f = await fixture(native);
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(true);
expect(await settle(f)).toBe(true);
expect(await state(f)).toEqual({ status: "in_review", externalConversationState: "waiting" });
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(false);
expect(await settle(f)).toBe(false);
expect(await db.select().from(activityLog).where(and(eq(activityLog.entityId, f.issueId), eq(activityLog.action, "issue.updated")))).toHaveLength(1);
expect(await db.select().from(issues).where(and(eq(issues.id, f.issueId), executionIssueCondition()))).toHaveLength(0);
expect((await issueService(db).getById(f.issueId))?.externalConversationState).toBe("waiting");
expect(await issueService(db).list(f.companyId)).toHaveLength(0);
expect((await dashboardService(db).summary(f.companyId)).tasks.open).toBe(0);
expect((await attentionService(db).list(f.companyId)).items.filter((item) => item.subject.id === f.issueId || item.relatedIssue?.id === f.issueId)).toHaveLength(0);
const results = await companySearchService(db).search(f.companyId, companySearchQuerySchema.parse({ q: "you there" }));
expect(results.results.find((result) => result.issue?.id === f.issueId)?.issue?.externalConversationState).toBe("waiting");
expect((await issueService(db).list(f.companyId, { q: "you there" }))[0]?.externalConversationState).toBe("waiting");
expect((await issueService(db).listReviewAttention(f.companyId, [{ id: f.issueId, companyId: f.companyId, status: "in_review" }])).get(f.issueId)?.state).toBe("none");
});
it("settles and resumes Speko-owned tasks without adopting existing task lifecycles", async () => {
const f = await fixture(true);
await db.update(chatEndpoints).set({ provider: "speko" }).where(eq(chatEndpoints.id, f.endpointId));
await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issueId, source: "chat:speko", wakeCommentId: f.wakeId, wakeCommentIds: [f.wakeId] } }).where(eq(heartbeatRuns.id, f.runId));
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(false);
expect(await settle(f)).toBe(false);
await db.update(issues).set({ originId: f.endpointId }).where(eq(issues.id, f.issueId));
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(true);
expect(await settle(f)).toBe(true);
expect(await state(f)).toMatchObject({ status: "in_review", externalConversationState: "waiting" });
await db.transaction(tx => resumeExternalConversation(tx as unknown as typeof db, f.companyId, f.issueId));
expect(await state(f)).toMatchObject({ status: "todo", externalConversationState: "active" });
});
it("preserves waiting state in compact list responses used by the Inbox", async () => {
const f = await fixture();
await settle(f);
const app = express();
app.use((req, _res, next) => {
(req as any).actor = { type: "board", userId: "local-board", source: "local_implicit", isInstanceAdmin: true };
next();
});
app.use("/api", issueRoutes(db, {} as any));
const response = await request(app).get(`/api/companies/${f.companyId}/issues`)
.query({ view: "compact", q: "you there" });
expect(response.status, JSON.stringify(response.body)).toBe(200);
expect(response.body).toEqual([expect.objectContaining({ id: f.issueId, status: "in_review", externalConversationState: "waiting" })]);
});
it("defers generic dispatch while the settlement worker owns an answered turn", async () => {
const f = await fixture();
const enqueueWakeup = vi.fn(async () => null);
const recovery = recoveryService(db, { enqueueWakeup });
const [{ createdAt }] = await db.select({ createdAt: issues.createdAt }).from(issues).where(eq(issues.id, f.issueId));
await recovery.reconcileStrandedAssignedIssues({ skipExternalConversationSettlement: true, issueCreatedAtGte: createdAt });
expect(await state(f)).toMatchObject({ status: "in_progress", externalConversationState: "active" });
expect(enqueueWakeup).not.toHaveBeenCalled();
expect(await settle(f)).toBe(true);
await recovery.reconcileStrandedAssignedIssues({ skipExternalConversationSettlement: true, issueCreatedAtGte: createdAt });
expect(enqueueWakeup).not.toHaveBeenCalled();
});
it("reconciles an already-answered conversation without another model run", async () => {
const f = await fixture();
const enqueueWakeup = vi.fn(async () => null);
const recovery = recoveryService(db, { enqueueWakeup });
// Restrict to this newly created fixture, like the rollout reconciler's candidate scan.
const [issue] = await db.select().from(issues).where(eq(issues.id, f.issueId));
await recovery.reconcileStrandedAssignedIssues({ issueCreatedAtGte: issue.createdAt });
expect((await state(f))?.externalConversationState).toBe("waiting");
await recovery.reconcileStrandedAssignedIssues({ issueCreatedAtGte: issue.createdAt });
expect(enqueueWakeup).not.toHaveBeenCalled();
});
it("waits for both finalization and delivery, regardless of their order", async () => {
for (const first of ["run", "publication"]) {
const f = await fixture();
if (first === "run") await db.update(chatPublications).set({ state: "pending" }).where(eq(chatPublications.id, f.publicationId));
else await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, f.runId));
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(false);
expect(await settle(f)).toBe(false);
await db.update(chatPublications).set({ state: "published" }).where(eq(chatPublications.id, f.publicationId));
await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.runId));
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(true);
expect(await settle(f)).toBe(true);
}
});
it("a new board message reactivates the same conversation and prevents stale settlement", async () => {
const f = await fixture();
await settle(f);
const followup = await issueService(db).addComment(f.issueId, "what's your name?", { userId: "local-board" });
expect((await state(f))?.externalConversationState).toBe("active");
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(false);
expect(await settle(f)).toBe(false);
expect((await state(f))?.status).toBe("todo");
await issueService(db).checkout(f.issueId, f.agentId, ["todo", "backlog", "blocked"], null);
expect((await state(f))?.status).toBe("in_progress");
const nextRun = randomUUID();
await db.insert(chatMessageLinks).values({ companyId: f.companyId, endpointId: f.endpointId, conversationId: f.conversationId, commentId: followup.id, providerMessageId: followup.id, direction: "inbound" });
await db.insert(heartbeatRuns).values({ id: nextRun, companyId: f.companyId, agentId: f.agentId, runtimeMode: "legacy", status: "succeeded", finishedAt: new Date(),
contextSnapshot: { issueId: f.issueId, source: "chat:slack", wakeCommentId: followup.id } });
const [response] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, authorAgentId: f.agentId, createdByRunId: nextRun, body: "Carl", metadata: { version: 1, authorizationReason: "allow_chat_run_presentation", sections: [] } }).returning();
await db.insert(chatPublications).values({ companyId: f.companyId, endpointId: f.endpointId, conversationId: f.conversationId, issueId: f.issueId, commentId: response.id,
idempotencyKey: nextRun, payload: { text: "Carl" }, state: "published", providerMessageId: nextRun });
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(true);
expect(await settle(f)).toBe(true);
expect((await state(f))?.externalConversationState).toBe("waiting");
});
it("serializes simultaneous finalizers and a new message without losing the message", async () => {
const f = await fixture();
await Promise.all([settle(f), settle(f), issueService(db).addComment(f.issueId, "another request", { userId: "local-board" })]);
expect((await state(f))?.externalConversationState).toBe("active");
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(false);
expect(await settle(f)).toBe(false);
});
it("lets an operator explicitly request review instead of keeping Idle", async () => {
const f = await fixture();
await settle(f);
await issueService(db).update(f.issueId, { status: "in_review", actorUserId: "local-board" });
expect(await state(f)).toEqual({ status: "in_review", externalConversationState: "active" });
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(false);
expect(await settle(f)).toBe(false);
});
it("duplicate board delivery does not reactivate an answered turn", async () => {
const f = await fixture();
const clientRequestId = randomUUID();
await db.update(issueComments).set({ clientRequestId }).where(eq(issueComments.id, f.wakeId));
await settle(f);
const duplicate = await issueService(db).addComment(f.issueId, "you there?", { userId: "local-board" }, { clientRequestId });
expect(duplicate.id).toBe(f.wakeId);
expect((await state(f))?.externalConversationState).toBe("waiting");
});
it("keeps delivery errors visible after a conversation has settled", async () => {
const f = await fixture();
await settle(f);
await db.update(chatPublications).set({ state: "failed" }).where(eq(chatPublications.id, f.publicationId));
expect((await state(f))?.externalConversationState).toBe("active");
expect(await db.select().from(issues).where(and(eq(issues.id, f.issueId), executionIssueCondition()))).toHaveLength(1);
});
it("does not settle past an unlinked Slack guest's newer message", async () => {
const f = await fixture();
const [message] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, authorType: "system", body: "one more thing" }).returning();
await db.insert(chatMessageLinks).values({ companyId: f.companyId, endpointId: f.endpointId, conversationId: f.conversationId, commentId: message.id, providerMessageId: message.id, direction: "inbound" });
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(false);
expect(await settle(f)).toBe(false);
});
it("requires committed native response-wait evidence", async () => {
const f = await fixture(true);
await db.update(nativeRunFinalizations).set({ phase: "arbitrating" }).where(eq(nativeRunFinalizations.runId, f.runId));
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(false);
expect(await settle(f)).toBe(false);
await db.update(nativeRunFinalizations).set({ phase: "committed" }).where(eq(nativeRunFinalizations.runId, f.runId));
await db.update(heartbeatRuns).set({ resultJson: { finalizationReasonCode: "governed_response_waiting" } }).where(eq(heartbeatRuns.id, f.runId));
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(false);
expect(await settle(f)).toBe(false);
});
it("settles an authorized published reply after an interaction continuation", async () => {
const f = await fixture(true);
const [answer] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId,
authorType: "user", authorUserId: "local-board", body: "Here is the requested answer", createdAt: new Date(f.start.getTime() + 2) }).returning();
await db.update(heartbeatRuns).set({ contextSnapshot: {
issueId: f.issueId, source: "issue.interaction.respond", externalChatContinuation: true, wakeCommentId: answer.id,
} }).where(eq(heartbeatRuns.id, f.runId));
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(true);
expect(await settle(f)).toBe(true);
});
it("waits for execution release before moving the issue to review", async () => {
const f = await fixture();
await db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId));
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(false);
expect(await settle(f)).toBe(false);
expect((await state(f))?.status).toBe("in_progress");
await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, f.issueId));
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(true);
expect(await settle(f)).toBe(true);
});
it.each([
{ status: "pending" },
{ status: "changes_requested" },
{ status: "idle", monitor: { status: "scheduled" } },
{ status: "idle", monitor: { status: "triggered" } },
])("preserves governed or monitored unfinished work: %j", async (executionState) => {
const f = await fixture();
await db.update(issues).set({ executionState }).where(eq(issues.id, f.issueId));
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(false);
expect(await settle(f)).toBe(false);
});
it("preserves pending interactions and removes Idle if a real review is requested later", async () => {
const f = await fixture();
await settle(f);
await db.insert(issueThreadInteractions).values({ companyId: f.companyId, issueId: f.issueId, kind: "request_confirmation", payload: { version: 1, prompt: "Approve?" } });
expect((await state(f))?.externalConversationState).toBe("active");
await db.update(issues).set({ status: "in_progress" }).where(eq(issues.id, f.issueId));
await db.update(chatConversations).set({ state: "active" }).where(eq(chatConversations.id, f.conversationId));
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(false);
expect(await settle(f)).toBe(false);
});
it.each(["failed", "cancelled", "timed_out", "queued", "running", "scheduled_retry"])("does not park a %s run", async (status) => {
const f = await fixture();
await db.update(heartbeatRuns).set({ status }).where(eq(heartbeatRuns.id, f.runId));
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(false);
expect(await settle(f)).toBe(false);
});
it.each(["pending", "failed", "delivery_unknown", "retry", "streaming", "awaiting_consent"] as const)("does not park %s delivery", async (state) => {
const f = await fixture();
await db.update(chatPublications).set({ state }).where(eq(chatPublications.id, f.publicationId));
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(false);
expect(await settle(f)).toBe(false);
});
it.each(["done", "blocked", "cancelled", "in_review"])("preserves explicit %s disposition", async (status) => {
const f = await fixture();
await db.update(issues).set({ status }).where(eq(issues.id, f.issueId));
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(false);
expect(await settle(f)).toBe(false);
});
it.each(["completed", "unavailable", "endpoint_removed"])("does not reactivate a %s conversation", async (state) => {
const f = await fixture();
await db.update(chatConversations).set({ state }).where(eq(chatConversations.id, f.conversationId));
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(false);
expect(await settle(f)).toBe(false);
});
it("respects company, provider, and connection boundaries", async () => {
const f = await fixture();
expect(await settleSlackConversation(db, randomUUID(), f.issueId)).toBe(false);
await db.update(chatEndpoints).set({ status: "revoked" }).where(eq(chatEndpoints.id, f.endpointId));
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(false);
expect(await settle(f)).toBe(false);
await db.update(chatEndpoints).set({ status: "active", provider: "telegram" }).where(eq(chatEndpoints.id, f.endpointId));
expect(await pendingExternalConversationSettlement(db, f.companyId, f.issueId)).toBe(false);
expect(await settle(f)).toBe(false);
});
it("preserves scheduled work, newer runs, queued wakes, monitors, blockers, and approvals", async () => {
const monitor = await fixture();
await db.update(issues).set({ monitorNextCheckAt: new Date(Date.now() + 60000) }).where(eq(issues.id, monitor.issueId));
expect(await settle(monitor)).toBe(false);
const queued = await fixture();
await db.insert(agentWakeupRequests).values({ companyId: queued.companyId, agentId: queued.agentId, source: "automation", status: "deferred_issue_execution", payload: { issueId: queued.issueId } });
expect(await settle(queued)).toBe(false);
const newer = await fixture();
await db.insert(heartbeatRuns).values({ companyId: newer.companyId, agentId: newer.agentId, status: "scheduled_retry", contextSnapshot: { issueId: newer.issueId } });
expect(await settle(newer)).toBe(false);
const blocked = await fixture();
const [blocker] = await db.insert(issues).values({ companyId: blocked.companyId, title: "dependency", status: "todo" }).returning();
await db.insert(issueRelations).values({ companyId: blocked.companyId, issueId: blocker.id, relatedIssueId: blocked.issueId, type: "blocks" });
expect(await settle(blocked)).toBe(false);
const review = await fixture();
const [approval] = await db.insert(approvals).values({ companyId: review.companyId, type: "hire_agent", status: "pending", payload: {} }).returning();
await db.insert(issueApprovals).values({ companyId: review.companyId, issueId: review.issueId, approvalId: approval.id });
expect(await settle(review)).toBe(false);
});
});