/** * Adapter-level execution wrapper for background routines with self-healing retries, * delta evaluation, and quiet notification delivery. */ import { runContinueJob } from "@rakazo/adapter-kit"; import type { RoutineNotifyMode } from "@rakazo/core"; import { calculateRoutineBackoffMs, evaluateRoutineNotificationPolicy, formatMinimalRoutineNotification, isTransientRoutineError, } from "@rakazo/core"; import { getLogger } from "@rakazo/logging"; import { notifyRun } from "./executor-computer.js"; import type { ExecutorDeps } from "./executor-types.js"; export const MAX_ROUTINE_ATTEMPTS = 3; export interface RoutineExecutionFinalizeContext { deps: ExecutorDeps; run: { id: string; spaceId: string; botId: string; threadId: string; userId: string; routineId: string | null; }; bot: { id: string; name: string }; routine: { id: string; name: string; notify: boolean; }; notifyMode: RoutineNotifyMode; outputText: string; hasChanged: boolean; deltaSummary?: string; } /** * Evaluates quiet policy and dispatches notifications only if allowed. */ export async function finalizeRoutineNotification( context: RoutineExecutionFinalizeContext, ): Promise<{ markUnread: boolean; isQuiet: boolean }> { const { deps, run, bot, routine, notifyMode, outputText, hasChanged, deltaSummary } = context; const decision = evaluateRoutineNotificationPolicy({ routineNotifyEnabled: routine.notify, notifyMode, hasChanged, outcome: "completed", }); if (decision.shouldNotify) { const payload = formatMinimalRoutineNotification({ botName: bot.name, routineName: routine.name, outcome: "completed", deltaSummary: deltaSummary || outputText, }); await notifyRun(deps, run, { kind: "completion", title: payload.title, body: payload.body, botId: bot.id, threadId: run.threadId, }).catch((error) => { getLogger().error("routine notification dispatch failed", error); }); } return { markUnread: decision.markUnread, isQuiet: decision.isQuiet, }; } /** * Handles routine failure notification respecting quiet policy. */ export async function finalizeRoutineFailureNotification( deps: ExecutorDeps, run: { spaceId: string; userId: string; botId: string; threadId: string; routineId: string | null; }, bot: { id: string; name: string }, routine: { id: string; name: string; notify: boolean }, notifyMode: RoutineNotifyMode, error: string, ): Promise { const decision = evaluateRoutineNotificationPolicy({ routineNotifyEnabled: routine.notify, notifyMode, hasChanged: false, outcome: "failed", }); if (decision.shouldNotify) { const payload = formatMinimalRoutineNotification({ botName: bot.name, routineName: routine.name, outcome: "failed", error, }); await notifyRun(deps, run, { kind: "failure", title: payload.title, body: payload.body, botId: bot.id, threadId: run.threadId, }).catch((err) => { getLogger().error("routine failure notification dispatch failed", err); }); return true; } return false; } /** * Handles transient error recovery for routine runs. * Returns true if the run was requeued for self-healing retry; false if retries exhausted * or error is permanent. */ export async function handleRoutineSelfHealing( deps: ExecutorDeps, runId: string, attemptNumber: number, error: unknown, ): Promise { if (!isTransientRoutineError(error)) { return false; } if (attemptNumber >= MAX_ROUTINE_ATTEMPTS) { getLogger().warn( `Routine run ${runId} exhausted self-healing retries (${attemptNumber}/${MAX_ROUTINE_ATTEMPTS})`, ); return false; } const backoffMs = calculateRoutineBackoffMs(attemptNumber); const nextAttemptAt = new Date(Date.now() + backoffMs); getLogger().info( `Self-healing transient error for routine run ${runId} (attempt ${attemptNumber}/${MAX_ROUTINE_ATTEMPTS}). Retrying in ${backoffMs}ms`, ); await deps.prisma.run.update({ where: { id: runId }, data: { status: "queued", leaseOwner: null, leaseExpiresAt: null, error: `Transient failure: ${error instanceof Error ? error.message : String(error)} (retrying)`, }, }); await deps.jobs.enqueue({ ...runContinueJob(runId), availableAt: nextAttemptAt, }); return true; }