mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-01 15:35:12 +01:00
fix(chat): recover committed native Agent Chat replies (#18465)
## Thinking Path > - Paperclip manages agent work and keeps its results available to users. > - Agent Chat needs a visible reply before the conversation can wait for more input. > - Native result commitment and reply presentation are separate obligations. > - A completed conversation can retain a valid final message without a saved reply. > - Existing reply recovery excludes these completed conversation turns. > - This pull request repairs that persistence gap through the existing recovery worker. ## Linked Issues or Issue Description **What happened?** A native Agent Chat turn completed successfully and saved its workspace. Its accepted result and final assistant message were durable, but no reply appeared in the conversation. The chat stayed active. The deployment hiring journey in #18351 then reached its deadline while waiting for the first reply. **Expected behavior** Recover the exact accepted turn's final reply once, then use normal conversation settlement. Keep later user input active. Do not run the agent again or replace a selected or deleted reply. **Steps to reproduce** 1. Create an active conversation with one admitted native turn. 2. Commit an accepted DONE result and a final-channel assistant message for that exact run, session and turn. 3. Finish workspace finalization before live heartbeat reply presentation. Leave no saved reply. 4. Run normal response recovery. Before this fix it finds no conversation candidate and leaves the chat active. **Paperclip version or commit** Reproduced with actual services at frozen master `8bcc5a01edd5d87540401521e6f30d3df43e1848`. The production predicate also exists in the deployed revision that exposed the failure. Related work: #18351. The broader receipt and recovery work in #18118 and #18409 does not touch these source files. ## What Changed - Discover unanswered committed Agent Chat turns through the existing native response presentation worker. - Recheck company, agent, responsible user, current execution and conversation generation under the normal transaction locks. - Require the accepted result digest and an exact final message from the same native run, session and accepted turn. Keep external publication authority separate. - Commit the reply and presentation receipt before normal conversation settlement. Later input stays active. Recovery does not repeat agent work. - Preserve skip decisions and selected or deleted replies. Add database controls for foreign identity, stale sessions, partial commitment, duplicates and ordinary internal task behavior. - Document the recovery boundary in the developer guide. ## Verification - Real-service RED/GREEN uses all 418 canonical migrations in isolated PGlite and the exact new regression fixture. Before the fix: zero candidates, no reply, recovery returns false and the chat stays active. After the fix: one candidate, one exact reply, recovery returns true and the chat waits. A repeat returns false and creates no second reply. - All 20 canonical regression cases pass locally through a fixture-only PGlite bridge. Only local database provisioning and `createDb` selection differ from the normal embedded PostgreSQL fixture. Production services, SQL queries, transactions, protocol validation and assertions remain real. Linux embedded PostgreSQL CI is required before merge. - Focused current-source no-emit typing passes for both production modules and the new regression file, with normal Express augmentation and source aliases. The broader local presentation/ownership and conversation-settlement PGlite bridge run was interrupted before it reported results; it is not counted as a pass. No repository-wide local build is claimed from borrowed workspace dependencies. Clean-build CI and canonical PostgreSQL consumer tests remain required. - `git diff --check` and the scoped source/test secret scan pass. No captured run data, credentials or private target identities are committed. - No live journey retry, deployment, configuration change or test-plan admission was performed. The original failed journey remains failed; this production fix still needs deployed qualification through the unchanged journey. - Current-head [canonical Linux CI](https://github.com/paperclipai/paperclip/actions/runs/41207018345) passed all 57 terminal checks with no failure or pending check. The new 20-case database regression actually passed in [server shard 8](https://github.com/paperclipai/paperclip/actions/runs/41207018345/job/124555006090), at 13:59:25 UTC (2,643 ms). Typecheck, build and the existing consumer lanes passed. Fresh [Greptile review](https://github.com/paperclipai/paperclip/pull/18465#issuecomment-6544208002) on exact `f32a432e10b377ae6277d6f728995d5fedcf9d7b` is 5/5; the review-thread inventory is empty with no pagination. No live deployment or journey pass is claimed by these checks. ## Risks Eligible active conversations can gain a previously missing reply. Recovery accepts only the exact saved final message under its accepted result and current conversation identity. It does not run a model, repeat hiring or change a prior test grade. Reply persistence precedes waiting-state settlement; existing conversation recovery retains ownership of a later interrupted settlement. No schema or public API changes are included. ## Model Used OpenAI Codex, GPT-6 family. The session does not expose a more specific backend model ID or fixed context window. Used reasoning, repository tools, TypeScript and real database test execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge
This commit is contained in:
1 parent
baf651c8d0
commit
12b0cc6304
4 files changed
+197
-10
No files matched your search
@@ -2660,6 +2660,13 @@ An upgrade readiness report contains preliminary native evidence, not permission
|
||||
to remove a deployment volume. Keep the checkpoint pool bounded and measure all
|
||||
pools when qualifying database cost; the current size of two is a default.
|
||||
|
||||
Committed native Agent Chat turns recover their unanswered final reply through
|
||||
the existing response presentation worker. Recovery verifies the accepted result
|
||||
and final turn, agent, responsible user, native session and conversation generation.
|
||||
It stores the exact final message before the normal conversation settlement moves
|
||||
the chat to waiting. Later user input remains active. Recovery does not run the
|
||||
agent again, replace a selected reply or restore a deleted reply.
|
||||
|
||||
For retained Codex event tails, keep source drain/accounting and target admission
|
||||
separate. `native-staged-attachment-recovery.ts` validates private source/target
|
||||
header evidence; `native-recovery-usage-delivery.ts` binds the source's accounting
|
||||
|
||||
@@ -0,0 +1,149 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import { agents, authUsers, companies, completionContracts, createDb, heartbeatRunEvents, heartbeatRuns, issueComments, issues,
|
||||
nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
|
||||
import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js";
|
||||
import { NativeRunCoordinatorStore } from "./native-run-coordinator-store.js";
|
||||
import { finalizeNativeRun, recoverCommittedNativeChatResponse } from "./native-run-finalizer.js";
|
||||
import { nativeResponsePresentationWork } from "./native-response-presentation-work.js";
|
||||
|
||||
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_AGENT_CHAT_RESPONSE_TEST_DATABASE_URL");
|
||||
const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
|
||||
describe.skipIf(!support.supported)("committed Agent Chat reply recovery", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let db: Db;
|
||||
beforeAll(async () => {
|
||||
if (!external) database = await startEmbeddedPostgresTestDatabase("agent-chat-response-");
|
||||
db = createDb(external ?? database!.connectionString);
|
||||
}, 60_000);
|
||||
afterAll(async () => { await db?.$client.end(); await database?.cleanup(); });
|
||||
|
||||
async function fixture(conversation = true) {
|
||||
const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID();
|
||||
const sessionId = randomUUID(), runnerId = randomUUID(), userId = `fixture-${randomUUID()}`, wakeId = randomUUID();
|
||||
await db.insert(authUsers).values({ id: userId, name: "Conversation owner", email: `${userId}@example.test`,
|
||||
createdAt: new Date("2020-01-01T00:00:00Z"), updatedAt: new Date("2020-01-01T00:00:00Z") });
|
||||
await db.insert(companies).values({ id: companyId, name: "Reply recovery fixture", issuePrefix: companyId.slice(0, 8) });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Conversation agent", adapterType: "paperclip_runner" });
|
||||
await db.insert(issues).values({ id: issueId, companyId, title: "Conversation reply", status: "in_progress", assigneeAgentId: agentId,
|
||||
...(conversation ? { conversationAgentId: agentId, conversationUserId: userId, conversationState: "active", conversationSessionGeneration: 0 } : {}) });
|
||||
await db.insert(issueComments).values({ id: wakeId, companyId, issueId, authorUserId: userId, body: "Hire one teammate and tell me when ready.",
|
||||
createdAt: new Date("2020-01-01T00:00:00Z") });
|
||||
const [contract] = await db.insert(completionContracts).values({ companyId, issueId, revision: 1,
|
||||
schemaVersion: "paperclip.completion-contract.v1", policyVersion: "test", risk: "standard", completionAuthority: "server_arbiter",
|
||||
incompleteCriteriaPolicy: "preserve_non_terminal", contractJson: { revision: "test", objective: "Answer the user", criteria: [] },
|
||||
canonicalSha256: "fixture-conversation-contract", createdByActorType: "system", createdByActorId: "test" }).returning();
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: issueId, nativeSessionId: sessionId,
|
||||
runnerInstanceId: runnerId, completionContractId: contract.id, completionContractSha256: contract.canonicalSha256,
|
||||
status: "running", runtimeMode: "native", responsibleUserId: userId,
|
||||
contextSnapshot: { issueId, source: "issue.comment", wakeCommentId: wakeId, ...(conversation ? { conversationSessionGeneration: 0 } : {}) } });
|
||||
await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId));
|
||||
const port = new NativeRunCoordinatorStore(db, { companyId, issueId, runId, agentId, normalizedSessionId: sessionId,
|
||||
runnerSourceInstanceId: runnerId, completionContractId: contract.id, completionContractSha256: contract.canonicalSha256,
|
||||
completionContractRevision: "test", completionContractCriterionIds: [] });
|
||||
const reply = "The teammate is hired and ready for your next instruction.";
|
||||
const prpEvent = { runId, normalizedSessionId: sessionId, turnId: "hire-turn", sourceEventId: "fixture-final-answer",
|
||||
payload: { kind: "agentMessage", channel: "final", providerPhase: "final", text: reply } };
|
||||
await db.insert(heartbeatRunEvents).values({ companyId, runId, agentId, seq: 170, eventType: "item.completed", payload: { prpEvent } });
|
||||
await port.completeRun({ result: { schema: "paperclip.run_result.v1", reportedWorkDisposition: "done",
|
||||
summary: "A completed hire, with its final message preserved separately.", completionClaim: { contractRevision: "test",
|
||||
objectiveSatisfied: true, criteria: [], remainingWork: [] }, evidence: [], verification: [], attentionRequests: [], artifacts: [] },
|
||||
terminal: { schema: "paperclip.prp.terminal.v1", runTerminalState: "succeeded", turnTerminalState: "completed", reportedWorkDisposition: "done" }, turnId: "hire-turn" });
|
||||
await finalizeNativeRun({ db, runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: false });
|
||||
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date("2020-01-01T00:02:00Z") }).where(eq(heartbeatRuns.id, runId));
|
||||
expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)))[0].phase).toBe("committed");
|
||||
const input = { companyId, issueId, runId };
|
||||
return { ...input, input, agentId, userId, wakeId, prpEvent, reply };
|
||||
}
|
||||
const comments = (runId: string) => db.select().from(issueComments).where(eq(issueComments.createdByRunId, runId));
|
||||
const issue = async (issueId: string) => (await db.select().from(issues).where(eq(issues.id, issueId)))[0];
|
||||
|
||||
it("finds a committed final Agent Chat turn, persists the exact reply once and only then settles waiting", async () => {
|
||||
const f = await fixture();
|
||||
expect(await comments(f.runId)).toEqual([]);
|
||||
expect(await issue(f.issueId)).toMatchObject({ status: "in_progress", conversationState: "active" });
|
||||
const work = nativeResponsePresentationWork(db), entries = await work.load(f.input);
|
||||
expect(entries).toEqual([expect.objectContaining({ id: f.runId })]);
|
||||
await expect(work.check(entries[0])).resolves.toBeNull();
|
||||
expect(await comments(f.runId)).toEqual([expect.objectContaining({ body: f.reply, authorAgentId: f.agentId, createdByRunId: f.runId })]);
|
||||
expect(await issue(f.issueId)).toMatchObject({ status: "in_review", conversationState: "waiting" });
|
||||
expect(await work.load(f.input)).toEqual([]);
|
||||
await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(false);
|
||||
expect(await comments(f.runId)).toHaveLength(1);
|
||||
});
|
||||
it("keeps a later user message actionable after recovering the earlier exact reply", async () => {
|
||||
const f = await fixture();
|
||||
await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, authorUserId: f.userId, body: "Another instruction",
|
||||
createdAt: new Date("2020-01-01T00:01:00Z") });
|
||||
await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(true);
|
||||
expect(await comments(f.runId)).toHaveLength(1);
|
||||
expect(await issue(f.issueId)).toMatchObject({ status: "in_progress", conversationState: "active" });
|
||||
});
|
||||
it("does not duplicate a reply under concurrent recovery", async () => {
|
||||
const f = await fixture();
|
||||
expect((await Promise.all([recoverCommittedNativeChatResponse(db, f.input), recoverCommittedNativeChatResponse(db, f.input)])).sort())
|
||||
.toEqual([false, true]);
|
||||
expect(await comments(f.runId)).toHaveLength(1);
|
||||
});
|
||||
it.each(["skip", "generation", "agent", "user", "execution", "waiting", "turn", "event-run", "session", "channel", "kind", "digest", "partial"])(
|
||||
"keeps %s mismatches unchanged instead of publishing an unauthorized or incomplete reply", async fault => {
|
||||
const f = await fixture();
|
||||
if (["skip", "generation"].includes(fault)) await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issueId,
|
||||
source: "issue.comment", wakeCommentId: f.wakeId, conversationSessionGeneration: fault === "generation" ? 1 : 0,
|
||||
...(fault === "skip" ? { skipIssueComment: true } : {}) } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
if (fault === "user") await db.update(heartbeatRuns).set({ responsibleUserId: "another-user" }).where(eq(heartbeatRuns.id, f.runId));
|
||||
if (fault === "agent") {
|
||||
const other = randomUUID(); await db.insert(agents).values({ id: other, companyId: f.companyId, name: "Other agent" });
|
||||
await db.update(issues).set({ conversationAgentId: other, assigneeAgentId: other }).where(eq(issues.id, f.issueId));
|
||||
}
|
||||
if (fault === "execution") {
|
||||
const other = randomUUID(); await db.insert(heartbeatRuns).values({ id: other, companyId: f.companyId, agentId: f.agentId });
|
||||
await db.update(issues).set({ executionRunId: other }).where(eq(issues.id, f.issueId));
|
||||
}
|
||||
if (fault === "waiting") await db.update(issues).set({ executionRunId: null, status: "in_review", conversationState: "waiting" }).where(eq(issues.id, f.issueId));
|
||||
if (["turn", "event-run", "session", "channel", "kind"].includes(fault)) {
|
||||
const event = structuredClone(f.prpEvent);
|
||||
if (fault === "turn") event.turnId = "foreign-turn";
|
||||
if (fault === "event-run") event.runId = randomUUID();
|
||||
if (fault === "session") event.normalizedSessionId = randomUUID();
|
||||
if (fault === "channel") event.payload.channel = "analysis";
|
||||
if (fault === "kind") event.payload.kind = "reasoning";
|
||||
await db.update(heartbeatRunEvents).set({ payload: { prpEvent: event } }).where(and(eq(heartbeatRunEvents.runId, f.runId), eq(heartbeatRunEvents.seq, 170)));
|
||||
}
|
||||
if (fault === "digest") await db.update(nativeRunResults).set({ canonicalSha256: "invalid-digest" }).where(eq(nativeRunResults.runId, f.runId));
|
||||
if (fault === "partial") await db.update(nativeRunFinalizations).set({ phase: "copying_back" }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
const before = await issue(f.issueId);
|
||||
await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(false);
|
||||
expect(await comments(f.runId)).toEqual([]); expect(await issue(f.issueId)).toEqual(before);
|
||||
},
|
||||
);
|
||||
it.each([false, true])("preserves a selected reply without replacing or resurrecting it (deleted=%s)", async deleted => {
|
||||
const f = await fixture();
|
||||
const [saved] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, authorAgentId: f.agentId,
|
||||
createdByRunId: f.runId, body: "Already selected", ...(deleted ? { deletedAt: new Date("2020-01-01T00:03:00Z") } : {}) }).returning();
|
||||
await db.update(heartbeatRuns).set({ resultJson: { presentationDecision: { commentId: saved.id, commentAction: "create" } } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(false);
|
||||
expect(await comments(f.runId)).toEqual([saved]);
|
||||
});
|
||||
it("never crosses company scope and keeps ordinary internal task response recovery intact", async () => {
|
||||
const f = await fixture(false);
|
||||
await expect(recoverCommittedNativeChatResponse(db, { ...f.input, companyId: randomUUID() })).resolves.toBe(false);
|
||||
const work = nativeResponsePresentationWork(db), entries = await work.load(f.input);
|
||||
expect(entries).toHaveLength(1); await expect(work.check(entries[0])).resolves.toBeNull();
|
||||
expect(await comments(f.runId)).toEqual([expect.objectContaining({ body: f.reply })]);
|
||||
expect((await issue(f.issueId)).conversationAgentId).toBeNull();
|
||||
});
|
||||
it("rolls back a partial reply write without changing the waiting state or consuming recovery", async () => {
|
||||
const f = await fixture(), abort = new Error("fixture-commit-aborted");
|
||||
const failing = new Proxy(db, { get(target, key) { return key === "transaction"
|
||||
? (callback: Parameters<Db["transaction"]>[0]) => db.transaction(async tx => { await callback(tx); throw abort; })
|
||||
: Reflect.get(target, key); } });
|
||||
await expect(recoverCommittedNativeChatResponse(failing, f.input)).rejects.toBe(abort);
|
||||
expect(await comments(f.runId)).toEqual([]);
|
||||
expect(await issue(f.issueId)).toMatchObject({ status: "in_progress", conversationState: "active" });
|
||||
await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(true);
|
||||
expect(await comments(f.runId)).toHaveLength(1);
|
||||
});
|
||||
});
|
||||
@@ -1,6 +1,6 @@
|
||||
import { QueryBuilder } from "drizzle-orm/pg-core";
|
||||
import { and, asc, eq, gt, inArray, isNotNull, isNull, notExists, or, sql } from "drizzle-orm";
|
||||
import { heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, statusDecisions, statusDecisionEffects } from "@paperclipai/db";
|
||||
import { and, asc, eq, exists, gt, inArray, isNotNull, isNull, notExists, or, sql } from "drizzle-orm";
|
||||
import { heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, statusDecisions, statusDecisionEffects } from "@paperclipai/db";
|
||||
import type { NativeResponsePresentationScope } from "../native-response-presentation-notifications.js";
|
||||
|
||||
export function nativeResponsePresentationCandidates(scope?: NativeResponsePresentationScope, after?: string, runId?: string) {
|
||||
@@ -30,7 +30,24 @@ export function nativeResponsePresentationCandidates(scope?: NativeResponsePrese
|
||||
sql`${nativeRunResults.resultJson} #>> '{terminal,turnTerminalState}' = 'completed'`,
|
||||
or(
|
||||
and(eq(heartbeatRuns.status, "succeeded"),
|
||||
sql`(coalesce(${issues.conversationAgentId}::text, '') = '' or coalesce(${issues.conversationUserId}, '') = '')`,
|
||||
or(sql`(coalesce(${issues.conversationAgentId}::text, '') = '' or coalesce(${issues.conversationUserId}, '') = '')`,
|
||||
and(eq(statusDecisions.reasonCode, "conversation_turn_finished"),
|
||||
eq(issues.conversationAgentId, heartbeatRuns.agentId), eq(issues.conversationUserId, heartbeatRuns.responsibleUserId),
|
||||
eq(issues.status, "in_progress"), eq(issues.conversationState, "active"),
|
||||
or(isNull(issues.executionRunId), eq(issues.executionRunId, heartbeatRuns.id)),
|
||||
sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issues.id}::text`,
|
||||
sql`jsonb_typeof(${heartbeatRuns.contextSnapshot}->'conversationSessionGeneration') = 'number'`,
|
||||
sql`${heartbeatRuns.contextSnapshot}->>'conversationSessionGeneration' = ${issues.conversationSessionGeneration}::text`,
|
||||
exists(db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and(
|
||||
eq(heartbeatRunEvents.companyId, heartbeatRuns.companyId), eq(heartbeatRunEvents.runId, heartbeatRuns.id),
|
||||
eq(heartbeatRunEvents.agentId, heartbeatRuns.agentId), eq(heartbeatRunEvents.eventType, "item.completed"),
|
||||
sql`${heartbeatRunEvents.payload} #>> '{prpEvent,runId}' = ${heartbeatRuns.id}::text`,
|
||||
sql`${heartbeatRunEvents.payload} #>> '{prpEvent,normalizedSessionId}' = ${heartbeatRuns.nativeSessionId}::text`,
|
||||
sql`${heartbeatRunEvents.payload} #>> '{prpEvent,turnId}' = ${nativeRunResults.turnId}`,
|
||||
sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,kind}' = 'agentMessage'`,
|
||||
sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,channel}' = 'final'`,
|
||||
sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,text}' ~ '[^[:space:]]'`,
|
||||
))))),
|
||||
sql`${nativeRunResults.resultJson} #>> '{result,reportedWorkDisposition}' = 'done'`,
|
||||
sql`${nativeRunResults.resultJson} #>> '{terminal,reportedWorkDisposition}' = 'done'`),
|
||||
and(sql`${nativeRunResults.resultJson} #>> '{result,reportedWorkDisposition}' = 'yielded'`,
|
||||
|
||||
@@ -29,7 +29,7 @@ import { readPersistedNativeFailureEvidence } from "./native-provider-failure-ev
|
||||
import { dismissAutomaticCompletionReviews } from "./automatic-completion-reviews.js";
|
||||
import { findSupersededExternalReviewRequests } from "./superseded-external-review-requests.js";
|
||||
import { getNativeReviewAssignment, readNativeReviewAssignmentContext } from "./native-review-participant.js";
|
||||
import { conversationNativeDecision, isConversation } from "../agent-conversations.js";
|
||||
import { conversationNativeDecision, isConversation, settleConversationTurnForRun } from "../agent-conversations.js";
|
||||
import { issueTreeControlService } from "../issue-tree-control.js";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { preserveNativeWorkspaceExportLease } from "./native-workspace-export-resume.js";
|
||||
@@ -747,6 +747,7 @@ async function recoverCommittedNativeChatResponseInScope(
|
||||
): Promise<boolean> {
|
||||
let agentId: string | null = null;
|
||||
let materialized = false;
|
||||
let conversationReply = false;
|
||||
const entered = beginRunPhase("response.transaction.entry");
|
||||
try {
|
||||
materialized = await measureRunPhase("response.transaction", () => db.transaction(async (transaction) => {
|
||||
@@ -789,6 +790,7 @@ async function recoverCommittedNativeChatResponseInScope(
|
||||
id: heartbeatRuns.id,
|
||||
companyId: heartbeatRuns.companyId,
|
||||
agentId: heartbeatRuns.agentId,
|
||||
responsibleUserId: heartbeatRuns.responsibleUserId,
|
||||
nativeIssueId: heartbeatRuns.nativeIssueId,
|
||||
nativeSessionId: heartbeatRuns.nativeSessionId,
|
||||
completionContractId: heartbeatRuns.completionContractId,
|
||||
@@ -803,6 +805,8 @@ async function recoverCommittedNativeChatResponseInScope(
|
||||
contextSnapshot: sql<Record<string, unknown> | null>`case when ${heartbeatRuns.contextSnapshot} is null then null
|
||||
else jsonb_build_object(
|
||||
'skipIssueComment', ${heartbeatRuns.contextSnapshot}->'skipIssueComment',
|
||||
'issueId', ${heartbeatRuns.contextSnapshot}->'issueId',
|
||||
'conversationSessionGeneration', ${heartbeatRuns.contextSnapshot}->'conversationSessionGeneration',
|
||||
'source', ${heartbeatRuns.contextSnapshot}->'source',
|
||||
'externalChatContinuation', ${heartbeatRuns.contextSnapshot}->'externalChatContinuation',
|
||||
'paperclipWake', jsonb_build_object('externalInteractionContinuation',
|
||||
@@ -909,12 +913,16 @@ async function recoverCommittedNativeChatResponseInScope(
|
||||
agentId = run.agentId;
|
||||
return true;
|
||||
}
|
||||
// Copyback can be owned by reconciliation before the live heartbeat
|
||||
// reaches its presentation step. Recover a completed internal task's
|
||||
// exact final reply from the same accepted turn, without rerunning work
|
||||
// or granting external-chat publication authority.
|
||||
const conversation = isConversation(issue);
|
||||
const currentConversation = conversation && decision.reasonCode === "conversation_turn_finished"
|
||||
&& issue.conversationAgentId === run.agentId && issue.conversationUserId === run.responsibleUserId
|
||||
&& issue.status === "in_progress" && issue.conversationState === "active"
|
||||
&& (!issue.executionRunId || issue.executionRunId === run.id)
|
||||
&& run.contextSnapshot?.issueId === issue.id
|
||||
&& typeof run.contextSnapshot.conversationSessionGeneration === "number"
|
||||
&& run.contextSnapshot.conversationSessionGeneration === issue.conversationSessionGeneration;
|
||||
if (
|
||||
!isConversation(issue) &&
|
||||
(!conversation || currentConversation) &&
|
||||
run.status === "succeeded" &&
|
||||
record(run.contextSnapshot).skipIssueComment !== true &&
|
||||
!isExternalChatPresentationContext(run.contextSnapshot) &&
|
||||
@@ -929,6 +937,9 @@ async function recoverCommittedNativeChatResponseInScope(
|
||||
eq(heartbeatRunEvents.companyId, input.companyId),
|
||||
eq(heartbeatRunEvents.runId, run.id),
|
||||
eq(heartbeatRunEvents.eventType, "item.completed"),
|
||||
conversation ? and(eq(heartbeatRunEvents.agentId, run.agentId),
|
||||
sql`${heartbeatRunEvents.payload}->'prpEvent'->>'runId' = ${run.id}`,
|
||||
sql`${heartbeatRunEvents.payload}->'prpEvent'->>'normalizedSessionId' = ${run.nativeSessionId}`) : undefined,
|
||||
sql`${heartbeatRunEvents.payload}->'prpEvent'->>'turnId' = ${accepted.turnId}`,
|
||||
sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,kind}' = 'agentMessage'`,
|
||||
sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,channel}' = 'final'`,
|
||||
@@ -963,6 +974,7 @@ async function recoverCommittedNativeChatResponseInScope(
|
||||
})}::jsonb`, updatedAt: new Date(),
|
||||
}).where(eq(heartbeatRuns.id, run.id));
|
||||
agentId = run.agentId;
|
||||
conversationReply = conversation;
|
||||
return true;
|
||||
}
|
||||
if (
|
||||
@@ -1039,12 +1051,14 @@ async function recoverCommittedNativeChatResponseInScope(
|
||||
} finally {
|
||||
entered();
|
||||
}
|
||||
if (materialized && agentId)
|
||||
if (materialized && agentId) {
|
||||
if (conversationReply) await settleConversationTurnForRun(db, input.companyId, input.runId);
|
||||
publishChatPublicationCommitSignal({
|
||||
...input,
|
||||
agentId,
|
||||
eventType: "run.presentation.resolved",
|
||||
});
|
||||
}
|
||||
return materialized;
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user