Submatter/apps/web/server/routers/convoys.ts

import { createLanguageModelFromResolved } from "@tasks/ai";
import {
  completionArtifacts,
  convoyOperatorMessages,
  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 { generateCompletionSummary } from "@tasks/shared/worker";
import { TRPCError } from "@trpc/server";
import { generateText } from "ai";
import { and, asc, desc, eq, inArray } from "drizzle-orm";
import { z } from "zod";

import {
  findConvoyBySlug,
  getConvoyBriefs,
  getConvoyCheckpointSummary,
  getConvoyPrUrl,
  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<typeof gateActionSchema>["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,
      );
      const pr = await getConvoyPrUrl(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,
        },
        pr,
      };
    }),

  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,
      };
    }),

  postOperatorNote: workspaceProcedure
    .input(
      z.object({
        workspace: z.string().min(1),
        convoyId: z.string().uuid(),
        briefId: z.string().uuid().nullable().optional(),
        delivery: z.enum(["queued", "direct"]),
        body: z.string().min(1).max(2_000),
      }),
    )
    .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 authorUserId = ctx.session.user.id;

      const [row] = await ctx.db
        .insert(convoyOperatorMessages)
        .values({
          workspaceId: ctx.workspace.id,
          convoyId: input.convoyId,
          briefId: input.briefId ?? null,
          delivery: input.delivery,
          body: input.body,
          authorUserId,
        })
        .returning();

      if (!row) {
        throw new TRPCError({
          code: "INTERNAL_SERVER_ERROR",
          message: "Failed to create operator note.",
        });
      }

      return row;
    }),

  listOperatorNotes: workspaceProcedure
    .input(
      z.object({
        workspace: z.string().min(1),
        convoyId: z.string().uuid(),
        briefId: z.string().uuid().nullable().optional(),
      }),
    )
    .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 rows = await ctx.db
        .select()
        .from(convoyOperatorMessages)
        .where(
          and(
            eq(convoyOperatorMessages.workspaceId, ctx.workspace.id),
            eq(convoyOperatorMessages.convoyId, input.convoyId),
          ),
        )
        .orderBy(asc(convoyOperatorMessages.createdAt));

      void input.briefId;

      return rows.map((r) => ({
        id: r.id,
        workspaceId: r.workspaceId,
        convoyId: r.convoyId,
        briefId: r.briefId,
        delivery: r.delivery,
        body: r.body,
        authorUserId: r.authorUserId,
        deliveredAt: r.deliveredAt ? r.deliveredAt.toISOString() : null,
        createdAt: r.createdAt.toISOString(),
      }));
    }),

  complete: workspaceProcedure
    .input(
      z.object({
        workspace: z.string().min(1),
        convoyId: z.string().uuid(),
        screenshotUrl: z.string().url().nullable().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 status = (convoy.status ?? "draft") as ConvoyStatus;
      if (status !== "merged" && status !== "published") {
        throw new TRPCError({
          code: "BAD_REQUEST",
          message: `Convoy must be in merged or published to complete, currently ${status}`,
        });
      }

      const contentParsed = convoyObjectContentSchema.safeParse(convoy.content);
      const briefs = await getConvoyBriefs(ctx.db, ctx.workspace.id, convoy.id);
      const pr = await getConvoyPrUrl(ctx.db, ctx.workspace.id, convoy.id);
      const events = await getRecentConvoyEvents(ctx.db, ctx.workspace.id, convoy.id, 30);

      let modelAlias: string | null = null;
      let harnessAlias: string | null = null;
      for (const ev of events) {
        const maybeModel = (ev as Record<string, unknown>).modelAlias as string | null | undefined;
        const maybeHarness = (ev as Record<string, unknown>).harnessAlias as string | null | undefined;
        if (!modelAlias && typeof maybeModel === "string" && maybeModel) modelAlias = maybeModel;
        if (!harnessAlias && typeof maybeHarness === "string" && maybeHarness) harnessAlias = maybeHarness;
        if (modelAlias && harnessAlias) break;
      }

      const slug = contentParsed.success ? contentParsed.data.convoySlug : convoy.title;
      const classification = contentParsed.success
        ? contentParsed.data.classification
        : ("feature" as const);
      const successMetric = contentParsed.success
        ? contentParsed.data.successMetric
        : "No success metric captured";

      const summaryMd = generateCompletionSummary({
        slug,
        classification,
        successMetric,
        briefs: briefs.map((b) => ({ title: b.title, status: b.status })),
        modelAlias,
        harnessAlias,
        prUrl: pr.prUrl,
      });

      let result: Awaited<ReturnType<typeof transitionConvoy>>;
      try {
        result = await transitionConvoy(ctx.db, {
          workspaceId: ctx.workspace.id,
          convoyId: input.convoyId,
          toStatus: "released",
          actor: "human-gate-3",
        });
      } catch (error) {
        if (error instanceof ConvoyTransitionError) {
          throw new TRPCError({
            code: "BAD_REQUEST",
            message: error.message,
            cause: { code: error.code },
          });
        }
        throw error;
      }

      const [artifact] = await ctx.db
        .insert(completionArtifacts)
        .values({
          workspaceId: ctx.workspace.id,
          convoyId: convoy.id,
          summaryMd,
          screenshotUrl: input.screenshotUrl ?? null,
          createdBy: ctx.session.user.id,
        })
        .returning();

      if (!artifact) {
        throw new TRPCError({
          code: "INTERNAL_SERVER_ERROR",
          message: "Failed to create completion artifact.",
        });
      }

      return {
        convoy: result.convoy,
        artifact: {
          id: artifact.id,
          workspaceId: artifact.workspaceId,
          convoyId: artifact.convoyId,
          summaryMd: artifact.summaryMd,
          screenshotUrl: artifact.screenshotUrl,
          createdBy: artifact.createdBy,
          createdAt: artifact.createdAt.toISOString(),
        },
      };
    }),

  listArtifacts: workspaceProcedure.query(async ({ ctx }) => {
    const rows = await ctx.db
      .select({
        artifact: completionArtifacts,
        convoyTitle: objects.title,
        convoyStatus: objects.status,
        convoyContent: objects.content,
      })
      .from(completionArtifacts)
      .leftJoin(objects, eq(objects.id, completionArtifacts.convoyId))
      .where(eq(completionArtifacts.workspaceId, ctx.workspace.id))
      .orderBy(desc(completionArtifacts.createdAt));

    const enriched = await Promise.all(
      rows.map(async (row) => {
        const pr = await getConvoyPrUrl(ctx.db, ctx.workspace.id, row.artifact.convoyId);
        const events = await getRecentConvoyEvents(
          ctx.db,
          ctx.workspace.id,
          row.artifact.convoyId,
          20,
        );
        const content = convoyObjectContentSchema.safeParse(row.convoyContent);
        return {
          id: row.artifact.id,
          workspaceId: row.artifact.workspaceId,
          convoyId: row.artifact.convoyId,
          convoyTitle: row.convoyTitle ?? "Untitled",
          convoySlug: content.success ? content.data.convoySlug : (row.convoyTitle ?? ""),
          convoyStatus: row.convoyStatus ?? "released",
          summaryMd: row.artifact.summaryMd,
          screenshotUrl: row.artifact.screenshotUrl,
          createdBy: row.artifact.createdBy,
          createdAt: row.artifact.createdAt.toISOString(),
          pr,
          timeline: events.map((ev) => ({
            id: ev.id,
            eventType: ev.eventType,
            role: ev.role,
            actor: ev.actor,
            occurredAt: ev.occurredAt.toISOString(),
            metadata: ev.metadata,
          })),
        };
      }),
    );

    return enriched;
  }),

  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<string>([
      ...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<Awaited<ReturnType<typeof listConvoyProjects>>[number] & { workspaceId: string; workspaceSlug: string; workspaceName: string }>;
    }> = [];
    const allConvoys: Array<Awaited<ReturnType<typeof listConvoyProjects>>[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<string, typeof allConvoys> = {};
    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 };
  }),
});