mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-21 08:22:20 +02:00
fix(chat): keep each chat conversation in its own wake batch (#17054)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Chat channels let a linked user talk to an agent from Slack, GitHub and other platforms. Each external thread is a chat conversation that is bound to one issue. > - One issue can have many conversations. A GitHub pull request review makes one conversation for each inline finding thread. > - When replies arrive behind an active run, the wake queue merges them into one deferred wake. The merge did not look at the conversation. > - A reviewed chat run can attest only one conversation. A merged batch from two threads always failed attestation with `setup_failed` / `reviewed_chat_execution_binding_not_authorized`, so no reply got an answer. > - This pull request keeps each chat conversation in its own wake batch. The single-conversation attestation check stays the same. > - The benefit is that replies in different review threads each get their own attested run, one after the other. ## Linked Issues or Issue Description Refs #16793 (related: chat origin during coalescing). No public issue exists, so the bug report follows. **What happened?** A linked user replied in three different inline review-finding threads of one pull request while the agent's review run was still active. The three wakes merged into one deferred wake. The promoted run failed in about 2 seconds with `setup_failed` / `reviewed_chat_execution_binding_not_authorized` and `providerWorkStarted=false`. None of the three replies got an answer. The failed-chat retry cannot recover this batch, because a retry source holds one conversation. **Steps to reproduce** 1. Bind two chat conversations (two GitHub finding threads) to one `in_review` issue. 2. Start a run for the assigned agent on that issue. 3. As the linked user, reply once in each thread. 4. Finish the active run. 5. See one promoted run that fails with `reviewed_chat_execution_binding_not_authorized`. **Expected behavior** Each thread gets its own run. Each run attests its one conversation and answers its reply. **Cause** `findExistingDeferredWake` matched on company, agent, issue and actor only. The active-run coalesce path had the same gap. `authorizeChatConversationForBoundRun` correctly denies a batch with more than one conversation. ## What Changed - `run-dispatch/domain/wake-context.ts`: add the `chatConversationId` wake-context key, a reader, and `wakeChatConversationsConflict`. A conflict means that both wakes have a conversation and the two conversations are different. - `issue-assignment-wakeup.ts` and `chat-channels/inbound/wakeups.ts`: stamp the conversation ID on each inbound chat wake. - `chat-channels/failed-run-retry.ts`: stamp the conversation ID on failed-chat retry wakes, so later replies do not merge into a retry from another conversation. - `wake-queue` admission: do not coalesce into an active run of another conversation. The deferred-wake lookup now matches only wakes of the same conversation or wakes without a conversation. - `heartbeat/queue.ts`: the same-scope queued or running coalesce target is skipped when the conversations conflict. - `doc/execution-semantics.md`: describe per-conversation chat wake batching. - Tests: an integration regression in the GitHub review block of `chat-channels.integration.test.ts`, and unit cases for the active-run branch in `wake-queue/application/use-cases.test.ts`. ## Verification - Red before the fix: `pnpm --dir server exec vitest run src/__tests__/chat-channels.integration.test.ts -t "replies in review finding threads"`. The main case fails with one run: `{ status: "failed", errorCode: "setup_failed", error: "reviewed_chat_execution_binding_not_authorized" }`. - Green after the fix, same command, 3 of 3 pass: - Replies in two finding threads give two successive runs. Each run succeeds, has `paperclipExternalChatExecutionBound: true`, and holds only its own comment and conversation. - Two replies in the same thread still give one run with both comments. - After the linked identity is revoked, each separated run is still denied (`setup_failed`) and the adapter does not execute. - `vitest run src/modules/wake-queue src/modules/run-dispatch`: all pass, with the new cases (different conversation, same conversation, run without a conversation, non-chat wake). - Related suites pass: `heartbeat-comment-wake-batching`, `durable-chat-wakeup`, `chat-failed-run-retry-boundary`, `issue-queued-comments-routes`, `heartbeat-stale-queue-invalidation`. - Full `chat-channels.integration.test.ts`: 823 passed, 1 timed out. The timed-out test (`GitHub terminal receipt composes pinned HTTP and durable source authority (held_token)`) passes alone with and without this change. The sandbox has a 4 GB memory cap. - Not run locally: `pnpm -r typecheck` and `pnpm build`. The server `tsc` process was killed at the 4 GB sandbox memory cap. CI runs both. - Runtime dependency note: the deferred-wake lookup adds one JSONB predicate to the existing deferred-wake query. It reads the same rows. The query count does not change. The integration regression runs it against embedded Postgres. ## Risks - Low risk. No schema change and no migration. - Behavior change: chat replies from different conversations on one issue now run one after the other, not in one batch. Before this change, such a batch always failed, so no working flow becomes slower. - Wakes without a conversation (non-chat wakes, and wakes queued before this deploy) keep the old coalescing. A deferred wake from before the deploy has no conversation key, so one later reply can still merge into it. That window closes when the queue drains. - The check keeps the single-conversation attestation in `authorizeChatConversationForBoundRun`. Revoked identities are still denied. ## Model Used - Anthropic Claude Opus 5.5 (`claude-opus-5-5`), in Claude Code through the Paperclip agent harness. It used tool use (shell, file editing, test runs) and extended reasoning. ## 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 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: devinfoley <139239+devinfoley@users.noreply.github.com> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
13 files changed
+198
-14
No files matched your search
@@ -170,6 +170,13 @@ to existing answered threads without another model invocation. Slack identity,
|
||||
permissions, and the ability to execute work in the same thread are unchanged.
|
||||
Other providers keep their existing lifecycle.
|
||||
|
||||
Several chat conversations can share one task, such as the inline finding
|
||||
threads of a GitHub pull request review. A reviewed chat run attests exactly one
|
||||
conversation, so wake coalescing keeps each conversation in its own batch.
|
||||
Messages from the same conversation still coalesce into one run. Messages from
|
||||
different conversations queue separately and run one after the other. Wakes
|
||||
without a conversation keep their normal coalescing.
|
||||
|
||||
### `done`
|
||||
|
||||
The work is complete and terminal.
|
||||
|
||||
@@ -184,12 +184,14 @@ import {
|
||||
CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON,
|
||||
} from "../services/chat-run-publications.js";
|
||||
import {
|
||||
extractWakeCommentIds,
|
||||
heartbeatService,
|
||||
resolveExternalChatWakeProvider,
|
||||
} from "../services/heartbeat.js";
|
||||
import { buildPaperclipWakePayload, resolveRunScopedMentionedSkillKeys } from "../services/heartbeat/run-preparation.js";
|
||||
import {
|
||||
registerServerAdapter,
|
||||
runningProcesses,
|
||||
unregisterServerAdapter,
|
||||
} from "../adapters/index.js";
|
||||
import { projectSafeChatPublicationText } from "../services/chat-publication-projection.js";
|
||||
@@ -3158,6 +3160,103 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
expect(await githubBotToolsForSession(db, { ...f.session, issueId: randomUUID() })).toEqual([]);
|
||||
expect(await readChatConversationsByEndpointId(f.endpoint.id)).toHaveLength(2);
|
||||
});
|
||||
describe("replies in review finding threads behind an active run", () => {
|
||||
const pr = "github:paperclipai/paperclip:17044";
|
||||
async function findingThreadFixture() {
|
||||
vi.stubEnv("PAPERCLIP_API_URL", "https://paperclip.example.test");
|
||||
onTestFinished(() => vi.unstubAllEnvs());
|
||||
const f = await subscribedGitHubFixture(pr);
|
||||
const adapterType = `github-finding-thread-test-${randomUUID()}`;
|
||||
const execute = vi.fn(async (_input: { runId: string; context: Record<string, unknown> }) => ({
|
||||
exitCode: 0, signal: null, timedOut: false, errorMessage: null, summary: "Answered the finding thread",
|
||||
}));
|
||||
registerServerAdapter({ type: adapterType, supportsLocalAgentJwt: false, execute,
|
||||
testEnvironment: async () => ({ adapterType, status: "pass", checks: [], testedAt: new Date().toISOString() }) });
|
||||
const heartbeat = heartbeatService(db);
|
||||
const activeRunId = randomUUID();
|
||||
onTestFinished(async () => {
|
||||
runningProcesses.delete(activeRunId);
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
unregisterServerAdapter(adapterType);
|
||||
});
|
||||
const issueId = f.conversation.issueId!;
|
||||
await db.update(companies).set({ defaultResponsibleUserId: "owner-user" }).where(eq(companies.id, f.companyId));
|
||||
await db.update(agents).set({ adapterType, runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } } })
|
||||
.where(eq(agents.id, f.assignedAgentId));
|
||||
// Published findings bind their inline threads to the PR review task.
|
||||
const { id: _id, ...base } = f.conversation;
|
||||
const threads = await Promise.all(["4332012267", "4332012626"].map(async (commentId) => {
|
||||
const thread = makeThread({ channelId: "paperclipai/paperclip", id: `${pr}:rc:${commentId}` }).thread;
|
||||
const [conversation] = await db.insert(chatConversations).values({ ...base, externalThreadId: thread.id,
|
||||
providerUrl: `https://github.com/paperclipai/paperclip/pull/17044#discussion_r${commentId}` }).returning();
|
||||
return { thread, conversation };
|
||||
}));
|
||||
// The review-request run still holds the task when the linked user replies.
|
||||
await db.insert(heartbeatRuns).values({ id: activeRunId, companyId: f.companyId, agentId: f.assignedAgentId,
|
||||
status: "running", startedAt: new Date(), contextSnapshot: { issueId, source: "chat:github" },
|
||||
resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } } });
|
||||
runningProcesses.set(activeRunId, { child: {} as never, graceSec: 0, processGroupId: null });
|
||||
await db.update(issues).set({ status: "in_review", executionRunId: activeRunId,
|
||||
executionAgentNameKey: "maya", executionLockedAt: new Date() }).where(eq(issues.id, issueId));
|
||||
f.wakeup.mockImplementation(heartbeat.wakeup);
|
||||
let messageId = 41950;
|
||||
const reply = (thread: Thread, text: string) => deliverMessage({ callbacks: f.callbacks, endpointId: f.endpoint.id,
|
||||
provider: "github", thread, trigger: "subscribed_message",
|
||||
message: makeMessage({ id: String(messageId++), text, userId: "42", userName: "octocat" }) });
|
||||
const commentFor = async (text: string) => (await readIssueCommentsByIssueId(issueId)).find((c) => c.body.includes(text))!.id;
|
||||
const finishActiveRun = async () => {
|
||||
runningProcesses.delete(activeRunId);
|
||||
await heartbeat.cancelRun(activeRunId, "Review request finished", { suppressImmediateRecovery: true });
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
};
|
||||
const followUpRuns = async () => (await readHeartbeatRunsByCompanyId(f.companyId))
|
||||
.filter((run) => run.id !== activeRunId && run.wakeupRequestId)
|
||||
.sort((a, b) => a.createdAt.getTime() - b.createdAt.getTime());
|
||||
return { ...f, issueId, threads, execute, reply, commentFor, finishActiveRun, followUpRuns };
|
||||
}
|
||||
|
||||
it("answers each finding thread in its own run instead of one unattestable batch", async () => {
|
||||
const f = await findingThreadFixture();
|
||||
await f.reply(f.threads[0].thread, "First finding is intentional");
|
||||
await f.reply(f.threads[1].thread, "Second finding is fixed in the next commit");
|
||||
const first = await f.commentFor("First finding"), second = await f.commentFor("Second finding");
|
||||
await f.finishActiveRun();
|
||||
const runs = await f.followUpRuns();
|
||||
expect(runs.map((run) => ({ status: run.status, errorCode: run.errorCode, error: run.error })))
|
||||
.toEqual([{ status: "succeeded", errorCode: null, error: null }, { status: "succeeded", errorCode: null, error: null }]);
|
||||
expect(runs.map((run) => extractWakeCommentIds(run.contextSnapshot))).toEqual([[first], [second]]);
|
||||
expect(runs.map((run) => run.contextSnapshot?.chatConversationId))
|
||||
.toEqual(f.threads.map(({ conversation }) => conversation.id));
|
||||
expect(f.execute.mock.calls.map(([input]) => [input.context.paperclipExternalChatExecutionBound,
|
||||
extractWakeCommentIds(input.context)])).toEqual([[true, [first]], [true, [second]]]);
|
||||
});
|
||||
|
||||
it("still batches several replies in one finding thread into one attested run", async () => {
|
||||
const f = await findingThreadFixture();
|
||||
await f.reply(f.threads[0].thread, "First note on this finding");
|
||||
await f.reply(f.threads[0].thread, "Second note on this finding");
|
||||
const ids = [await f.commentFor("First note"), await f.commentFor("Second note")];
|
||||
await f.finishActiveRun();
|
||||
const runs = await f.followUpRuns();
|
||||
expect(runs).toHaveLength(1);
|
||||
expect(runs[0]).toMatchObject({ status: "succeeded", errorCode: null });
|
||||
expect(extractWakeCommentIds(runs[0].contextSnapshot)).toEqual(ids);
|
||||
expect(f.execute).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("still denies each separated finding-thread run after the linked identity is revoked", async () => {
|
||||
const f = await findingThreadFixture();
|
||||
await f.reply(f.threads[0].thread, "First finding is intentional");
|
||||
await f.reply(f.threads[1].thread, "Second finding is fixed in the next commit");
|
||||
await db.update(chatIdentityLinks).set({ status: "revoked", revokedAt: new Date() })
|
||||
.where(eq(chatIdentityLinks.principalId, f.principal.id));
|
||||
await f.finishActiveRun();
|
||||
const runs = await f.followUpRuns();
|
||||
expect(runs.length).toBeGreaterThan(0);
|
||||
for (const run of runs) expect(run).toMatchObject({ status: "failed", errorCode: "setup_failed" });
|
||||
expect(f.execute).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
it.each([
|
||||
{ label: "returns the confirmed GitHub assessment when publication takes longer than ten seconds", delayMs: 11_000, timeoutMs: undefined },
|
||||
{ label: "preserves an explicit shorter GitHub tool deadline and reconciles without duplicate publication", delayMs: 200, timeoutMs: 50 },
|
||||
|
||||
@@ -75,6 +75,25 @@ export function deriveCommentId(
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* The external chat conversation a wake answers. Several conversations can
|
||||
* share one issue (for example GitHub review finding threads), but a reviewed
|
||||
* chat run can attest only one, so wakes for different conversations must
|
||||
* never share a batch.
|
||||
*/
|
||||
export const CHAT_CONVERSATION_ID_KEY = "chatConversationId";
|
||||
|
||||
export function readWakeChatConversationId(contextSnapshot: unknown): string | null {
|
||||
return readNonEmptyString(parseObject(contextSnapshot)[CHAT_CONVERSATION_ID_KEY]);
|
||||
}
|
||||
|
||||
/** True when both wakes answer a chat conversation and they are different ones. */
|
||||
export function wakeChatConversationsConflict(existing: unknown, incoming: unknown): boolean {
|
||||
const existingId = readWakeChatConversationId(existing);
|
||||
const incomingId = readWakeChatConversationId(incoming);
|
||||
return existingId !== null && incomingId !== null && existingId !== incomingId;
|
||||
}
|
||||
|
||||
/**
|
||||
* `allowedWakeReasons` is the issue-tree-control module's own set of wake
|
||||
* reasons that excuse an interaction wake. This file stays free of service
|
||||
|
||||
@@ -17,6 +17,9 @@ export {
|
||||
INTERACTION_CONTINUATION_INFRA_RETRY_REASON,
|
||||
INTERACTION_CONTINUATION_INFRA_WAKE_REASON,
|
||||
WAKE_COMMENT_IDS_KEY,
|
||||
CHAT_CONVERSATION_ID_KEY,
|
||||
readWakeChatConversationId,
|
||||
wakeChatConversationsConflict,
|
||||
RESOLVED_INTERACTION_CONTINUATION_STATUSES,
|
||||
isNonAssigneeWorkspaceBusyRetry,
|
||||
extractWakeCommentIds,
|
||||
|
||||
@@ -40,7 +40,7 @@ import {
|
||||
queuedCommentIdsFromWakePayload,
|
||||
withQueuedCommentIdsInWakePayload,
|
||||
} from "../../../services/issue-queued-comment-queue.js";
|
||||
import { extractWakeCommentIds } from "../../run-dispatch/index.js";
|
||||
import { CHAT_CONVERSATION_ID_KEY, extractWakeCommentIds } from "../../run-dispatch/index.js";
|
||||
import { hasInteractionContinuationWakeContext } from "../domain/context.js";
|
||||
import { decidePreDrain, type PreDrainFacts } from "../domain/policy.js";
|
||||
import {
|
||||
@@ -890,7 +890,7 @@ export function createWakeAdmissionReader(): WakeAdmissionReader {
|
||||
|
||||
async findExistingDeferredWake(
|
||||
scope,
|
||||
{ companyId, agentId, issueId, durableActor },
|
||||
{ companyId, agentId, issueId, chatConversationId, durableActor },
|
||||
) {
|
||||
const tx = requireAdmissionTx(scope, companyId);
|
||||
const row = await tx
|
||||
@@ -902,6 +902,11 @@ export function createWakeAdmissionReader(): WakeAdmissionReader {
|
||||
eq(agentWakeupRequests.agentId, agentId),
|
||||
eq(agentWakeupRequests.status, DEFERRED_WAKE_STATUS),
|
||||
sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`,
|
||||
...(chatConversationId
|
||||
? [
|
||||
sql`coalesce(${agentWakeupRequests.payload} -> ${DEFERRED_WAKE_CONTEXT_KEY}::text ->> ${CHAT_CONVERSATION_ID_KEY}::text, ${chatConversationId}) = ${chatConversationId}`,
|
||||
]
|
||||
: []),
|
||||
...(durableActor
|
||||
? [
|
||||
durableActor.type === null
|
||||
|
||||
@@ -407,6 +407,8 @@ export interface WakeAdmissionReader {
|
||||
companyId: string;
|
||||
agentId: string;
|
||||
issueId: string;
|
||||
/** Matches only deferred wakes for this chat conversation or for no conversation. */
|
||||
chatConversationId?: string | null;
|
||||
durableActor?: { type: string | null; id: string | null };
|
||||
},
|
||||
): Promise<ExistingDeferredWake | null>;
|
||||
|
||||
@@ -937,6 +937,29 @@ describe("admitWakeBehindIssueExecution", () => {
|
||||
expect(writer.insertNewDeferredWake).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ label: "another conversation's queued run", active: "conversation-a", incoming: "conversation-b", coalesces: false },
|
||||
{ label: "the same conversation's queued run", active: "conversation-a", incoming: "conversation-a", coalesces: true },
|
||||
{ label: "a queued run without a conversation", active: undefined, incoming: "conversation-b", coalesces: true },
|
||||
{ label: "a conversation's queued run from a non-chat wake", active: "conversation-a", incoming: undefined, coalesces: true },
|
||||
])("coalesces a chat reply into $label only when one conversation is attested", async ({ active, incoming, coalesces }) => {
|
||||
const writer = createFakeAdmissionWriter();
|
||||
const reader = createFakeAdmissionReader();
|
||||
const admit = createAdmitWakeBehindIssueExecution({ reader, writer, helpers: createFakeAdmissionHelpers() });
|
||||
await admit(SCOPE, admissionInput({
|
||||
contextSnapshot: incoming ? { chatConversationId: incoming } : {},
|
||||
activeExecutionRun: { ...ACTIVE_EXECUTION_RUN, status: "queued",
|
||||
contextSnapshot: active ? { chatConversationId: active } : {} },
|
||||
}));
|
||||
expect(writer.coalesceIntoActiveExecutionRun).toHaveBeenCalledTimes(coalesces ? 1 : 0);
|
||||
if (!coalesces) {
|
||||
// The separated reply still merges with its own conversation's deferred wake.
|
||||
expect(reader.findExistingDeferredWake).toHaveBeenCalledExactlyOnceWith(SCOPE,
|
||||
expect.objectContaining({ chatConversationId: incoming }));
|
||||
expect(writer.insertNewDeferredWake).toHaveBeenCalledOnce();
|
||||
}
|
||||
});
|
||||
|
||||
it("partitions durable admission by the exact actor before considering a deferred merge", async () => {
|
||||
const durableReceipt = {
|
||||
id: "other-actor-receipt",
|
||||
@@ -983,6 +1006,7 @@ describe("admitWakeBehindIssueExecution", () => {
|
||||
companyId: "company-1",
|
||||
agentId: "wake-agent",
|
||||
issueId: "issue-1",
|
||||
chatConversationId: null,
|
||||
durableActor: { type: "user", id: "other-user" },
|
||||
},
|
||||
);
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { enrichPromotedWakeContext } from "../domain/context.js";
|
||||
import { readWakeChatConversationId } from "../../run-dispatch/index.js";
|
||||
import { enrichPromotedWakeContext, wakeMustNotJoinActiveRun } from "../domain/context.js";
|
||||
import {
|
||||
decideQueuedCommentAction,
|
||||
decideReleaseRecovery,
|
||||
@@ -11,7 +12,6 @@ import {
|
||||
isConfigurationIncompleteFailedRun,
|
||||
isWorkspaceValidationFailedRun,
|
||||
readNonEmptyString,
|
||||
parseObject,
|
||||
} from "../domain/values.js";
|
||||
import type {
|
||||
AdmitWakeBehindIssueExecutionResult,
|
||||
@@ -840,11 +840,10 @@ export function createAdmitWakeBehindIssueExecution(deps: {
|
||||
requestedByActorType: input.requestedByActorType,
|
||||
requestedByActorId: input.requestedByActorId,
|
||||
}));
|
||||
// Each resolved card has its own immutable decision and continuation. Never
|
||||
// merge it with comments or another card, in either arrival order.
|
||||
// A resolved card never merges with comments or another card.
|
||||
const interactionReceipt = Boolean(input.contextSnapshot.interactionId || input.payload?.interactionId);
|
||||
const decision = decideWakeAdmission({
|
||||
allowRunCoalescing: interactionReceipt || parseObject(input.activeExecutionRun.contextSnapshot).interactionId
|
||||
allowRunCoalescing: wakeMustNotJoinActiveRun(input.activeExecutionRun.contextSnapshot, input.contextSnapshot, interactionReceipt)
|
||||
? false : input.allowRunCoalescing,
|
||||
sameDurableActor,
|
||||
isSameExecutionAgent,
|
||||
@@ -893,6 +892,7 @@ export function createAdmitWakeBehindIssueExecution(deps: {
|
||||
companyId: input.companyId,
|
||||
agentId: input.agentId,
|
||||
issueId: input.issueId,
|
||||
chatConversationId: readWakeChatConversationId(input.contextSnapshot),
|
||||
...(input.durableReceipt
|
||||
? {
|
||||
durableActor: {
|
||||
|
||||
@@ -4,8 +4,8 @@
|
||||
// shape; this file holds only the small pieces that are specific to
|
||||
// building the release half's promoted-run snapshot.
|
||||
|
||||
import { extractWakeCommentIds, WAKE_COMMENT_IDS_KEY } from "../../run-dispatch/index.js";
|
||||
import { readNonEmptyString } from "./values.js";
|
||||
import { extractWakeCommentIds, WAKE_COMMENT_IDS_KEY, wakeChatConversationsConflict } from "../../run-dispatch/index.js";
|
||||
import { parseObject, readNonEmptyString } from "./values.js";
|
||||
|
||||
const PAPERCLIP_WAKE_PAYLOAD_KEY = "paperclipWake";
|
||||
const PAPERCLIP_WAKE_COMMENT_KEY = "paperclipWakeComment";
|
||||
@@ -69,6 +69,22 @@ export function hasInteractionContinuationWakeContext(contextSnapshot: Record<st
|
||||
return INTERACTION_CONTINUATION_CONTEXT_KEYS.some((key) => readNonEmptyString(contextSnapshot[key]));
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether a wake must stay out of the active issue run. Each resolved card has
|
||||
* its own immutable decision and continuation, so it never merges with
|
||||
* comments or another card, in either arrival order. A reviewed chat run
|
||||
* attests exactly one conversation, so a reply from another conversation on
|
||||
* the same issue queues as its own wake.
|
||||
*/
|
||||
export function wakeMustNotJoinActiveRun(
|
||||
activeRunContext: unknown,
|
||||
wakeContext: Record<string, unknown>,
|
||||
interactionReceipt: boolean,
|
||||
): boolean {
|
||||
return interactionReceipt || Boolean(parseObject(activeRunContext).interactionId) ||
|
||||
wakeChatConversationsConflict(activeRunContext, wakeContext);
|
||||
}
|
||||
|
||||
function isInteractionResolutionWakePayload(payload: Record<string, unknown> | null | undefined): boolean {
|
||||
return readNonEmptyString(payload?.mutation) === "interaction";
|
||||
}
|
||||
|
||||
@@ -79,6 +79,7 @@ import {
|
||||
NativeChatReviewPresentationContentionError,
|
||||
} from "../native-runtime/native-chat-review-presentation.js";
|
||||
import { isExternalChatWaitAuthorizationContention } from "../native-runtime/chat-attachment-reuse.js";
|
||||
import { CHAT_CONVERSATION_ID_KEY } from "../../modules/run-dispatch/index.js";
|
||||
import { projectSafeChatPublication } from "../chat-publication-projection.js";
|
||||
|
||||
import {
|
||||
@@ -1598,6 +1599,7 @@ export function createChatFailedRunRetry(db: Db, options: ChatFailedRunRetryOpti
|
||||
} : {}),
|
||||
taskKey: source.taskKey,
|
||||
source: `chat:${source.provider}`,
|
||||
[CHAT_CONVERSATION_ID_KEY]: source.conversationId,
|
||||
wakeCommentId: source.commentIds.at(-1)!,
|
||||
wakeCommentIds: source.commentIds,
|
||||
retryOfRunId: source.failedRunId,
|
||||
|
||||
@@ -362,6 +362,7 @@ export function createInboundWakeups(db: Db, options: WakeupOptions) {
|
||||
requestedByActorId: request.requestedByActorId,
|
||||
taskKey: context.issue.identifier,
|
||||
wakeCommentId: request.commentId,
|
||||
chatConversationId: context.conversation.id,
|
||||
attachmentOmissionReasons: claimed.payload.attachmentOmissionReasons as
|
||||
| Record<string, number>
|
||||
| undefined,
|
||||
|
||||
@@ -147,7 +147,7 @@ import {
|
||||
WORKSPACE_WORKTREE_REQUIRES_PROJECT_REMEDIATION,
|
||||
} from "../execution-workspace-policy.js";
|
||||
import { readContinuationAttempt } from "../recovery/index.js";
|
||||
import { allowsIssueInteractionWake } from "../../modules/run-dispatch/index.js";
|
||||
import { allowsIssueInteractionWake, wakeChatConversationsConflict } from "../../modules/run-dispatch/index.js";
|
||||
import { withAgentStartLock } from "../agent-start-lock.js";
|
||||
import {
|
||||
evaluateAgentInvokability,
|
||||
@@ -4414,10 +4414,12 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
|
||||
? null
|
||||
: (sameScopeRunningRun ?? null)));
|
||||
|
||||
const coalescedTargetRun = filterZombieCoalesceTarget(
|
||||
rawCoalescedTarget,
|
||||
liveRunExecutions,
|
||||
);
|
||||
// A reviewed chat run attests one conversation, so never fold another
|
||||
// conversation's reply into it; that reply queues its own run instead.
|
||||
const coalescedTargetRun = rawCoalescedTarget &&
|
||||
wakeChatConversationsConflict(rawCoalescedTarget.contextSnapshot, enrichedContextSnapshot)
|
||||
? null
|
||||
: filterZombieCoalesceTarget(rawCoalescedTarget, liveRunExecutions);
|
||||
|
||||
if (coalescedTargetRun) {
|
||||
const mergedContextSnapshot = mergeCoalescedContextSnapshot(
|
||||
|
||||
@@ -37,6 +37,9 @@ export function queueIssueAssignmentWakeup(input: {
|
||||
/** Closed, server-derived omission counts for provider attachments on the
|
||||
* exact wake comment. These are prompt diagnostics, never authorization. */
|
||||
attachmentOmissionReasons?: Record<string, number> | null;
|
||||
/** External chat conversation the wake answers. The wake queue keeps each
|
||||
* conversation in its own batch, because a reviewed chat run attests one. */
|
||||
chatConversationId?: string | null;
|
||||
rethrowOnError?: boolean;
|
||||
durableChatRequest?: DurableChatWakeupRequest;
|
||||
}) {
|
||||
@@ -63,6 +66,7 @@ export function queueIssueAssignmentWakeup(input: {
|
||||
source: input.contextSource,
|
||||
...(input.taskKey ? { taskKey: input.taskKey } : {}),
|
||||
...(input.wakeCommentId ? { wakeCommentId: input.wakeCommentId } : {}),
|
||||
...(input.chatConversationId ? { chatConversationId: input.chatConversationId } : {}),
|
||||
...(input.wakeCommentId && input.attachmentOmissionReasons
|
||||
? {
|
||||
externalAttachmentOmissions: [
|
||||
|
||||
Reference in new issue
Block a user