template-app/modules/agent/core/runs.ts

// The agent module's use cases: start a run, read one, list them, cancel one, and the approval
// queue. Every write is the run row and the outbox event in one transaction; the worker's
// subscriber enqueues the run job, and the engine (@conusai-kit/agent) does the durable execution.

import { AGENT_ENTITIES, admitPlan, agentRuns, clampBudget, dataRunStore } from "@conusai-kit/agent"
import { notFound, validationFailed } from "@conusai-kit/core"
import { type ApprovalView, RUN_JOB, type RunView, runResumeRequested } from "../contract/index.js"
import type { AgentDeps, AgentScope } from "./ports.js"

const DEFAULT_BUDGET = { maxCostMicrocents: 100_000_000, maxSteps: 10, maxDurationSeconds: 600 }

const PAGE = 50

export const runViewOf = async (deps: AgentDeps, scope: AgentScope, run: { id: string }): Promise<RunView> => {
  const store = dataRunStore(deps.data, scope)
  const record = await store.get(run.id)
  if (!record) throw notFound("no such run")
  const steps = await store.listSteps(run.id)
  const stepViews = record.plan.map((planned) => {
    const logged = steps.find((candidate) => candidate.key === planned.key)
    return {
      key: planned.key,
      label: planned.label,
      tool: planned.tool,
      args: planned.args,
      status: logged?.status ?? ("pending" as const),
      attempts: logged?.attempts ?? 0,
      error: logged?.error ?? null,
    }
  })
  return {
    runId: record.id,
    goal: record.goal,
    status: record.status,
    failureCode: record.failureCode,
    failureDetail: record.failureDetail,
    plan: stepViews,
    usageCostMicrocents: record.usageCostMicrocents,
    usageSteps: record.usageSteps,
    percent: record.plan.length === 0 ? 0 : Math.min(record.usageSteps, record.plan.length) / record.plan.length,
    cancelRequested: record.cancelRequestedAt !== null,
    actorId: record.actorId,
    createdAt: record.createdAt,
    startedAt: record.startedAt,
    finishedAt: record.finishedAt,
  }
}

/**
 * Starts a run: the plan is admitted against the hard ceilings before anything is written, the run
 * row and the resume event land in one transaction, and the subscriber's enqueue (singleton on
 * the run id) is what actually starts the work. Idempotency is the caller's idempotency key at
 * the transport; a repeated start with the same key answers the first run.
 */
export async function startRun(
  deps: AgentDeps,
  scope: AgentScope,
  input: {
    goal: string
    plan: { key: string; label: string; tool: string; args: Record<string, unknown> }[]
    budget?: {
      maxCostMicrocents?: number | undefined
      maxSteps?: number | undefined
      maxDurationSeconds?: number | undefined
    }
  },
): Promise<RunView> {
  const budget = clampBudget({
    maxCostMicrocents: input.budget?.maxCostMicrocents ?? DEFAULT_BUDGET.maxCostMicrocents,
    maxSteps: input.budget?.maxSteps ?? DEFAULT_BUDGET.maxSteps,
    maxDurationSeconds: input.budget?.maxDurationSeconds ?? DEFAULT_BUDGET.maxDurationSeconds,
  })
  // The plan is data here; the engine's steps come from these rows in the worker.
  admitPlan(
    input.plan.map((planned) => ({
      key: planned.key,
      label: planned.label,
      invoke: () => Promise.resolve({ ok: true as const }),
    })),
    budget,
  )
  const run = await deps.data.unitOfWork.run(scope, async (tx) => {
    const row = await tx.of(agentRuns).insert({
      goal: input.goal,
      plan: input.plan.map((planned) => ({
        key: planned.key,
        label: planned.label,
        tool: planned.tool,
        args: planned.args,
      })),
      status: "queued",
      failureCode: null,
      failureDetail: null,
      actorId: scope.actor.userId,
      budgetCostMicrocents: budget.maxCostMicrocents,
      budgetSteps: budget.maxSteps,
      budgetSeconds: budget.maxDurationSeconds,
      usageCostMicrocents: 0,
      usageSteps: 0,
      cancelRequestedAt: null,
      startedAt: null,
      finishedAt: null,
    })
    await tx.publish(runResumeRequested, {
      orgId: scope.orgId,
      workspaceId: scope.workspaceId,
      projectId: scope.projectId,
      runId: row.id,
    })
    return row
  })
  return runViewOf(deps, scope, run)
}

