From d711fe390c58263dee3ca7f84df025c83638a448 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 21 Aug 2026 10:53:45 -0500 Subject: [PATCH] fix(runner): stream progress and final replies Publish runner-authored comments to live issue threads, keep progress distinct from completion, and make follow-up comments authoritative for native completion contracts. Co-Authored-By: Paperclip --- .../heartbeat-context-summary.test.ts | 20 ++++++ .../__tests__/heartbeat-run-summary.test.ts | 24 +++++++ server/src/services/heartbeat-run-summary.ts | 25 ++++++++ server/src/services/heartbeat.ts | 64 ++++++++++++++++--- .../completion-contracts.test.ts | 26 ++++++++ .../native-runtime/completion-contracts.ts | 25 +++++--- .../paperclip-runner-tool-authority.test.ts | 19 +++++- .../paperclip-runner-tool-authority.ts | 36 +++++++++-- 8 files changed, 215 insertions(+), 24 deletions(-) create mode 100644 server/src/services/native-runtime/completion-contracts.test.ts diff --git a/server/src/__tests__/heartbeat-context-summary.test.ts b/server/src/__tests__/heartbeat-context-summary.test.ts index 072ae42e319..05f4057175c 100644 --- a/server/src/__tests__/heartbeat-context-summary.test.ts +++ b/server/src/__tests__/heartbeat-context-summary.test.ts @@ -133,6 +133,26 @@ describe("buildPaperclipTaskMarkdown", () => { expect(compact).toContain("Please also update the changelog."); }); + it("makes the latest wake comment the immediate follow-up request", () => { + const commentWake = buildPaperclipTaskMarkdown({ + issue: { + id: "issue-follow-up", + identifier: "PAP-418", + title: "Original task", + workMode: "standard", + description: "Reply with the original answer.", + }, + wakeComment: { + id: "comment-follow-up", + body: "Reply with the new answer instead.", + }, + }); + + expect(commentWake).toContain("The latest wake comment is the immediate request for this run."); + expect(commentWake).toContain("Do not repeat an earlier requested output from the issue description"); + expect(commentWake).toContain("Reply with the new answer instead."); + }); + it("prefers ordinary comment planning guidance over stale accepted confirmation state", () => { const commentWake = buildPaperclipTaskMarkdown({ issue: { diff --git a/server/src/__tests__/heartbeat-run-summary.test.ts b/server/src/__tests__/heartbeat-run-summary.test.ts index 74051ab4695..84415052029 100644 --- a/server/src/__tests__/heartbeat-run-summary.test.ts +++ b/server/src/__tests__/heartbeat-run-summary.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest"; import { summarizeHeartbeatRunResultJson, buildHeartbeatRunIssueComment, + findHeartbeatRunCompletionComment, mergeHeartbeatRunResultJson, } from "../services/heartbeat-run-summary.js"; @@ -118,6 +119,29 @@ describe("buildHeartbeatRunIssueComment", () => { }); }); +describe("findHeartbeatRunCompletionComment", () => { + it("does not let semantic progress satisfy the final comment", () => { + const progress = { id: "progress-comment" }; + const final = { id: "final-comment" }; + const resultJson = { + semanticToolReceipts: { + progress: { + operationId: "report_progress", + result: { commentId: progress.id }, + }, + }, + }; + + expect(findHeartbeatRunCompletionComment([progress], resultJson)).toBeNull(); + expect(findHeartbeatRunCompletionComment([final, progress], resultJson)).toEqual(final); + }); + + it("preserves the fallback-only behavior for ordinary agent comments", () => { + const comment = { id: "manual-comment" }; + expect(findHeartbeatRunCompletionComment([comment], { summary: "done" })).toEqual(comment); + }); +}); + describe("mergeHeartbeatRunResultJson", () => { it("adds adapter summaries into stored result json for comment posting", () => { const merged = mergeHeartbeatRunResultJson( diff --git a/server/src/services/heartbeat-run-summary.ts b/server/src/services/heartbeat-run-summary.ts index cdd1bdeca92..73252a29ac4 100644 --- a/server/src/services/heartbeat-run-summary.ts +++ b/server/src/services/heartbeat-run-summary.ts @@ -126,3 +126,28 @@ export function buildHeartbeatRunIssueComment( return text; } + +export function findHeartbeatRunCompletionComment( + comments: T[], + resultJson: Record | null | undefined, +): T | null { + const receipts = resultJson?.semanticToolReceipts; + if (!receipts || typeof receipts !== "object" || Array.isArray(receipts)) { + return comments[0] ?? null; + } + + const progressCommentIds = new Set(); + for (const receipt of Object.values(receipts)) { + if (!receipt || typeof receipt !== "object" || Array.isArray(receipt)) continue; + const receiptRecord = receipt as Record; + if (receiptRecord.operationId !== "report_progress") continue; + const result = receiptRecord.result; + if (!result || typeof result !== "object" || Array.isArray(result)) continue; + const commentId = (result as Record).commentId; + if (typeof commentId === "string" && commentId.length > 0) { + progressCommentIds.add(commentId); + } + } + + return comments.find((comment) => !progressCommentIds.has(comment.id)) ?? null; +} diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 150efd2318b..f3ceaa3aced 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -124,6 +124,7 @@ import { secretService, type MissingRuntimeBinding } from "./secrets.js"; import { resolveDefaultAgentWorkspaceDir, resolveManagedProjectWorkspaceDir } from "../home-paths.js"; import { buildHeartbeatRunIssueComment, + findHeartbeatRunCompletionComment, HEARTBEAT_RUN_RESULT_OUTPUT_MAX_CHARS, HEARTBEAT_RUN_RESULT_SUMMARY_MAX_CHARS, HEARTBEAT_RUN_SAFE_RESULT_JSON_MAX_BYTES, @@ -6307,7 +6308,14 @@ export function buildPaperclipTaskMarkdown(input: { } } if (wakeComment?.body.trim()) { - lines.push("", "Latest wake comment:", fenceTaskText(wakeComment.body.trim())); + lines.push( + "", + "Follow-up directive:", + "The latest wake comment is the immediate request for this run. Address it directly. Do not repeat an earlier requested output from the issue description unless the latest comment asks you to.", + "", + "Latest wake comment:", + fenceTaskText(wakeComment.body.trim()), + ); } lines.push("", "Use this task context as the current assignment."); return lines.join("\n"); @@ -9880,8 +9888,13 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) .then((rows) => rows[0] ?? null); } - async function findRunIssueComment(runId: string, companyId: string, issueId: string) { - return db + async function findRunIssueComment( + runId: string, + companyId: string, + issueId: string, + resultJson?: Record | null, + ) { + const comments = await db .select({ id: issueComments.id, }) @@ -9893,9 +9906,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) eq(issueComments.createdByRunId, runId), ), ) - .orderBy(desc(issueComments.createdAt), desc(issueComments.id)) - .limit(1) - .then((rows) => rows[0] ?? null); + .orderBy(desc(issueComments.createdAt), desc(issueComments.id)); + return findHeartbeatRunCompletionComment(comments, resultJson); } async function refreshContinuationSummaryForRun( @@ -10109,7 +10121,12 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) return { outcome: "not_applicable" as const, queuedRun: null }; } - const postedComment = await findRunIssueComment(run.id, run.companyId, issueId); + const postedComment = await findRunIssueComment( + run.id, + run.companyId, + issueId, + parseObject(run.resultJson), + ); if (postedComment) { await patchRunIssueCommentStatus(run.id, { issueCommentStatus: "satisfied", @@ -15866,6 +15883,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) companyId: agent.companyId, issue: issueRef, actorId: agent.id, + immediateRequest: safeWakeCommentContext?.body ?? null, }); nativeRunnerInstanceId = run.runnerInstanceId ?? randomUUID(); const nativeSessionId = run.nativeSessionId ?? randomUUID(); @@ -16589,11 +16607,39 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const skipRunIssueComment = parseObject(livenessRun.contextSnapshot).skipIssueComment === true; if (issueId && outcome === "succeeded" && !skipRunIssueComment) { try { - const existingRunComment = await findRunIssueComment(livenessRun.id, livenessRun.companyId, issueId); + const existingRunComment = await findRunIssueComment( + livenessRun.id, + livenessRun.companyId, + issueId, + persistedResultJson, + ); if (!existingRunComment) { const issueComment = buildHeartbeatRunIssueComment(persistedResultJson); if (issueComment) { - await issuesSvc.addComment(issueId, issueComment, { agentId: agent.id, runId: livenessRun.id }); + const comment = await issuesSvc.addComment( + issueId, + issueComment, + { agentId: agent.id, runId: livenessRun.id }, + ); + await logActivity(db, { + companyId: livenessRun.companyId, + actorType: "agent", + actorId: agent.id, + agentId: agent.id, + runId: livenessRun.id, + issueId, + action: "issue.comment_added", + entityType: "issue", + entityId: issueId, + details: { + commentId: comment.id, + bodySnippet: comment.body.slice(0, 120), + identifier: issueRef?.identifier ?? null, + issueTitle: issueRef?.title ?? null, + authorizationReason: "internal_agent_write", + source: "heartbeat_run_summary", + }, + }); } } } catch (err) { diff --git a/server/src/services/native-runtime/completion-contracts.test.ts b/server/src/services/native-runtime/completion-contracts.test.ts new file mode 100644 index 00000000000..5d31e819e2c --- /dev/null +++ b/server/src/services/native-runtime/completion-contracts.test.ts @@ -0,0 +1,26 @@ +import { describe, expect, it } from "vitest"; +import { buildNativeCompletionContract } from "./completion-contracts.js"; + +describe("buildNativeCompletionContract", () => { + it("uses the issue description for an assignment", () => { + expect(buildNativeCompletionContract({ + title: "Original task", + description: "Return the original result.", + })).toEqual({ + revision: "1", + objective: "Original task", + criteria: [{ id: "objective", requirement: "Return the original result." }], + }); + }); + + it("makes the latest comment authoritative for a follow-up run", () => { + expect(buildNativeCompletionContract( + { title: "Original task", description: "Return the original result." }, + " Return the follow-up result. ", + )).toEqual({ + revision: "1", + objective: "Respond to the latest comment on Original task", + criteria: [{ id: "objective", requirement: "Return the follow-up result." }], + }); + }); +}); diff --git a/server/src/services/native-runtime/completion-contracts.ts b/server/src/services/native-runtime/completion-contracts.ts index bda276b69bc..03a9e03114b 100644 --- a/server/src/services/native-runtime/completion-contracts.ts +++ b/server/src/services/native-runtime/completion-contracts.ts @@ -7,20 +7,29 @@ import { nativeSha256 } from "./canonical.js"; export const NATIVE_COMPLETION_CONTRACT_SCHEMA = "paperclip.completion-contract.v1"; export const NATIVE_COMPLETION_POLICY_VERSION = "phase6-v2"; +export function buildNativeCompletionContract( + issue: { title: string; description: string | null }, + immediateRequest?: string | null, +): StrictCompletionContractInput { + const followUp = immediateRequest?.trim(); + return { + revision: "1", + objective: followUp ? `Respond to the latest comment on ${issue.title}` : issue.title, + criteria: [{ + id: "objective", + requirement: followUp || issue.description?.trim() || `Complete: ${issue.title}`, + }], + }; +} + export async function ensureNativeCompletionContract(input: { db: Db; companyId: string; issue: { id: string; title: string; description: string | null }; actorId: string; + immediateRequest?: string | null; }) { - const contract: StrictCompletionContractInput = { - revision: "1", - objective: input.issue.title, - criteria: [{ - id: "objective", - requirement: input.issue.description?.trim() || `Complete: ${input.issue.title}`, - }], - }; + const contract = buildNativeCompletionContract(input.issue, input.immediateRequest); const canonicalSha256 = nativeSha256({ schemaVersion: NATIVE_COMPLETION_CONTRACT_SCHEMA, policyVersion: NATIVE_COMPLETION_POLICY_VERSION, diff --git a/server/src/services/native-runtime/paperclip-runner-tool-authority.test.ts b/server/src/services/native-runtime/paperclip-runner-tool-authority.test.ts index 647f33a797d..8837107b569 100644 --- a/server/src/services/native-runtime/paperclip-runner-tool-authority.test.ts +++ b/server/src/services/native-runtime/paperclip-runner-tool-authority.test.ts @@ -1,6 +1,6 @@ import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { eq } from "drizzle-orm"; -import { agents, companies, createDb, documents, heartbeatRuns, issueComments, issueThreadInteractions, issues } from "@paperclipai/db"; +import { activityLog, agents, companies, createDb, documents, heartbeatRuns, issueComments, issueThreadInteractions, issues } from "@paperclipai/db"; import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; import { PaperclipRunnerToolAuthority } from "./paperclip-runner-tool-authority.js"; @@ -78,6 +78,23 @@ describe("PaperclipRunnerToolAuthority", () => { expect(replay).toEqual(first); expect(await db.select().from(issueComments).where(eq(issueComments.issueId, issueId))) .toHaveLength(1); + const progressActivity = await db.select().from(activityLog).where(eq(activityLog.entityId, issueId)); + expect(progressActivity).toHaveLength(1); + expect(progressActivity[0]).toMatchObject({ + action: "issue.comment_added", + actorType: "agent", + actorId: agentId, + agentId, + runId, + entityType: "issue", + entityId: issueId, + details: expect.objectContaining({ + bodySnippet: "Runner progress", + identifier: "RNT-1", + issueTitle: "Exercise real runner tools", + source: "paperclip_runner_protocol", + }), + }); await expect(authority.execute({ ...call, arguments: { body: "Changed", idempotencyKey: "progress-1" }, diff --git a/server/src/services/native-runtime/paperclip-runner-tool-authority.ts b/server/src/services/native-runtime/paperclip-runner-tool-authority.ts index d444319aa86..ae6e60469b3 100644 --- a/server/src/services/native-runtime/paperclip-runner-tool-authority.ts +++ b/server/src/services/native-runtime/paperclip-runner-tool-authority.ts @@ -113,7 +113,7 @@ export class PaperclipRunnerToolAuthority { .where(eq(issueApprovals.approvalId, approval.id)); return { approval, tasks: tasks.map((row) => row.issue) }; } - case "report_progress": return this.#reportProgress(call.callId, call.arguments); + case "report_progress": return this.#reportProgress(call.callId, input, context.issue); case "request_human_input": return this.#requestHumanInput(input, context.issue); default: throw new Error("paperclip_runner_tool_not_bound"); } @@ -146,12 +146,15 @@ export class PaperclipRunnerToolAuthority { return row; } - async #reportProgress(callId: string, value: unknown): Promise { - const input = record(value); + async #reportProgress( + callId: string, + input: Record, + issue: typeof issues.$inferSelect, + ): Promise { const body = typeof input.body === "string" ? input.body.trim() : ""; const idempotencyKey = typeof input.idempotencyKey === "string" ? input.idempotencyKey.trim() : ""; if (!body || !idempotencyKey) throw new Error("paperclip_runner_tool_input_invalid"); - return this.db.transaction(async (tx) => { + const mutation = await this.db.transaction(async (tx) => { const [run] = await tx.select().from(heartbeatRuns) .where(and(eq(heartbeatRuns.id, this.binding.runId), eq(heartbeatRuns.companyId, this.binding.companyId))) .for("update").limit(1); @@ -165,7 +168,7 @@ export class PaperclipRunnerToolAuthority { if (prior.operationId !== "report_progress" || canonicalJson(prior.input) !== canonicalJson(input)) { throw new Error("paperclip_runner_tool_idempotency_conflict"); } - return prior.result; + return { result: prior.result, publication: null }; } const comment = await issueService(this.db).addComment( this.binding.issueId, @@ -180,8 +183,29 @@ export class PaperclipRunnerToolAuthority { resultJson: { ...resultJson, semanticToolReceipts: receipts }, updatedAt: new Date(), }).where(eq(heartbeatRuns.id, this.binding.runId)); - return result; + const activity = await persistActivity(tx as unknown as Db, { + companyId: this.binding.companyId, + actorType: "agent", + actorId: this.binding.agentId, + agentId: this.binding.agentId, + runId: this.binding.runId, + issueId: this.binding.issueId, + action: "issue.comment_added", + entityType: "issue", + entityId: this.binding.issueId, + details: { + commentId: comment.id, + bodySnippet: comment.body.slice(0, 120), + identifier: issue.identifier, + issueTitle: issue.title, + authorizationReason: "paperclip_runner_protocol", + source: "paperclip_runner_protocol", + }, + }); + return { result, publication: activity.publication }; }); + if (mutation.publication) publishActivity(mutation.publication); + return mutation.result; } async #writeDocument(input: Record): Promise {