diff --git a/server/src/services/native-runtime/native-deliverable-feedback.ts b/server/src/services/native-runtime/native-deliverable-feedback.ts index b9369cd2959..3cba6e59d5c 100644 --- a/server/src/services/native-runtime/native-deliverable-feedback.ts +++ b/server/src/services/native-runtime/native-deliverable-feedback.ts @@ -1,6 +1,9 @@ import { and, eq } from "drizzle-orm"; -import { assets, companies, documents, documentRevisions, heartbeatRuns, issueDocuments, issueAttachments, issues, issueWorkProducts, mcpAttachmentUploads, dotRunnerAssignments, type Db } from "@paperclipai/db"; -import type { PrpStructuredRunResult } from "../../vendor/paperclip-runner/index.js"; +import { createHash } from "node:crypto"; +import { assets, companies, completionContracts, documents, documentRevisions, heartbeatRuns, issueDocuments, issueAttachments, issues, issueThreadInteractions, issueWorkProducts, mcpAttachmentUploads, dotRunnerAssignments, type Db } from "@paperclipai/db"; +import { projectHumanInteractionResponse } from "../execution-continuation.js"; + +export class NativeDeliverableEvidenceError extends Error {} function evidenceRefs(value: unknown): string[] { if (!Array.isArray(value)) return []; @@ -16,9 +19,24 @@ function record(value: unknown): Record { ? value as Record : {}; } +type NativePublicationEvidence = { + disposition: unknown; + artifacts: string[]; + refs: string[]; +}; + +function publicationEvidence(value: unknown): NativePublicationEvidence { + const result = record(value); + const artifacts = evidenceRefs(result.artifacts); + const criteria = record(result.completionClaim).criteria; + return { disposition: result.reportedWorkDisposition, artifacts, + refs: [...evidenceRefs(result.evidence), ...artifacts, + ...(Array.isArray(criteria) ? criteria.flatMap(entry => evidenceRefs(record(entry).evidenceRefs)) : [])] }; +} + async function hasCurrentPublicationReceipt(db: Db, companyId: string, receipts: unknown, attachment: { id: string; originatingRunId: string | null; filename: string | null; byteSize: number; sha256: string; -}): Promise { +}, lockSources = false): Promise { for (const value of Object.values(record(receipts))) { const receipt = record(value); const input = record(receipt.input); @@ -32,13 +50,14 @@ async function hasCurrentPublicationReceipt(db: Db, companyId: string, receipts: && /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/iu.test(input.workSessionId) && input.filename === attachment.filename && input.byteSize === attachment.byteSize && input.sha256 === attachment.sha256) { - const [upload] = await db.select({ id: mcpAttachmentUploads.id }).from(mcpAttachmentUploads) + const query = db.select({ id: mcpAttachmentUploads.id }).from(mcpAttachmentUploads) .innerJoin(dotRunnerAssignments, and(eq(dotRunnerAssignments.id, mcpAttachmentUploads.workSessionId), eq(dotRunnerAssignments.companyId, companyId))) .where(and(eq(mcpAttachmentUploads.id, input.uploadId), eq(mcpAttachmentUploads.companyId, companyId), eq(mcpAttachmentUploads.workSessionId, input.workSessionId), eq(mcpAttachmentUploads.attachmentId, attachment.id), eq(dotRunnerAssignments.runId, attachment.originatingRunId), eq(mcpAttachmentUploads.originalFilename, attachment.filename), eq(mcpAttachmentUploads.byteSize, attachment.byteSize), eq(mcpAttachmentUploads.sha256, attachment.sha256))).limit(1); + const [upload] = await (lockSources ? query.for("share", { of: [mcpAttachmentUploads, dotRunnerAssignments], noWait: true }) : query); if (upload) return true; } if (receipt.operationId === "register_deliverable" && @@ -64,10 +83,11 @@ async function hasCurrentPublicationReceipt(db: Db, companyId: string, receipts: const uuid = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/iu; if (typeof source.attachmentId !== "string" || !uuid.test(source.attachmentId) || typeof source.commentId !== "string" || !uuid.test(source.commentId)) continue; - const [original] = await db.select({ filename: assets.originalFilename, byteSize: assets.byteSize, sha256: assets.sha256 }) + const query = db.select({ filename: assets.originalFilename, byteSize: assets.byteSize, sha256: assets.sha256 }) .from(issueAttachments).innerJoin(assets, and(eq(assets.id, issueAttachments.assetId), eq(assets.companyId, companyId))) .where(and(eq(issueAttachments.id, source.attachmentId), eq(issueAttachments.companyId, companyId), eq(issueAttachments.issueCommentId, source.commentId))).limit(1); + const [original] = await (lockSources ? query.for("share", { of: [issueAttachments, assets], noWait: true }) : query); if (original?.filename === attachment.filename && original.byteSize === attachment.byteSize && original.sha256.toLowerCase() === attachment.sha256.toLowerCase()) return true; } @@ -178,12 +198,51 @@ export function explicitlyRequestsTaskDocumentOutput(objective: string): boolean }); } +async function declinedTaskDocumentNotRequired(db: Db, binding: { + companyId: string; issueId: string; runId: string; objective: string; lockSources?: boolean; +}): Promise { + if (!/\bif I (?:decline|reject) the action\b/iu.test(binding.objective) || + !/\bno (?:briefing|document|doc)(?: document)? is required after a (?:decline|rejection)\b/iu.test(binding.objective)) return false; + const runQuery = db.select({ context: heartbeatRuns.contextSnapshot, description: issues.description }).from(heartbeatRuns) + .innerJoin(issues, and(eq(issues.id, binding.issueId), eq(issues.companyId, binding.companyId))).where(and( + eq(heartbeatRuns.id, binding.runId), eq(heartbeatRuns.companyId, binding.companyId), eq(heartbeatRuns.nativeIssueId, binding.issueId), + )).limit(1); + const [run] = await (binding.lockSources ? runQuery.for("share", { of: [heartbeatRuns, issues], noWait: true }) : runQuery); + const continuation = record(run?.context?.executionContinuation); + const trigger = record(continuation.trigger); + const objectiveSource = record(continuation.objectiveSource); + const uuid = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/iu; + if (run?.description?.trim() !== binding.objective.trim() || continuation.objective !== binding.objective || typeof trigger.interactionId !== "string" || + !uuid.test(trigger.interactionId) || typeof trigger.sourceRunId !== "string" || !uuid.test(trigger.sourceRunId) || + objectiveSource.kind !== "description" || objectiveSource.id !== binding.issueId || + objectiveSource.revision !== createHash("sha256").update(binding.objective.trim()).digest("hex")) return false; + const query = db.select({ interaction: issueThreadInteractions, sourceContext: heartbeatRuns.contextSnapshot, + sourceContract: completionContracts.contractJson }).from(issueThreadInteractions) + .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, issueThreadInteractions.sourceRunId), + eq(heartbeatRuns.companyId, binding.companyId), eq(heartbeatRuns.nativeIssueId, binding.issueId), eq(heartbeatRuns.runtimeMode, "native"))) + .innerJoin(completionContracts, and(eq(completionContracts.id, heartbeatRuns.completionContractId), + eq(completionContracts.companyId, binding.companyId), eq(completionContracts.issueId, binding.issueId))) + .where(and(eq(issueThreadInteractions.id, trigger.interactionId), eq(issueThreadInteractions.companyId, binding.companyId), + eq(issueThreadInteractions.issueId, binding.issueId), eq(issueThreadInteractions.sourceRunId, trigger.sourceRunId))).limit(1); + const [saved] = await (binding.lockSources ? query.for("share", { of: [issueThreadInteractions, heartbeatRuns, completionContracts], noWait: true }) : query); + if (!saved || saved.interaction.kind !== "request_confirmation" || saved.interaction.dismissedAt || + projectHumanInteractionResponse(saved.interaction)?.status !== "rejected" || + Object.keys(record(record(saved.interaction.payload).toolAction)).length === 0) return false; + const source = record(saved.sourceContext?.executionContinuation); + const originalSource = record(source.objectiveSource); + const originalCriteria = record(saved.sourceContract).criteria; + return source.objective === binding.objective && + originalSource.kind === objectiveSource.kind && originalSource.id === objectiveSource.id && + originalSource.revision === objectiveSource.revision && + Array.isArray(originalCriteria) && originalCriteria.some(value => record(value).requirement === binding.objective); +} + /** Current attached revisions with server-owned publication proof. Joining the * revision to its originating run preserves completed work across continuations * without accepting stale, foreign-task, or provider-invented document refs. */ -export async function publishedTaskDocuments(db: Db, binding: { companyId: string; issueId: string }) { - const saved = await db.select({ id: documents.id, revisionId: documents.latestRevisionId, +export async function publishedTaskDocuments(db: Db, binding: { companyId: string; issueId: string; lockSources?: boolean }) { + const query = db.select({ id: documents.id, revisionId: documents.latestRevisionId, key: issueDocuments.key, resultJson: heartbeatRuns.resultJson, issuePrefix: companies.issuePrefix, issueIdentifier: issues.identifier }) .from(issueDocuments) @@ -196,6 +255,7 @@ export async function publishedTaskDocuments(db: Db, binding: { companyId: strin eq(heartbeatRuns.companyId, binding.companyId), eq(heartbeatRuns.nativeIssueId, binding.issueId), eq(heartbeatRuns.runtimeMode, "native"))) .where(and(eq(issueDocuments.companyId, binding.companyId), eq(issueDocuments.issueId, binding.issueId))); + const saved = await (binding.lockSources ? query.for("share", { of: [issueDocuments, documents, documentRevisions, heartbeatRuns], noWait: true }) : query); return saved.filter(document => Object.values(record(document.resultJson?.semanticToolReceipts)).some(value => { const receipt = record(value), result = record(receipt.result), published = record(result.document); return receipt.operationId === "write_document" && ["applied", "duplicate"].includes(String(result.disposition)) @@ -207,18 +267,16 @@ export async function publishedTaskDocuments(db: Db, binding: { companyId: strin /** Files cited as completed output must be reachable outside the agent workspace. */ export async function validateNativeDeliverableEvidence( db: Db, - binding: { companyId: string; issueId: string; runId: string; objective: string; semanticToolReceipts: unknown }, - result: PrpStructuredRunResult, + binding: { companyId: string; issueId: string; runId: string; objective: string; semanticToolReceipts: unknown; lockSources?: boolean; authoritativeDone?: true }, + result: unknown, ): Promise { - if (result.reportedWorkDisposition !== "done") return; + const evidence = publicationEvidence(result); + if (!binding.authoritativeDone && evidence.disposition !== "done") return; const fileRequested = explicitlyRequestsFileOutput(binding.objective); - const taskDocumentRequested = explicitlyRequestsTaskDocumentOutput(binding.objective); - const artifactRefs = new Set(evidenceRefs(result.artifacts)); - const refs = new Set([ - ...evidenceRefs(result.evidence), - ...artifactRefs, - ...result.completionClaim.criteria.flatMap(({ evidenceRefs }) => evidenceRefs), - ]); + const taskDocumentRequested = explicitlyRequestsTaskDocumentOutput(binding.objective) && + !await declinedTaskDocumentNotRequired(db, binding); + const artifactRefs = new Set(evidence.artifacts); + const refs = new Set(evidence.refs); const citesTaskDocument = [...refs].some(ref => /^\/[^/?#]+\/issues\/[^/?#]+#document-[^/?#]+$/u.test(ref.trim())); const publishedDocuments = taskDocumentRequested || citesTaskDocument ? await publishedTaskDocuments(db, binding) : []; const publishedTaskDocument = taskDocumentRequested && publishedDocuments.length > 0; @@ -232,23 +290,23 @@ export async function validateNativeDeliverableEvidence( if (ref.startsWith("deliverable:") || attachmentPath) { const id = attachmentPath?.[1] ?? ref.slice("deliverable:".length); const uuid = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/iu; - const [attachment] = uuid.test(id) - ? await db.select({ id: issueAttachments.id, originatingRunId: issueAttachments.originatingRunId, + const query = db.select({ id: issueAttachments.id, originatingRunId: issueAttachments.originatingRunId, filename: assets.originalFilename, byteSize: assets.byteSize, sha256: assets.sha256 }).from(issueAttachments) .innerJoin(assets, and(eq(assets.id, issueAttachments.assetId), eq(assets.companyId, binding.companyId))) .where(and(eq(issueAttachments.id, id), eq(issueAttachments.companyId, binding.companyId), eq(issueAttachments.issueId, binding.issueId))) - .limit(1) - : []; + .limit(1); + const [attachment] = uuid.test(id) + ? await (binding.lockSources ? query.for("share", { of: [issueAttachments, assets], noWait: true }) : query) : []; if (!attachment) { - throw new Error("Completion cites no registered attachment on this task. Use register_deliverable for the requested file and cite deliverable: from its receipt. No human completion approval was created."); + throw new NativeDeliverableEvidenceError("Completion cites no registered attachment on this task. Use register_deliverable for the requested file and cite deliverable: from its receipt. No human completion approval was created."); } // A prior output (or user input) can be useful context, but does not prove // this run published the newly requested output. The receipt survives a // controller restart of this run; a replacement can re-register preserved // workspace bytes internally rather than asking the user to confirm them. if ((fileRequested || taskDocumentRequested) && attachment.originatingRunId !== binding.runId) continue; - if ((fileRequested || taskDocumentRequested) && !await hasCurrentPublicationReceipt(db, binding.companyId, binding.semanticToolReceipts, attachment)) { - throw new Error("This attachment has no matching verified publication receipt for this run's requested output. Inspect any preserved file and use register_deliverable to verify its current filename, size, and SHA-256, then cite the new receipt. No human completion approval was created."); + if ((fileRequested || taskDocumentRequested) && !await hasCurrentPublicationReceipt(db, binding.companyId, binding.semanticToolReceipts, attachment, binding.lockSources)) { + throw new NativeDeliverableEvidenceError("This attachment has no matching verified publication receipt for this run's requested output. Inspect any preserved file and use register_deliverable to verify its current filename, size, and SHA-256, then cite the new receipt. No human completion approval was created."); } registeredAttachment = true; continue; @@ -258,19 +316,21 @@ export async function validateNativeDeliverableEvidence( const localFile = /^(?:file:|\.{0,2}\/|[a-z]:[\\/])/iu.test(ref) || (!/^[a-z][a-z0-9+.-]*:/iu.test(ref) && /^[^\r\n]+\.[a-z0-9]{1,16}(?::\d+(?::\d+)?)?$/iu.test(ref)); if (localFile && taskDocumentRequested && !publishedTaskDocument) { - throw new Error("The requested task document is only a workspace file. Publish it on this task with write_document, or attach the verified file with register_deliverable and cite deliverable:. Reuse completed work; do not request a new completion approval."); + throw new NativeDeliverableEvidenceError("The requested task document is only a workspace file. Publish it on this task with write_document, or attach the verified file with register_deliverable and cite deliverable:. Reuse completed work; do not request a new completion approval."); } if (localFile && (fileRequested || artifactRefs.has(value))) { - throw new Error("Completion cites a workspace-only file that the user cannot download. Before finishing, use register_deliverable for requested file outputs and cite deliverable: from the receipt, with /api/attachments//content as the download link. For repository changes, cite an accessible PR or registered work product instead. No human completion approval was created."); + throw new NativeDeliverableEvidenceError("Completion cites a workspace-only file that the user cannot download. Before finishing, use register_deliverable for requested file outputs and cite deliverable: from the receipt, with /api/attachments//content as the download link. For repository changes, cite an accessible PR or registered work product instead. No human completion approval was created."); } } if (taskDocumentRequested && !publishedTaskDocument && !registeredAttachment) { - throw new Error("The requested document has not been published on this task. Use write_document, or register_deliverable with its verified attachment receipt. A workspace path or final message alone is not the requested task document. Reuse completed work without requesting a new completion approval."); + throw new NativeDeliverableEvidenceError("The requested document has not been published on this task. Use write_document, or register_deliverable with its verified attachment receipt. A workspace path or final message alone is not the requested task document. Reuse completed work without requesting a new completion approval."); } if (fileRequested && !registeredAttachment) { - const products = refs.size ? await db.select().from(issueWorkProducts).where(and( + const query = db.select().from(issueWorkProducts).where(and( eq(issueWorkProducts.companyId, binding.companyId), eq(issueWorkProducts.issueId, binding.issueId), - )) : []; + eq(issueWorkProducts.createdByRunId, binding.runId), + )); + const products = refs.size ? await (binding.lockSources ? query.for("share", { noWait: true }) : query) : []; const accessibleProduct = products.some(product => { if (product.createdByRunId !== binding.runId) return false; if (["failed", "cancelled", "archived"].includes(product.status)) return false; @@ -281,6 +341,6 @@ export async function validateNativeDeliverableEvidence( return accessible && [product.url, `work_product:${product.id}`, `work-product:${product.id}`, `artifact:${product.id}`] .some(ref => typeof ref === "string" && refs.has(ref)); }); - if (!accessibleProduct) throw new Error("The requested file has no accessible delivery evidence. Use register_deliverable and cite deliverable:, or cite an accessible work product registered by this run for this task. Empty evidence, prior-run output, and a verification result cannot substitute for the requested file. Continue publishing or report a concrete blocker; no human completion approval was created."); + if (!accessibleProduct) throw new NativeDeliverableEvidenceError("The requested file has no accessible delivery evidence. Use register_deliverable and cite deliverable:, or cite an accessible work product registered by this run for this task. Empty evidence, prior-run output, and a verification result cannot substitute for the requested file. Continue publishing or report a concrete blocker; no human completion approval was created."); } } diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index b1c903b0239..55482228c82 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -12,6 +12,7 @@ import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export- import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { readRequiredQualityReview } from "../required-quality-review.js"; +import { NativeDeliverableEvidenceError, validateNativeDeliverableEvidence } from "./native-deliverable-feedback.js"; import { pendingNativeGovernance } from "./native-governance.js"; import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; import { onboardingFirstTaskCompletionAllowed } from "../onboarding-first-task-admission.js"; @@ -1466,7 +1467,7 @@ export async function finalizeNativeRun(input: { ? await readNativeMonitorDelivery(input.db, { companyId: run.companyId, issueId: coordinator.issueId, runId: run.id }) : null; const qualityReview = await readRequiredQualityReview(input.db, authoritativeIssue); - let proposedDecision = resolveNativeFinalizerStatus({ + const statusInput: Parameters[0] = { requiredQualityReviewSatisfied: qualityReview.satisfied, caseHandoffAuthorized: Boolean(await assertNativeCaseHandoffResult(input.db, run.id, authoritativeIssue.id, result)), monitorWaitDispatched: monitorDelivery !== null, @@ -1509,7 +1510,8 @@ export async function finalizeNativeRun(input: { null, agentId: run.agentId, priorIssueStatus: authoritativeStatus(authoritativeIssue.status), - }); + }; + let proposedDecision = resolveNativeFinalizerStatus(statusInput); // Already accepted results still require the server-owned first-task decision. // Keep the task open without inventing an interaction or rerunning the work. if (proposedDecision.statusAction === "done" && !await onboardingFirstTaskCompletionAllowed(input.db, { @@ -1518,6 +1520,22 @@ export async function finalizeNativeRun(input: { proposedDecision = { ...proposedDecision, statusAction: "in_review", toStatus: "in_review", reasonCode: "onboarding_first_task_decision_required", unblockDescriptor: null, effects: [{ kind: "release_checkout" }] }; } + if (proposedDecision.statusAction === "done") { + const continuation = record(run.contextSnapshot?.executionContinuation); + const objective = typeof continuation.objective === "string" ? continuation.objective + : [authoritativeIssue.title, authoritativeIssue.description].filter(Boolean).join("\n"); + try { + await validateNativeDeliverableEvidence(input.db, { companyId: run.companyId, issueId: authoritativeIssue.id, + runId: run.id, objective, semanticToolReceipts: run.resultJson?.semanticToolReceipts }, result); + } catch (error) { + if (!(error instanceof NativeDeliverableEvidenceError)) throw error; + const incomplete = resolveNativeFinalizerStatus({ ...statusInput, + assessment: { ...assessment, hasBlockingRemainingWork: true, continuation: null } }); + proposedDecision = { ...incomplete, effects: incomplete.effects.map(effect => + effect.kind === "enqueue_continuation" ? { ...effect, summary: error.message } + : effect.kind === "record_finalization_error" ? { ...effect, nextAction: error.message } : effect) }; + } + } const decision = conversationNativeDecision({ conversation: isConversation(authoritativeIssue), terminalState, workspaceFinalizeStatus: input.workspaceFinalizeStatus, hasGovernanceGate: !!governanceGate, diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts b/server/src/services/native-runtime/paperclip-control-plane-port.test.ts index f1a88c8378a..bae43dd7f39 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts +++ b/server/src/services/native-runtime/paperclip-control-plane-port.test.ts @@ -4,7 +4,10 @@ import { plugins } from "@paperclipai/db"; import { readTaskPanelActivations } from "../plugin-task-panels.js"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { and, asc, eq, sql } from "drizzle-orm"; -import { generateKeyPairSync, randomUUID } from "node:crypto"; +import { createHash, generateKeyPairSync, randomUUID } from "node:crypto"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; import { renderPaperclipWakePrompt } from "@paperclipai/adapter-utils/server-utils"; import { ONBOARDING_FIRST_TASK_ORIGIN_KIND } from "@paperclipai/shared"; import { readCompletedAssistantMessageCandidate, resolveHeartbeatRunResponse, selectHeartbeatRunFinalAgentMessage } from "../heartbeat-run-summary.js"; @@ -12,6 +15,7 @@ import { activityLog, agentWakeupRequests, agents, + assets, authUsers, companies, companyMemberships, @@ -59,6 +63,11 @@ import { claimQueuedNativeReviewRun } from "./native-review-dispatch.js"; import { stageNativeRunnerWakeAttachments } from "./native-runner-file-handoff.js"; import { PaperclipRunnerToolAuthority } from "./paperclip-runner-tool-authority.js"; import { nativeCompletionFeedback } from "./native-completion-feedback.js"; +import { ensureNativeCompletionContract, nativeCompletionSource } from "./completion-contracts.js"; +import { commitNativeStatusDecision } from "./status-decision-committer.js"; +import { NATIVE_STATUS_ARBITER_POLICY_VERSION } from "./status-arbiter.js"; +import { createStorageService } from "../../storage/service.js"; +import { createLocalDiskStorageProvider } from "../../storage/local-disk-provider.js"; import { buildOnboardingFirstTaskOpeningQuestion, ONBOARDING_FIRST_TASK_OPENING_QUESTION_ID } from "../onboarding-first-task-assets.js"; import { assertCurrentWakeCommentsRead } from "./current-wake-comments.js"; import { buildExecutionContinuation } from "../execution-continuation.js"; @@ -74,6 +83,7 @@ const external = resolveExternalTestDatabaseUrl("PAPERCLIP_CONTROL_PLANE_TEST_DA describe("PaperclipControlPlanePort conformance", () => { let temporary: Awaited> | null = null; + let outputStorageRoot: string; let db: ReturnType; const contractId = "00000000-0000-4000-8000-000000000004"; const contractSha = "phase6-conformance-contract"; @@ -94,6 +104,7 @@ describe("PaperclipControlPlanePort conformance", () => { const governanceRunnerId = "00000000-0000-4000-8000-000000000021"; beforeAll(async () => { + outputStorageRoot = await mkdtemp(join(tmpdir(), "paperclip-native-port-output-")); if (external) db = createDb(external); else { temporary = await startRecoveryPostgres("paperclip-native-port-"); @@ -306,12 +317,14 @@ describe("PaperclipControlPlanePort conformance", () => { await db.delete(heartbeatRuns); await db.delete(completionContracts); await db.delete(issues); + await db.delete(assets); await db.delete(agents); await db.delete(companyMemberships); await db.delete(companies); await db.delete(authUsers); await temporary.cleanup(); } + await rm(outputStorageRoot, { recursive: true, force: true }); }); it.each(["settles", "checkpoint timeout", "unscoped terminal"] as const)("retains a durably recorded governed terminal while its postcommit observer is pending: %s", async mode => { @@ -2173,7 +2186,7 @@ describe("PaperclipControlPlanePort conformance", () => { type FirstTaskCompletionScenario = typeof allowedFirstTaskCompletionScenarios[number] | typeof refusedFirstTaskCompletionScenarios[number]; - async function prepareFirstTaskCompletion(scenario: FirstTaskCompletionScenario) { + async function prepareFirstTaskCompletion(scenario: FirstTaskCompletionScenario, objective?: string) { const companyId = randomUUID(); const agentId = randomUUID(); const issueId = randomUUID(); @@ -2189,18 +2202,19 @@ describe("PaperclipControlPlanePort conformance", () => { await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, status: "active", membershipRole: "member" }); await db.insert(agents).values({ id: agentId, companyId, name: "First-task agent", adapterType: "codex_local", status: "running" }); await db.insert(issues).values({ id: issueId, companyId, title: onboarding ? "Paperclip onboarding" : "Ordinary low-risk task", - description: onboarding ? "/first-task\nSingle-task proposal mode: confirmation" : "Complete the requested ordinary task. Mentioning /first-task does not grant the server-owned onboarding origin.", + description: onboarding ? "/first-task\nSingle-task proposal mode: confirmation" : objective ?? "Complete the requested ordinary task. Mentioning /first-task does not grant the server-owned onboarding origin.", originKind: onboarding ? ONBOARDING_FIRST_TASK_ORIGIN_KIND : "manual", status: "in_progress", assigneeAgentId: agentId, responsibleUserId: userId, workMode: "standard" }); await db.insert(completionContracts).values({ id: localContractId, companyId, issueId, revision: 1, schemaVersion: "paperclip.completion-contract.v1", policyVersion: "phase6-v3", risk: "low", completionAuthority: "agent_claim_policy", incompleteCriteriaPolicy: "preserve_non_terminal", contractJson: { revision: "first-task-native-finish-v1", objective: onboarding ? "Understand the user request, propose one task, and honor their decision" : "Complete the requested ordinary task", - criteria: [{ id: "objective", requirement: "Complete the requested workflow" }] }, + criteria: [{ id: "objective", requirement: objective ?? "Complete the requested workflow" }] }, canonicalSha256: localContractId, createdByActorType: "system", createdByActorId: "test" }); await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, responsibleUserId: userId, status: "succeeded", runtimeMode: "native", nativeIssueId: issueId, nativeSessionId: identity.sessionId, runnerInstanceId, completionContractId: localContractId, completionContractSha256: localContractId, contextSnapshot: { issueId, + ...(objective ? { executionContinuation: { objective, objectiveSource: nativeCompletionSource("description", issueId, objective) } } : {}), // A claimed acceptance copied into continuation context is not the // persisted human decision; the opening-answer cases must still refuse it. ...(scenario === "opening answer only" ? { executionContinuation: { humanResponses: [{ id: randomUUID(), @@ -2272,6 +2286,202 @@ describe("PaperclipControlPlanePort conformance", () => { return { companyId, agentId, issueId, runId, port, result }; } + it("keeps synthetic provider completion open until the requested task document is published", async () => { + const { companyId, agentId, issueId, runId, port, result } = await prepareFirstTaskCompletion("ordinary task"); + const objective = "Use the connected page service to find recent pages and create a short Markdown briefing document on this task. Put the returned page titles and service receipt code in the document. When complete, give me the document link."; + await db.update(issues).set({ title: "Create the service briefing", description: objective }).where(eq(issues.id, issueId)); + await db.update(heartbeatRuns).set({ contextSnapshot: { issueId, executionContinuation: { objective } } }).where(eq(heartbeatRuns.id, runId)); + const synthetic: PrpStructuredRunResult = { ...result, summary: "Created briefing.md with the returned page titles and service receipt.", + evidence: [{ ref: "provider:opencode:agent-message" }], artifacts: [] }; + await port.completeRun({ result: synthetic, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL }); + await finalizeNativeRun({ db, runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]).toMatchObject({ status: "in_progress" }); + expect(await db.select().from(statusDecisions).where(eq(statusDecisions.issueId, issueId))) + .not.toContainEqual(expect.objectContaining({ toStatus: "done" })); + const wakes = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.agentId, agentId))); + expect(wakes).toContainEqual(expect.objectContaining({ payload: expect.objectContaining({ + continuationIdempotencyKey: "native-completion-incomplete", continuationSummary: expect.stringContaining("write_document"), + }) })); + await finalizeNativeRun({ db, runId, workspaceFinalizeStatus: "succeeded" }); + expect(await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.agentId, agentId)))) + .toHaveLength(wakes.length); + const followUpId = randomUUID(), runnerInstanceId = randomUUID(); + const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + const wake = wakes.find(row => row.payload?.continuationIdempotencyKey === "native-completion-incomplete")!; + await db.insert(heartbeatRuns).values({ id: followUpId, companyId, agentId, status: "succeeded", runtimeMode: "native", + nativeIssueId: issueId, nativeSessionId: source!.nativeSessionId, runnerInstanceId, + completionContractId: source!.completionContractId, completionContractSha256: source!.completionContractSha256, + wakeupRequestId: wake.id, contextSnapshot: source!.contextSnapshot }); + await db.update(agentWakeupRequests).set({ runId: followUpId }).where(eq(agentWakeupRequests.id, wake.id)); + await db.update(issues).set({ executionRunId: followUpId }).where(eq(issues.id, issueId)); + const followUp = new PaperclipControlPlanePort(db, { companyId, issueId, agentId, runId: followUpId, + sessionId: source!.nativeSessionId!, sourceInstanceId: runnerInstanceId, + completionContractId: source!.completionContractId!, completionContractSha256: source!.completionContractSha256!, + controlPlaneSourceInstanceId: "output-follow-up-control" }); + await followUp.openRun({ identity: { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, companyId, issueId, agentId, runId: followUpId }, + backendKind: "mock", sourceInstanceId: runnerInstanceId }); + await followUp.completeRun({ result: synthetic, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL }); + await finalizeNativeRun({ db, runId: followUpId, workspaceFinalizeStatus: "succeeded" }); + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]!.status).toBe("in_progress"); + expect(await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.agentId, agentId)))) + .toHaveLength(wakes.length); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId))) + .toContainEqual(expect.objectContaining({ ownerAgentId: agentId, cause: "completion_evidence_incomplete", nextAction: expect.stringContaining("write_document") })); + expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runId)))[0]!.resultJson) + .toMatchObject({ result: { reportedWorkDisposition: "done", evidence: synthetic.evidence, artifacts: [] } }); + }); + + it.each(["saved document", "registered attachment", "plain Ask"] as const)( + "accepts native completion with %s", async (delivery) => { + const objective = delivery === "plain Ask" ? "What is two plus two?" : "Create a Markdown briefing document on this task."; + const { companyId, agentId, issueId, runId, port, result } = await prepareFirstTaskCompletion("ordinary task", objective); + await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, runId)); + const report: PrpStructuredRunResult = { ...result, evidence: [{ ref: "provider:opencode:agent-message" }], artifacts: [] }; + if (delivery === "saved document") { + const authority = new PaperclipRunnerToolAuthority(db, { companyId, agentId, issueId, runId }); + await authority.execute({ tool: "write_document", callId: "briefing", arguments: { + idempotencyKey: "briefing", key: "briefing", title: "Briefing", body: "Verified service titles and receipt.", baseRevisionId: null, + } }); + } else if (delivery === "registered attachment") { + const bytes = Buffer.from("# Briefing\nVerified service titles and receipt.\n"); + const authority = new PaperclipRunnerToolAuthority(db, { companyId, agentId, issueId, runId, + workspaceRoot: "/workspace/output-control", executionTargetKind: "remote", readRemoteTaskFile: async () => bytes, + storage: createStorageService(createLocalDiskStorageProvider(outputStorageRoot)) }); + const receipt = await authority.execute({ tool: "register_deliverable", callId: "briefing", arguments: { + idempotencyKey: "briefing", filename: "briefing.md", contentType: "text/markdown", contentRef: "briefing.md", + byteSize: bytes.length, sha256: createHash("sha256").update(bytes).digest("hex"), title: "Briefing", + } }) as { entityRefs: string[] }; + report.artifacts = [{ kind: "file", ref: `deliverable:${receipt.entityRefs[0]}` }]; + } + await port.completeRun({ result: report, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL }); + await finalizeNativeRun({ db, runId, workspaceFinalizeStatus: "succeeded" }); + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]!.status).toBe("done"); + }); + + it.each(["current human rejection", "model-only decline", "different objective revision", "different trigger", + "changed description after admission", "changed description at commit"] as const)( + "binds an explicit declined-document fallback to %s", async (scenario) => { + const objective = "Use the connected page service to find recent pages and create a short Markdown briefing document on this task. Put the returned page titles and service receipt code in the document. When complete, give me the document link. If I decline the action, do not try again or request another connection. Instead, give me a brief explanation that you could not retrieve the data. That explanation is the complete allowed fallback; no briefing is required after a decline."; + const source = await prepareFirstTaskCompletion("ordinary task", objective); + const { companyId, agentId, issueId } = source; + const [issue] = await db.select().from(issues).where(eq(issues.id, issueId)); + const interactionId = randomUUID(); + await db.insert(issueThreadInteractions).values({ id: interactionId, companyId, issueId, sourceRunId: source.runId, + createdByAgentId: agentId, kind: "request_confirmation", status: "pending", requestedResolverPolicy: "human_only", + effectiveResolverPolicy: "human_only", continuationPolicy: "wake_assignee", payload: { version: 1, prompt: "Approve service read", + toolAction: { version: 1, actionRequestId: randomUUID(), invocationId: randomUUID(), toolName: "pages.read", + toolDisplayName: "Read pages", connectionId: null, applicationId: null, appDisplayName: null, risk: "read", + previewMarkdown: "Read pages", argumentsSummaryJson: "{}", argumentsHash: "test-hash", expiresAt: "2099-01-01T00:00:00Z" } } }); + if (scenario !== "model-only decline") { + await issueThreadInteractionService(db).rejectInteraction({ id: issueId, companyId }, interactionId, + { reason: "Do not call the service." }, { userId: issue!.responsibleUserId! }); + } + const contract = await ensureNativeCompletionContract({ db, companyId, issue: issue!, actorId: agentId, humanResponseId: interactionId }); + const runId = randomUUID(), runnerInstanceId = randomUUID(); + const continuedObjective = scenario === "different objective revision" ? `${objective} Still create the document even if I decline.` : objective; + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "succeeded", runtimeMode: "native", nativeIssueId: issueId, + nativeSessionId: CONTROL_PLANE_CONFORMANCE_OPEN.identity.sessionId, runnerInstanceId, completionContractId: contract.row.id, + completionContractSha256: contract.row.canonicalSha256, contextSnapshot: { issueId, executionContinuation: { + objective: continuedObjective, objectiveSource: nativeCompletionSource("description", issueId, continuedObjective), + trigger: { interactionId: scenario === "different trigger" ? randomUUID() : interactionId, sourceRunId: source.runId }, + } } }); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId)); + const port = new PaperclipControlPlanePort(db, { companyId, issueId, agentId, runId, + sessionId: CONTROL_PLANE_CONFORMANCE_OPEN.identity.sessionId, sourceInstanceId: runnerInstanceId, + completionContractId: contract.row.id, completionContractSha256: contract.row.canonicalSha256, + controlPlaneSourceInstanceId: "declined-output-control" }); + await port.openRun({ identity: { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, companyId, issueId, agentId, runId }, + backendKind: "mock", sourceInstanceId: runnerInstanceId }); + const report: PrpStructuredRunResult = { ...source.result, summary: "The service action was declined, so I could not retrieve the data.", + evidence: [{ ref: "provider:opencode:agent-message" }], completionClaim: { ...source.result.completionClaim, + contractRevision: contract.contract.revision, criteria: contract.contract.criteria.map(entry => ({ criterionId: entry.id, status: "satisfied", evidenceRefs: [] })) } }; + await port.completeRun({ result: report, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL }); + const changedDescription = "Create the Markdown briefing document on this task. The prior decline does not waive the required document."; + if (scenario === "changed description after admission") { + await db.update(issues).set({ description: changedDescription }).where(eq(issues.id, issueId)); + } + const trigger = `decline_revision_race_${randomUUID().replaceAll("-", "")}`; + if (scenario === "changed description at commit") { + await db.execute(sql.raw(`create function ${trigger}() returns trigger language plpgsql as $$ + begin + if NEW.issue_id = '${issueId}'::uuid and NEW.run_id = '${runId}'::uuid then + update issues set description = '${changedDescription}' where id = '${issueId}'::uuid and company_id = '${companyId}'::uuid; + end if; + return NEW; + end $$`)); + await db.execute(sql.raw(`create trigger ${trigger} before insert on work_assessments + for each row execute function ${trigger}()`)); + } + try { + await finalizeNativeRun({ db, runId, workspaceFinalizeStatus: "succeeded" }); + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]!.status) + .toBe(scenario === "current human rejection" ? "done" : scenario === "model-only decline" ? "in_review" : "in_progress"); + } finally { + if (scenario === "changed description at commit") { + await db.execute(sql.raw(`drop trigger if exists ${trigger} on work_assessments`)); + await db.execute(sql.raw(`drop function if exists ${trigger}()`)); + } + } + }); + + it("rechecks publication removed between assessment and native done commit", async () => { + const objective = "Create a Markdown briefing document on this task."; + const { companyId, agentId, issueId, runId, port, result } = await prepareFirstTaskCompletion("ordinary task", objective); + await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, runId)); + const authority = new PaperclipRunnerToolAuthority(db, { companyId, agentId, issueId, runId }); + await authority.execute({ tool: "write_document", callId: "briefing", arguments: { + idempotencyKey: "briefing", key: "briefing", title: "Briefing", body: "Verified service titles and receipt.", baseRevisionId: null, + } }); + const report = { ...result, evidence: [{ ref: "provider:opencode:agent-message" }], artifacts: [] }; + await port.completeRun({ result: report, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL }); + const trigger = `publication_race_${randomUUID().replaceAll("-", "")}`; + await db.execute(sql.raw(`create function ${trigger}() returns trigger language plpgsql as $$ + begin + if NEW.issue_id = '${issueId}'::uuid and NEW.run_id = '${runId}'::uuid then + delete from issue_documents where issue_id = '${issueId}'::uuid and company_id = '${companyId}'::uuid; + end if; + return NEW; + end $$`)); + try { + await db.execute(sql.raw(`create trigger ${trigger} before insert on work_assessments + for each row execute function ${trigger}()`)); + await finalizeNativeRun({ db, runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]!.status).toBe("in_progress"); + expect(await db.select().from(statusDecisions).where(eq(statusDecisions.issueId, issueId))) + .not.toContainEqual(expect.objectContaining({ toStatus: "done" })); + expect(await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.agentId, agentId)))) + .toContainEqual(expect.objectContaining({ payload: expect.objectContaining({ + continuationIdempotencyKey: "native-completion-incomplete", continuationSummary: expect.stringContaining("write_document"), + }) })); + } finally { + await db.execute(sql.raw(`drop trigger if exists ${trigger} on work_assessments`)); + await db.execute(sql.raw(`drop function if exists ${trigger}()`)); + } + }); + + it.each([false, true])("checks requested publication for an authoritative done decision with an opaque accepted ledger (%s)", async requested => { + const objective = requested ? "Create a Markdown briefing document on this task." : "What is two plus two?"; + const { companyId, agentId, issueId, runId } = await prepareFirstTaskCompletion("ordinary task", objective); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + const [issue] = await db.select().from(issues).where(eq(issues.id, issueId)); + const [saved] = await db.insert(nativeRunResults).values({ companyId, issueId, runId, + completionContractId: run!.completionContractId!, serverFingerprint: randomUUID(), schemaStatus: "accepted", + resultJson: {}, canonicalSha256: randomUUID() }).returning(); + const [assessment] = await db.insert(workAssessments).values({ companyId, issueId, runId, + contractId: run!.completionContractId!, resultId: saved!.id, triggerKind: "native_result", triggerActorCompanyId: companyId, + priorIssueStatus: issue!.status, priorStatusVersion: issue!.statusVersion, policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + assessmentJson: {}, inputDigest: randomUUID() }).returning(); + await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "arbitrating", resultId: saved!.id }); + const commit = commitNativeStatusDecision({ db, companyId, issueId, runId, assessmentId: assessment!.id, + priorStatus: issue!.status as "in_progress", priorStatusVersion: issue!.statusVersion, priorDecisionId: issue!.lastStatusDecisionId, + decision: { policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, statusAction: "done", toStatus: "done", + reasonCode: "completion_contract_satisfied", unblockDescriptor: null, effects: [{ kind: "release_checkout" }] } }); + if (requested) await expect(commit).rejects.toThrow("native_status_race"); + else await expect(commit).resolves.toMatchObject({ issue: { status: "done" } }); + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]!.status).toBe(requested ? "in_progress" : "done"); + expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, saved!.id)))[0]!.resultJson).toEqual({}); + }); + it.each(refusedFirstTaskCompletionScenarios)( "refuses native completion feedback without a trusted current first-task decision (%s)", async (scenario) => { const { issueId, runId, result } = await prepareFirstTaskCompletion(scenario); diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts index 2697d029fe8..3cdb31eea0b 100644 --- a/server/src/services/native-runtime/status-decision-committer.ts +++ b/server/src/services/native-runtime/status-decision-committer.ts @@ -15,6 +15,7 @@ import { notifyInstructionCollection } from "../instruction-collection-notificat import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; import { scheduledIssueMonitorWait } from "../issue-monitors.js"; import { onboardingFirstTaskCompletionAllowed } from "../onboarding-first-task-admission.js"; +import { NativeDeliverableEvidenceError, validateNativeDeliverableEvidence } from "./native-deliverable-feedback.js"; import { hasPublishedGitHubReview, readInheritedGitHubReviewWait } from "../chat-channels/github/review-wait.js"; import { lockNativeGovernanceSources, pendingNativeGovernance } from "./native-governance.js"; import { findSupersededExternalReviewRequests, supersedeExternalReviewRequestsInTransaction } from "./superseded-external-review-requests.js"; @@ -1791,6 +1792,29 @@ export async function commitNativeStatusDecision(input: { if (input.decision.statusAction === "done" && !await onboardingFirstTaskCompletionAllowed(tx as unknown as Db, { companyId: input.companyId, issueId: input.issueId, runId: input.runId, })) throw new NativeStatusRaceError(); + if (input.decision.statusAction === "done") { + try { + const [saved] = await tx.select({ context: heartbeatRuns.contextSnapshot, receipts: heartbeatRuns.resultJson, + envelope: nativeRunResults.resultJson }).from(heartbeatRuns) + .innerJoin(nativeRunResults, and(eq(nativeRunResults.runId, heartbeatRuns.id), + eq(nativeRunResults.id, coordinator.resultId!), eq(nativeRunResults.companyId, input.companyId), + eq(nativeRunResults.issueId, input.issueId), eq(nativeRunResults.schemaStatus, "accepted"))) + .where(and(eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId), + eq(heartbeatRuns.nativeIssueId, input.issueId), eq(heartbeatRuns.runtimeMode, "native"))) + .limit(1).for("share", { of: [heartbeatRuns, nativeRunResults], noWait: true }); + if (!saved) throw new NativeStatusRaceError(); + const continuation = record(saved.context?.executionContinuation); + const objective = typeof continuation.objective === "string" ? continuation.objective + : [issue.title, issue.description].filter(Boolean).join("\n"); + await validateNativeDeliverableEvidence(tx as unknown as Db, { companyId: input.companyId, issueId: input.issueId, + runId: input.runId, objective, semanticToolReceipts: saved.receipts?.semanticToolReceipts, lockSources: true, + authoritativeDone: true }, record(saved.envelope).result); + } catch (error) { + if (error instanceof NativeDeliverableEvidenceError || record(error).code === "55P03" || + record(record(error).cause).code === "55P03") throw new NativeStatusRaceError(); + throw error; + } + } if (reasonCode === "scheduled_monitor_waiting") { const expected = input.requireMonitorWait; const inReview = input.decision.statusAction === "in_review" && input.decision.toStatus === "in_review";