diff --git a/server/src/services/native-runtime/native-routine-status.test.ts b/server/src/services/native-runtime/native-routine-status.test.ts new file mode 100644 index 00000000000..668de83a4d5 --- /dev/null +++ b/server/src/services/native-runtime/native-routine-status.test.ts @@ -0,0 +1,92 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { agents, companies, completionContracts, createDb, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, pluginManagedResources, plugins, routineRuns, routines, workAssessments } from "@paperclipai/db"; +import type { Db } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { routineService } from "../routines.js"; +import { commitNativeStatusDecision } from "./status-decision-committer.js"; +import { NATIVE_STATUS_ARBITER_POLICY_VERSION, type NativeStatusDecision } from "./status-arbiter.js"; + +const support = await getEmbeddedPostgresTestSupport(); +if (!support.supported) console.warn(`Skipping native routine status tests: ${support.reason}`); +(support.supported ? describe : describe.skip)("native routine history status projection", () => { + let db: ReturnType, database: Awaited>; + const companyId = randomUUID(), agentId = randomUUID(); + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("native-routine-status-"); db = createDb(database.connectionString); + await db.insert(companies).values({ id: companyId, name: "Routine status fixture", issuePrefix: "NRS" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Routine worker", adapterType: "paperclip_runner", status: "running" }); + }, 60_000); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); + async function seed(mode = "matching") { + let issueId: string = randomUUID(), invocationId: string = randomUUID(); + const runId = randomUUID(), contractId = randomUUID(), resultId = randomUUID(), assessmentId = randomUUID(); + const [routine] = await db.insert(routines).values({ companyId, title: "Daily summary", assigneeAgentId: agentId }).returning(); + const otherCompanyId = randomUUID(); + if (mode === "foreign-company") await db.insert(companies).values({ id: otherCompanyId, name: "Other fixture", issuePrefix: `X${otherCompanyId.slice(0, 5)}` }); + if (mode === "custom-origin") { + const pluginKey = `paperclip.routine-status-${randomUUID()}`, originId = `operation:${randomUUID()}`; + const [plugin] = await db.insert(plugins).values({ pluginKey, packageName: pluginKey, version: "0.1.0", + manifestJson: { id: pluginKey, apiVersion: 1, version: "0.1.0", displayName: "Routine fixture", description: "Routine fixture", + author: "Paperclip", categories: ["automation"], capabilities: ["routines.managed"], entrypoints: { worker: "./worker.js" } } }).returning(); + await db.insert(pluginManagedResources).values({ companyId, pluginId: plugin!.id, pluginKey, resourceKind: "routine", resourceKey: "summary", + resourceId: routine!.id, defaultsJson: { issueTemplate: { originId } } }); + const invocation = await routineService(db, { heartbeat: { wakeup: async () => undefined } }).runRoutine(routine!.id, { source: "manual", idempotencyKey: randomUUID() }); + invocationId = invocation.id; issueId = invocation.linkedIssueId!; + const [created] = await db.select().from(issues).where(eq(issues.id, issueId)); + expect(created).toMatchObject({ originKind: "routine_execution", originId, originRunId: invocationId }); + expect(created.originId).not.toBe(routine!.id); + await db.update(issues).set({ status: "in_progress" }).where(eq(issues.id, issueId)); + } else { + await db.insert(issues).values({ id: issueId, companyId, title: "Summary execution", status: "in_progress", workMode: "standard", assigneeAgentId: agentId, + originKind: mode === "ordinary" ? "manual" : "routine_execution", originId: routine!.id, originRunId: mode === "wrong-invocation" ? randomUUID() : invocationId }); + await db.insert(routineRuns).values({ id: invocationId, companyId: mode === "foreign-company" ? otherCompanyId : companyId, + routineId: routine!.id, linkedIssueId: mode === "wrong-issue" ? null : issueId, source: "manual", status: "issue_created", triggerPayload: { retainedInput: "Launch" } }); + } + const [priorIssue] = await db.select().from(issues).where(eq(issues.id, issueId)); + await db.insert(completionContracts).values({ id: contractId, companyId, issueId, revision: 1, schemaVersion: "paperclip.completion-contract.v1", + policyVersion: "routine-status-test", risk: "standard", completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", + contractJson: { revision: "1", objective: "Save summary", criteria: [] }, canonicalSha256: `contract-${contractId}`, createdByActorType: "system", createdByActorId: "test" }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", runtimeMode: "native", nativeIssueId: issueId, + completionContractId: contractId, completionContractSha256: `contract-${contractId}`, contextSnapshot: { issueId } }); + await db.insert(nativeRunResults).values({ id: resultId, companyId, issueId, runId, completionContractId: contractId, + serverFingerprint: `fp-${resultId}`, schemaStatus: "accepted", resultJson: {}, canonicalSha256: `sha-${resultId}` }); + await db.insert(workAssessments).values({ id: assessmentId, companyId, issueId, runId, contractId, resultId, triggerKind: "native_result", + triggerActorCompanyId: companyId, priorIssueStatus: "in_progress", priorStatusVersion: priorIssue.statusVersion, policyVersion: "routine-status-test", assessmentJson: {}, inputDigest: `assessment-${assessmentId}` }); + await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "arbitrating", resultId }); + return { issueId, invocationId, runId, assessmentId, priorStatusVersion: priorIssue.statusVersion, priorDecisionId: priorIssue.lastStatusDecisionId }; + } + async function commit(f: Awaited>, status: "done" | "blocked" | "cancelled" | "in_progress", executor: Db = db) { + const decision: NativeStatusDecision = { policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, statusAction: status, toStatus: status, + reasonCode: "routine_status_test", unblockDescriptor: null, effects: [] }; + return commitNativeStatusDecision({ db: executor, companyId, issueId: f.issueId, runId: f.runId, assessmentId: f.assessmentId, + priorStatus: "in_progress", priorStatusVersion: f.priorStatusVersion, priorDecisionId: f.priorDecisionId, decision }); + } + it.each(["done", "blocked", "cancelled", "in_progress"] as const)("projects real %s task status into the exact routine invocation", async status => { + const f = await seed(); const first = await commit(f, status); + const [saved] = await db.select().from(routineRuns).where(eq(routineRuns.id, f.invocationId)); + expect(saved).toMatchObject({ linkedIssueId: f.issueId, status: status === "done" ? "completed" : status === "in_progress" ? "issue_created" : "failed", triggerPayload: expect.objectContaining({ retainedInput: "Launch" }) }); + expect(saved.completedAt !== null).toBe(status !== "in_progress"); + const replay = await commit(f, status); expect(replay.replayed).toBe(true); expect(replay.decision.id).toBe(first.decision.id); + expect((await db.select().from(routineRuns).where(eq(routineRuns.id, f.invocationId)))[0]).toEqual(saved); + }); + it.each(["ordinary", "wrong-invocation", "wrong-issue", "foreign-company"])("does not settle an invocation with %s custody", async mode => { + const f = await seed(mode); const [before] = await db.select().from(routineRuns).where(eq(routineRuns.id, f.invocationId)); + await commit(f, "done"); + expect((await db.select().from(routineRuns).where(eq(routineRuns.id, f.invocationId)))[0]).toEqual(before); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("done"); + }); + it("settles a publicly dispatched plugin-managed invocation with its supported custom origin", async () => { + const f = await seed("custom-origin"); await commit(f, "done"); + expect((await db.select().from(routineRuns).where(eq(routineRuns.id, f.invocationId)))[0]).toMatchObject({ linkedIssueId: f.issueId, status: "completed", completedAt: expect.any(Date) }); + }); + it("rolls back the task and routine projection together", async () => { + const f = await seed(); + await expect(db.transaction(async tx => { await commit(f, "done", tx as unknown as Db); + expect((await tx.select().from(routineRuns).where(eq(routineRuns.id, f.invocationId)))[0].status).toBe("completed"); + throw new Error("rollback fixture"); })).rejects.toThrow("rollback fixture"); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0].status).toBe("in_progress"); + expect((await db.select().from(routineRuns).where(eq(routineRuns.id, f.invocationId)))[0].status).toBe("issue_created"); + }); +}); diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts index 9c74f4d765c..b0d0913a8b3 100644 --- a/server/src/services/native-runtime/status-decision-committer.ts +++ b/server/src/services/native-runtime/status-decision-committer.ts @@ -53,6 +53,7 @@ import { resolveExternalChatResponseWaitAuthorizationInTransaction, } from "./chat-attachment-reuse.js"; import { issueService } from "../issues.js"; +import { syncRoutineRunStatusForIssue } from "../routine-run-status.js"; import { issueThreadInteractionService } from "../issue-thread-interactions.js"; import { issueRecoveryActionService } from "../issue-recovery-actions.js"; import { buildIssueBlockersResolvedWakeIdempotencyKey } from "../issue-dependency-wakeups.js"; @@ -2115,6 +2116,9 @@ export async function commitNativeStatusDecision(input: { ); if (!projected) throw new NativeStatusRaceError(); updated = projected; + if (updated.originKind === "routine_execution") { + await syncRoutineRunStatusForIssue(tx as unknown as Db, input.issueId); + } materialized.unshift({ effectKind: "issue_status_projection", targetType: "issue", diff --git a/server/src/services/routine-run-status.ts b/server/src/services/routine-run-status.ts new file mode 100644 index 00000000000..77341f9cba4 --- /dev/null +++ b/server/src/services/routine-run-status.ts @@ -0,0 +1,137 @@ +import { and, eq } from "drizzle-orm"; +import type { Db } from "@paperclipai/db"; +import { issues, routineRuns, routines } from "@paperclipai/db"; + +const EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE = "execution_issue_status"; +const EXECUTION_ISSUE_TRANSIENT_FAILURE_STATUSES = ["blocked", "cancelled"] as const; + +type ExecutionIssueTransientFailureStatus = (typeof EXECUTION_ISSUE_TRANSIENT_FAILURE_STATUSES)[number]; + +function executionIssueTransientFailureReason(status: ExecutionIssueTransientFailureStatus) { + return `Execution issue moved to ${status}`; +} + +function executionIssueTransientFailureStatusFromPayload(payload: unknown): ExecutionIssueTransientFailureStatus | null { + if (!payload || typeof payload !== "object" || Array.isArray(payload)) return null; + const transientFailure = (payload as Record).transientFailure; + if (!transientFailure || typeof transientFailure !== "object" || Array.isArray(transientFailure)) return null; + const record = transientFailure as Record; + if (record.code !== EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE) return null; + return EXECUTION_ISSUE_TRANSIENT_FAILURE_STATUSES.find((status) => record.status === status) ?? null; +} + +function executionIssueTransientFailureClearedAtFromPayload(payload: unknown): string | null { + if (!payload || typeof payload !== "object" || Array.isArray(payload)) return null; + const transientFailure = (payload as Record).transientFailure; + if (!transientFailure || typeof transientFailure !== "object" || Array.isArray(transientFailure)) return null; + const clearedAt = (transientFailure as Record).clearedAt; + return typeof clearedAt === "string" ? clearedAt : null; +} + +function legacyExecutionIssueTransientFailureStatus( + failureReason: string | null, +): ExecutionIssueTransientFailureStatus | null { + return EXECUTION_ISSUE_TRANSIENT_FAILURE_STATUSES.find( + (status) => failureReason === executionIssueTransientFailureReason(status), + ) ?? null; +} + +export async function finalizeRoutineRun(executor: Db, runId: string, patch: Partial) { + return executor + .update(routineRuns) + .set({ + ...patch, + updatedAt: new Date(), + }) + .where(eq(routineRuns.id, runId)) + .returning() + .then((rows) => rows[0] ?? null); +} + +export async function syncRoutineRunStatusForIssue(executor: Db, issueId: string) { + const issue = await executor + .select({ + id: issues.id, + companyId: issues.companyId, + status: issues.status, + originKind: issues.originKind, + originRunId: issues.originRunId, + }) + .from(issues) + .where(eq(issues.id, issueId)) + .then((rows) => rows[0] ?? null); + if (!issue || issue.originKind !== "routine_execution" || !issue.originRunId) return null; + const run = await executor + .select({ + id: routineRuns.id, + status: routineRuns.status, + failureReason: routineRuns.failureReason, + triggerPayload: routineRuns.triggerPayload, + }) + .from(routineRuns) + .innerJoin(routines, and(eq(routines.id, routineRuns.routineId), eq(routines.companyId, issue.companyId))) + .where(and(eq(routineRuns.id, issue.originRunId), eq(routineRuns.companyId, issue.companyId), + eq(routineRuns.linkedIssueId, issue.id))) + .then((rows) => rows[0] ?? null); + if (!run) return null; + if (issue.status === "done") { + const transientFailureStatus = executionIssueTransientFailureStatusFromPayload(run.triggerPayload) + ?? legacyExecutionIssueTransientFailureStatus(run.failureReason); + const transientFailureClearedAt = executionIssueTransientFailureClearedAtFromPayload(run.triggerPayload); + return finalizeRoutineRun(executor, issue.originRunId, { + status: "completed", + failureReason: null, + completedAt: new Date(), + ...(transientFailureStatus + ? { + triggerPayload: { + ...(run.triggerPayload ?? {}), + transientFailure: { + code: EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE, + status: transientFailureStatus, + reason: executionIssueTransientFailureReason(transientFailureStatus), + clearedAt: transientFailureClearedAt ?? new Date().toISOString(), + }, + }, + } + : {}), + }); + } + if (issue.status === "blocked" || issue.status === "cancelled") { + const failureReason = executionIssueTransientFailureReason(issue.status); + return finalizeRoutineRun(executor, issue.originRunId, { + status: "failed", + failureReason, + completedAt: new Date(), + triggerPayload: { + ...(run.triggerPayload ?? {}), + transientFailure: { + code: EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE, + status: issue.status, + reason: failureReason, + recordedAt: new Date().toISOString(), + }, + }, + }); + } + const transientFailureStatus = executionIssueTransientFailureStatusFromPayload(run.triggerPayload) + ?? legacyExecutionIssueTransientFailureStatus(run.failureReason); + if (run.status === "failed" && transientFailureStatus) { + return finalizeRoutineRun(executor, issue.originRunId, { + status: "issue_created", + failureReason: null, + completedAt: null, + triggerPayload: { + ...(run.triggerPayload ?? {}), + transientFailure: { + code: EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE, + status: transientFailureStatus, + reason: executionIssueTransientFailureReason(transientFailureStatus), + clearedAt: new Date().toISOString(), + }, + }, + }); + } + return null; +} + diff --git a/server/src/services/routines.ts b/server/src/services/routines.ts index 9ea1884369c..bea7a1488c6 100644 --- a/server/src/services/routines.ts +++ b/server/src/services/routines.ts @@ -73,6 +73,7 @@ import { secretService } from "./secrets.js"; import { getSecretProvider } from "../secrets/provider-registry.js"; import { parseCron, validateCron } from "./cron.js"; import { heartbeatService } from "./heartbeat.js"; +import { finalizeRoutineRun, syncRoutineRunStatusForIssue } from "./routine-run-status.js"; import { instanceSettingsService, isTruthyRuntimeEnvValue, @@ -89,8 +90,6 @@ const LIVE_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_retry"]; const TERMINAL_ISSUE_STATUSES = new Set(["done", "cancelled"]); const MAX_CATCH_UP_RUNS = 25; const MAX_ROUTINE_REVISIONS = 100; -const EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE = "execution_issue_status"; -const EXECUTION_ISSUE_TRANSIENT_FAILURE_STATUSES = ["blocked", "cancelled"] as const; const ACTIVITY_GATE_IGNORED_ACTIONS = [ "issue.read_marked", "issue.read_unmarked", @@ -114,37 +113,6 @@ export function routineWebhookUrl(publicId: string): string { return `${baseUrl.replace(/\/+$/, "")}/api/routine-triggers/public/${publicId}/fire`; } -type ExecutionIssueTransientFailureStatus = (typeof EXECUTION_ISSUE_TRANSIENT_FAILURE_STATUSES)[number]; - -function executionIssueTransientFailureReason(status: ExecutionIssueTransientFailureStatus) { - return `Execution issue moved to ${status}`; -} - -function executionIssueTransientFailureStatusFromPayload(payload: unknown): ExecutionIssueTransientFailureStatus | null { - if (!payload || typeof payload !== "object" || Array.isArray(payload)) return null; - const transientFailure = (payload as Record).transientFailure; - if (!transientFailure || typeof transientFailure !== "object" || Array.isArray(transientFailure)) return null; - const record = transientFailure as Record; - if (record.code !== EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE) return null; - return EXECUTION_ISSUE_TRANSIENT_FAILURE_STATUSES.find((status) => record.status === status) ?? null; -} - -function executionIssueTransientFailureClearedAtFromPayload(payload: unknown): string | null { - if (!payload || typeof payload !== "object" || Array.isArray(payload)) return null; - const transientFailure = (payload as Record).transientFailure; - if (!transientFailure || typeof transientFailure !== "object" || Array.isArray(transientFailure)) return null; - const clearedAt = (transientFailure as Record).clearedAt; - return typeof clearedAt === "string" ? clearedAt : null; -} - -function legacyExecutionIssueTransientFailureStatus( - failureReason: string | null, -): ExecutionIssueTransientFailureStatus | null { - return EXECUTION_ISSUE_TRANSIENT_FAILURE_STATUSES.find( - (status) => failureReason === executionIssueTransientFailureReason(status), - ) ?? null; -} - async function resolveCompanyDefaultResponsibleUserId(db: Db, companyId: string) { const company = await db .select({ defaultResponsibleUserId: companies.defaultResponsibleUserId }) @@ -1591,15 +1559,7 @@ export function routineService( } async function finalizeRun(runId: string, patch: Partial, executor: Db = db) { - return executor - .update(routineRuns) - .set({ - ...patch, - updatedAt: new Date(), - }) - .where(eq(routineRuns.id, runId)) - .returning() - .then((rows) => rows[0] ?? null); + return finalizeRoutineRun(executor, runId, patch); } async function createWebhookSecret( @@ -2023,89 +1983,7 @@ export function routineService( let statusReconcileCursor: string | null = null; async function syncRunStatusForIssue(issueId: string, executor: Db = db) { - const issue = await executor - .select({ - id: issues.id, - companyId: issues.companyId, - status: issues.status, - originKind: issues.originKind, - originRunId: issues.originRunId, - }) - .from(issues) - .where(eq(issues.id, issueId)) - .then((rows) => rows[0] ?? null); - if (!issue || issue.originKind !== "routine_execution" || !issue.originRunId) return null; - const run = await executor - .select({ - id: routineRuns.id, - status: routineRuns.status, - failureReason: routineRuns.failureReason, - triggerPayload: routineRuns.triggerPayload, - }) - .from(routineRuns) - .innerJoin(routines, and(eq(routines.id, routineRuns.routineId), eq(routines.companyId, issue.companyId))) - .where(and(eq(routineRuns.id, issue.originRunId), eq(routineRuns.companyId, issue.companyId))) - .then((rows) => rows[0] ?? null); - if (!run) return null; - if (issue.status === "done") { - const transientFailureStatus = executionIssueTransientFailureStatusFromPayload(run.triggerPayload) - ?? legacyExecutionIssueTransientFailureStatus(run.failureReason); - const transientFailureClearedAt = executionIssueTransientFailureClearedAtFromPayload(run.triggerPayload); - return finalizeRun(issue.originRunId, { - status: "completed", - failureReason: null, - completedAt: new Date(), - ...(transientFailureStatus - ? { - triggerPayload: { - ...(run.triggerPayload ?? {}), - transientFailure: { - code: EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE, - status: transientFailureStatus, - reason: executionIssueTransientFailureReason(transientFailureStatus), - clearedAt: transientFailureClearedAt ?? new Date().toISOString(), - }, - }, - } - : {}), - }, executor); - } - if (issue.status === "blocked" || issue.status === "cancelled") { - const failureReason = executionIssueTransientFailureReason(issue.status); - return finalizeRun(issue.originRunId, { - status: "failed", - failureReason, - completedAt: new Date(), - triggerPayload: { - ...(run.triggerPayload ?? {}), - transientFailure: { - code: EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE, - status: issue.status, - reason: failureReason, - recordedAt: new Date().toISOString(), - }, - }, - }, executor); - } - const transientFailureStatus = executionIssueTransientFailureStatusFromPayload(run.triggerPayload) - ?? legacyExecutionIssueTransientFailureStatus(run.failureReason); - if (run.status === "failed" && transientFailureStatus) { - return finalizeRun(issue.originRunId, { - status: "issue_created", - failureReason: null, - completedAt: null, - triggerPayload: { - ...(run.triggerPayload ?? {}), - transientFailure: { - code: EXECUTION_ISSUE_TRANSIENT_FAILURE_CODE, - status: transientFailureStatus, - reason: executionIssueTransientFailureReason(transientFailureStatus), - clearedAt: new Date().toISOString(), - }, - }, - }, executor); - } - return null; + return syncRoutineRunStatusForIssue(executor, issueId); }