import { createLanguageModelFromResolved } from "@tasks/ai"; import { modelProviders, objects, roleModelRoutes, workspaceMembers, workspaces, } from "@tasks/database/schema"; import { convoyClassificationSchema, convoyObjectContentSchema, ConvoyTransitionError, GATE_INBOX_STATUSES, humanGateActorForStatus, MISSION_CONTROL_KANBAN_COLUMNS, type ConvoyStatus, } from "@tasks/shared/convoy-lifecycle"; import { providerTypeSchema, resolveModel, type ModelProvider, type RoleModelRoute, } from "@tasks/shared/model-routing"; import { TRPCError } from "@trpc/server"; import { generateText } from "ai"; import { asc, eq, inArray } from "drizzle-orm"; import { z } from "zod"; import { findConvoyBySlug, getConvoyBriefs, getConvoyCheckpointSummary, getRecentConvoyEvents, insertConvoyEvent, listConvoyProjects, loadConvoyProject, transitionConvoy, } from "@/server/lib/convoy/convoy-service"; import { protectedProcedure, router, workspaceProcedure } from "@/server/trpc"; import type { WorkspaceProcedureContext } from "@/server/trpc"; type ProviderRow = typeof modelProviders.$inferSelect; type RouteRow = typeof roleModelRoutes.$inferSelect; function mapProviderRow(row: ProviderRow): ModelProvider { return { id: row.id, workspaceId: row.workspaceId, alias: row.alias, displayName: row.displayName, providerType: providerTypeSchema.parse(row.providerType), modelId: row.modelId, baseUrl: row.baseUrl, apiKeyEnvVar: row.apiKeyEnvVar, extraHeaders: row.extraHeaders, defaultParams: row.defaultParams, metadata: row.metadata, enabled: row.enabled, }; } function mapRouteRow(row: RouteRow): RoleModelRoute { return { id: row.id, workspaceId: row.workspaceId, roleId: row.roleId, providerAliasId: row.providerAliasId, enabled: row.enabled, sortOrder: row.sortOrder, }; } function assertWorkspaceAdmin(ctx: WorkspaceProcedureContext): void { if (ctx.workspace.role !== "owner" && ctx.workspace.role !== "admin") { throw new TRPCError({ code: "FORBIDDEN", message: "Only workspace owners and admins can manage convoys.", }); } } function readApiKey(envVarName: string): string | undefined { return process.env[envVarName]?.trim() || undefined; } const gateActionSchema = z.object({ workspace: z.string().min(1), convoyId: z.string().uuid(), action: z.enum(["approve", "revise", "request_changes", "promote"]), reason: z.string().min(1).max(2_000).optional(), }); function resolveGateTransition( status: ConvoyStatus, action: z.infer["action"], ): { actor: string; toStatus: ConvoyStatus } { const actor = humanGateActorForStatus(status); if (!actor) { throw new TRPCError({ code: "BAD_REQUEST", message: `No human gate applies to status ${status}`, }); } if (status === "convoy_review") { if (action === "approve") return { actor, toStatus: "ready" }; if (action === "revise") return { actor, toStatus: "draft" }; } if (status === "pr_review") { if (action === "approve") return { actor, toStatus: "merged" }; if (action === "request_changes") return { actor, toStatus: "in_progress" }; } if (status === "doc_review") { if (action === "approve") return { actor, toStatus: "published" }; if (action === "request_changes") return { actor, toStatus: "in_progress" }; } if ((status === "merged" || status === "published") && action === "promote") { return { actor, toStatus: "released" }; } throw new TRPCError({ code: "BAD_REQUEST", message: `Action ${action} is not valid for status ${status}`, }); } export const convoysRouter = router({ kanban: workspaceProcedure.query(async ({ ctx }) => { const convoys = await listConvoyProjects(ctx.db, ctx.workspace.id); const columns = MISSION_CONTROL_KANBAN_COLUMNS.map((column) => ({ ...column, convoys: convoys.filter((convoy) => convoy.status === column.status), })); return { columns, gateInbox: convoys.filter((convoy) => GATE_INBOX_STATUSES.includes(convoy.status)), }; }), get: workspaceProcedure .input( z.object({ workspace: z.string().min(1), convoyId: z.string().uuid(), }), ) .query(async ({ ctx, input }) => { const convoy = await loadConvoyProject(ctx.db, ctx.workspace.id, input.convoyId); if (!convoy) { throw new TRPCError({ code: "NOT_FOUND", message: "Convoy not found." }); } const content = convoyObjectContentSchema.safeParse(convoy.content); const briefs = await getConvoyBriefs(ctx.db, ctx.workspace.id, convoy.id); const events = await getRecentConvoyEvents(ctx.db, ctx.workspace.id, convoy.id, 30); const checkpoints = await getConvoyCheckpointSummary( ctx.db, ctx.workspace.id, convoy.id, ); return { convoy: { id: convoy.id, title: convoy.title, status: convoy.status ?? "draft", description: convoy.description, content: content.success ? content.data : null, updatedAt: convoy.updatedAt, }, briefs, events, checkpoints: { count: checkpoints.count, lastOccurredAt: checkpoints.lastOccurredAt ? checkpoints.lastOccurredAt.toISOString() : null, }, }; }), create: workspaceProcedure .input( z.object({ workspace: z.string().min(1), slug: z .string() .min(1) .max(120) .regex(/^[a-z0-9]+(?:-[a-z0-9]+)*$/), classification: convoyClassificationSchema, skipFlags: z.array(z.string()).default([]), successMetric: z.string().min(1).max(2_000), ideaMarkdown: z.string().min(1), planSlug: z .string() .min(1) .max(200) .regex(/^[a-z0-9-]+$/) .optional(), epicSlug: z .string() .min(1) .max(200) .regex(/^[a-z0-9-]+$/) .optional(), backlogItemId: z.string().uuid().optional(), repoPath: z.string().min(1).optional(), briefFiles: z.array(z.string()).optional(), }), ) .mutation(async ({ ctx, input }) => { assertWorkspaceAdmin(ctx); const existing = await findConvoyBySlug(ctx.db, ctx.workspace.id, input.slug); if (existing) { throw new TRPCError({ code: "CONFLICT", message: `Convoy slug '${input.slug}' already exists.`, }); } const content = convoyObjectContentSchema.parse({ convoySlug: input.slug, classification: input.classification, skipFlags: input.skipFlags, successMetric: input.successMetric, planSlug: input.planSlug, epicSlug: input.epicSlug, backlogItemId: input.backlogItemId, repoPath: input.repoPath, briefFiles: input.briefFiles ?? (input.repoPath ? [input.repoPath] : undefined), }); const [created] = await ctx.db .insert(objects) .values({ type: "project", title: input.slug, workspaceId: ctx.workspace.id, description: input.ideaMarkdown, status: "draft", content, }) .returning(); if (!created) { throw new TRPCError({ code: "INTERNAL_SERVER_ERROR", message: "Failed to create convoy.", }); } await insertConvoyEvent(ctx.db, { workspaceId: ctx.workspace.id, convoyId: created.id, role: "role-conductor", actor: "role-conductor", eventType: "status_change", metadata: { from: null, to: "draft", source: "mission_control" }, }); return { convoyId: created.id }; }), gateAction: workspaceProcedure.input(gateActionSchema).mutation(async ({ ctx, input }) => { const convoy = await loadConvoyProject(ctx.db, ctx.workspace.id, input.convoyId); if (!convoy) { throw new TRPCError({ code: "NOT_FOUND", message: "Convoy not found." }); } const status = (convoy.status ?? "draft") as ConvoyStatus; const { actor, toStatus } = resolveGateTransition(status, input.action); try { const result = await transitionConvoy(ctx.db, { workspaceId: ctx.workspace.id, convoyId: input.convoyId, toStatus, actor, reason: input.reason, }); return result; } catch (error) { if (error instanceof ConvoyTransitionError) { throw new TRPCError({ code: "BAD_REQUEST", message: error.message, cause: { code: error.code }, }); } throw error; } }), gateCopilot: workspaceProcedure .input( z.object({ workspace: z.string().min(1), convoyId: z.string().uuid(), question: z.string().min(1).max(4_000).optional(), }), ) .mutation(async ({ ctx, input }) => { const convoy = await loadConvoyProject(ctx.db, ctx.workspace.id, input.convoyId); if (!convoy) { throw new TRPCError({ code: "NOT_FOUND", message: "Convoy not found." }); } const content = convoyObjectContentSchema.safeParse(convoy.content); const providerRows = await ctx.db .select() .from(modelProviders) .where(eq(modelProviders.workspaceId, ctx.workspace.id)) .orderBy(asc(modelProviders.alias)); const routeRows = await ctx.db .select() .from(roleModelRoutes) .where(eq(roleModelRoutes.workspaceId, ctx.workspace.id)); const resolved = resolveModel({ roleId: "gate-copilot", providers: providerRows.map(mapProviderRow), routes: routeRows.map(mapRouteRow), convoyPolicy: content.success ? content.data.modelPolicy : undefined, getApiKey: readApiKey, }); if (!resolved.ok) { throw new TRPCError({ code: "BAD_REQUEST", message: resolved.error.message, }); } const modelResult = createLanguageModelFromResolved(resolved.model); if (!modelResult.ok) { throw new TRPCError({ code: "BAD_REQUEST", message: modelResult.reason, }); } const status = convoy.status ?? "draft"; const system = [ "You are the gate co-pilot for an agent-pipeline convoy.", "Summarize the plan, highlight risks, and recommend approve / revise / request changes.", "Be concise and actionable.", "", `Convoy: ${content.success ? content.data.convoySlug : convoy.title}`, `Status: ${status}`, content.success ? `Classification: ${content.data.classification}` : "", content.success ? `Success metric: ${content.data.successMetric}` : "", convoy.description ? `Plan:\n${convoy.description}` : "", ] .filter(Boolean) .join("\n"); const userPrompt = input.question?.trim() || "Summarize this convoy for the operator and suggest the best gate action."; const { text } = await generateText({ model: modelResult.model, system, prompt: userPrompt, temperature: 0.4, maxTokens: 1_500, }); return { text: text.trim(), resolvedAlias: resolved.resolvedAlias, modelId: resolved.model.modelId, providerType: resolved.model.providerType, }; }), fleet: protectedProcedure.query(async ({ ctx }) => { const userId = ctx.session.user.id; const memberRows = await ctx.db .select({ workspaceId: workspaceMembers.workspaceId }) .from(workspaceMembers) .where(eq(workspaceMembers.userId, userId)); const ownedRows = await ctx.db .select({ id: workspaces.id }) .from(workspaces) .where(eq(workspaces.ownerUserId, userId)); const idSet = new Set([ ...memberRows.map((r) => r.workspaceId), ...ownedRows.map((r) => r.id), ]); const workspaceIds = [...idSet]; if (workspaceIds.length === 0) { return { byWorkspace: [], byStatus: {}, gateInboxAcross: [] }; } const wsRows = await ctx.db .select({ id: workspaces.id, slug: workspaces.slug, name: workspaces.name }) .from(workspaces) .where(inArray(workspaces.id, workspaceIds)); const byWorkspace: Array<{ workspace: { id: string; slug: string; name: string }; convoys: Array>[number] & { workspaceId: string; workspaceSlug: string; workspaceName: string }>; }> = []; const allConvoys: Array>[number] & { workspaceId: string; workspaceSlug: string; workspaceName: string }> = []; for (const ws of wsRows) { const convoys = await listConvoyProjects(ctx.db, ws.id); const enriched = convoys.map((c) => ({ ...c, workspaceId: ws.id, workspaceSlug: ws.slug, workspaceName: ws.name, })); byWorkspace.push({ workspace: ws, convoys: enriched }); allConvoys.push(...enriched); } const byStatus: Record = {}; for (const convoy of allConvoys) { const key = convoy.status; if (!byStatus[key]) byStatus[key] = []; byStatus[key].push(convoy); } const gateInboxAcross = allConvoys.filter((c) => (GATE_INBOX_STATUSES as readonly string[]).includes(c.status), ); return { byWorkspace, byStatus, gateInboxAcross }; }), });