export async function getRun(deps: AgentDeps, scope: AgentScope, runId: string): Promise<RunView> {
  return runViewOf(deps, scope, { id: runId })
}

export async function listRuns(
  deps: AgentDeps,
  scope: AgentScope,
  input: { cursor?: string | undefined; limit?: number | undefined },
): Promise<{ runs: RunView[]; cursor?: string; hasMore: boolean }> {
  const store = dataRunStore(deps.data, scope, { pageSize: PAGE })
  const page = await store.list({
    ...(input.cursor === undefined ? {} : { cursor: input.cursor }),
    ...(input.limit === undefined ? {} : { limit: input.limit }),
  })
  return {
    runs: await Promise.all(page.runs.map((run) => runViewOf(deps, scope, run))),
    ...("cursor" in page ? { cursor: page.cursor } : {}),
    hasMore: page.hasMore,
  }
}

/** Cancels a run: the persisted request is honored between steps and inside the in-flight call. */
export async function cancelRun(
  deps: AgentDeps,
  scope: AgentScope,
  runId: string,
): Promise<{ runId: string; cancelled: boolean }> {
  const store = dataRunStore(deps.data, scope)
  const record = await store.get(runId)
  if (!record) throw notFound("no such run")
  if (record.cancelRequestedAt !== null) return { runId, cancelled: true }
  await store.requestCancel(runId, new Date().toISOString())
  // A parked run only notices the request when it is resumed; enqueue the resume so the cancel
  // lands even while the run waits in awaiting_approval.
  await deps.data.unitOfWork.run(scope, async (tx) => {
    await tx.publish(runResumeRequested, {
      orgId: scope.orgId,
      workspaceId: scope.workspaceId,
      projectId: scope.projectId,
      runId,
    })
  })
  return { runId, cancelled: true }
}

/**
 * The approval queue: every step that parked an approval, with the EXACT stored proposed
 * arguments. Nothing here is a model's narration; the arguments are what the tool was called
 * with, copied before they were proposed.
 */
export async function listApprovals(deps: AgentDeps, scope: AgentScope): Promise<{ approvals: ApprovalView[] }> {
  const approvals: ApprovalView[] = []
  let cursor: string | undefined
  for (;;) {
    const page = await deps.data
      .forScope(scope)
      .of(agentRuns)
      .list({
        order: [{ field: "createdAt", direction: "desc" }],
        limit: PAGE,
        ...(cursor === undefined ? {} : { cursor }),
      })
    for (const run of page.items) {
      const steps = await dataRunStore(deps.data, scope).listSteps(run.id)
      for (const step of steps) {
        if (step.approval === null) continue
        approvals.push({
          runId: run.id,
          stepKey: step.key,
          goal: run.goal,
          tool: step.approval.tool,
          arguments: step.approval.arguments,
          reviewUrl: step.approval.reviewUrl,
          createdAt: run.createdAt,
        })
      }
    }
    if (!page.hasMore) break
    cursor = page.cursor
  }
  return { approvals }
}

/**
 * Records a person's decision and resumes the run. Approval re-enters the parked step: the tools
 * adapter re-calls with the step's idempotency key, and the platform executes the approved,
 * persisted proposal exactly once. Rejection cancels the run: the proposal will not be executed.
 */
export async function reviewApproval(
  deps: AgentDeps,
  scope: AgentScope,
  input: { runId: string; stepKey: string; decision: "approve" | "reject" },
): Promise<{ runId: string; status: RunView["status"] }> {
  const store = dataRunStore(deps.data, scope)
  const record = await store.get(input.runId)
  if (!record) throw notFound("no such run")
  const steps = await store.listSteps(input.runId)
  const step = steps.find((candidate) => candidate.key === input.stepKey)
  if (!step || step.approval === null) {
    throw validationFailed([{ path: "stepKey", message: "this step has no parked approval" }], "nothing to review")
  }
  if (input.decision === "reject") {
    await store.requestCancel(input.runId, new Date().toISOString())
  }
  await deps.data.unitOfWork.run(scope, async (tx) => {
    await tx.publish(runResumeRequested, {
      orgId: scope.orgId,
      workspaceId: scope.workspaceId,
      projectId: scope.projectId,
      runId: input.runId,
    })
  })
  return { runId: input.runId, status: (await store.get(input.runId))?.status ?? record.status }
}

export { AGENT_ENTITIES, RUN_JOB }