From 0791da0cc5458c11da9d10f000645036ffd6f4e2 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 28 Oct 2026 11:09:49 -0500 Subject: [PATCH] fix(server): synchronize routine history during native status commit (#18033) https://entire.com/gh/paperclipai/-/trails/173/changes/paperclip/173 ## Thinking Path > - Paperclip manages work assigned to AI agents. > - A routine stores each invocation and its execution task. > - Native completion commits task status without using the public issue route. > - That route synchronizes routine history, but the native status transaction did not. > - A successful task could remain listed as an unfinished routine invocation on a local instance. > - This pull request uses the existing database projection in the native status transaction. > - Task status and routine history now commit or roll back together. ## Linked Issues or Issue Description **What happened?** A real routine execution completed successfully. Its task had status `done` and retained its routine origin and invocation identity. Public routine history still returned `issue_created`, with `completedAt: null` and a linked task whose status was `done`. The existing five-minute reconciliation runs only on cloud-managed instances. It does not close this path for a local instance. **Expected behavior** An authoritative native task status change must update the matching routine invocation through the existing routine status rules. It must not update another task, routine or company. **Steps to reproduce** 1. Create a routine on a local instance and run it through the public routine API. 2. Let its native execution finish and commit the task as done. 3. Read the routine run history. Before this change, the invocation can remain `issue_created` while its linked task is done. **Paperclip version or commit** The retained failure used merged production `39e811c9c5`. The missing hook is also present on this branch's base, `8049fbf51d`. Related: #16633 contains the existing cloud-only reconciliation. A bounded search found no duplicate native synchronization fix. ## What Changed - Synchronize routine history inside the native status transaction after the task status projection. - Require matching company, invocation and linked task before synchronization. Preserve supported custom display origins. - Add database-backed tests for terminal and active states, replay, invalid custody, rollback and a real plugin-managed routine dispatch with a custom origin. - Keep the routine status projection in a database-only helper. Both the routine service and native status committer use it without loading heartbeat execution from the committer. ## Verification - Own-source server TypeScript check passed after the helper extraction. It includes the new test source and production dependencies. Local package exports point at another checkout, so the check used explicit paths to this checkout's source and disabled declaration emit. - Existing pure `status-arbiter.test.ts`: 20 tests passed. `git diff --check` passed. - Initial required CI run 40711603938 built and typechecked successfully. It executed all ten new database tests: nine passed; the public-dispatch fixture failed because it assumed status version zero. The fixture now reads the saved version and decision binding. - That CI run also exposed a heartbeat mock dependency cycle. The database-only helper removes the committer-to-routine-service-to-heartbeat import. Required CI on the new head must confirm both repairs. - Initial browser shard 3/8 failed during cleanup: legacy run cancellation waited for a heartbeat row lock. Its cause is under investigation; the failure remains recorded. - A local runtime-execution check failed during module import from stale package exports, before any tests or database setup ran. It is not passing evidence. No local database, provider or live eval was run. - Latest-head required CI run 40717315212 executed all ten new database tests successfully in server shard 5/12 (2,163 passed; three unrelated skips). The heartbeat runtime suite executed 23 tests successfully in server shard 9/12 (1,704 passed). - Latest-head browser shard 3/8 passed all 33 tests. This does not establish the cause of the preserved earlier cleanup lock timeout. - Required CI run 40717315212 completed successfully on head `2d77f4302b36303741abad07266bbe9e376ba6a5`. All 48 workflow jobs are green after one failed-jobs-only infrastructure retry. The first attempt had seven explicit runner shutdowns and three lost-runner communication failures. The exact chat shard host was independently confirmed as an AWS Spot termination. Original logs and annotations remain retained. - Greptile completed at 5/5 on this exact head, with no outstanding findings. All review threads are resolved. The independent follow-up review found no issues. - No live qualification or corrected eval pass is claimed. The PR remains open and unmerged, with auto-merge off. ## Risks - Matching routine executions perform the existing bounded status synchronization queries inside the status transaction. A database failure rolls back both projections. - Inconsistent invocation or linked-task metadata now prevents synchronization. No unrelated invocation is inferred or repaired. - The existing scheduled repair remains unchanged. This PR does not change accounting, provider behavior or eval budgets. ## Model Used - OpenAI `gpt-6.1-sol`, with reasoning and tool-assisted source inspection and coding. The session did not supply a context-window size. ## 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 references) - [x] My branch name describes the change and contains no internal ticket id or instance-derived details - [x] I have run relevant credential-free tests locally and they pass; database coverage is pending CI - [x] I have added or updated tests where applicable - [x] I have checked documentation; the existing routine status contract is unchanged - [x] I have considered and documented 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 --------- Co-authored-by: Paperclip --- .../native-routine-status.test.ts | 92 ++++++++++++ .../status-decision-committer.ts | 4 + server/src/services/routine-run-status.ts | 137 ++++++++++++++++++ server/src/services/routines.ts | 128 +--------------- 4 files changed, 236 insertions(+), 125 deletions(-) create mode 100644 server/src/services/native-runtime/native-routine-status.test.ts create mode 100644 server/src/services/routine-run-status.ts 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); }