86155455a7
- Draft PR via forge (apps/worker/src/git/pr.ts) push branch + createPullRequest draft with echodo-convoy marker, expose prUrl in Mission Control (convoy-service, convoys router, detail panel) - Operator notes mid-build: convoy_operator_messages (0016), tRPC post/list, worker poll between tool rounds (queued) and before generateText (direct), detail panel thread for in_progress - Doc-writer: harness doc-writer examples (docs/examples/research-pipeline.md), SOP template (docs/templates/doc-sop.md), role-doc-writer wiring Convoy: .convoys/worker-closeout-docwriter Co-authored-by: orchestrator
524 lines
16 KiB
TypeScript
524 lines
16 KiB
TypeScript
import { createLanguageModelFromResolved } from "@tasks/ai";
|
|
import {
|
|
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 { TRPCError } from "@trpc/server";
|
|
import { generateText } from "ai";
|
|
import { and, asc, 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(),
|
|
}));
|
|
}),
|
|
|
|
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 };
|
|
}),
|
|
});
|