zopu-code/packages/backend/convex/signalRouting.ts

import { env } from "@code/env/convex";
import { ConvexError, v } from "convex/values";

import type { Id } from "./_generated/dataModel";
import { mutation, query } from "./_generated/server";

const requireAgent = (token: string): void => {
  if (token !== env.FLUE_DB_TOKEN) {
    throw new ConvexError("Invalid agent control token");
  }
};

export const listProjects = query({
  args: { organizationId: v.id("organizations"), token: v.string() },
  handler: async (ctx, args) => {
    requireAgent(args.token);
    const projects = await ctx.db
      .query("projects")
      .withIndex("by_organizationId_and_createdAt", (q) =>
        q.eq("organizationId", args.organizationId)
      )
      .order("desc")
      .take(50);
    return projects.map((project) => ({
      _id: project._id,
      description: project.description ?? null,
      name: project.name,
    }));
  },
});

export const listEvidence = query({
  args: { organizationId: v.id("organizations"), token: v.string() },
  handler: async (ctx, args) => {
    requireAgent(args.token);
    const conversation = await ctx.db
      .query("conversations")
      .withIndex("by_organizationId", (q) =>
        q.eq("organizationId", args.organizationId)
      )
      .unique();
    if (!conversation) {
      return [];
    }
    const messages = await ctx.db
      .query("conversationMessages")
      .withIndex("by_conversationId_and_ordinal", (q) =>
        q.eq("conversationId", conversation._id)
      )
      .order("desc")
      .take(100);
    const evidence: {
      messageId: string;
      rawText: string;
      createdAt: number;
      submissionId: string | null;
    }[] = [];
    for (const message of messages) {
      if (message.role !== "user") {
        continue;
      }
      const turn = await ctx.db.get(message.turnId);
      if (!turn || !["processing", "completed"].includes(turn.status)) {
        continue;
      }
      const consumed = await ctx.db
        .query("signalSources")
        .withIndex("by_messageId", (q) => q.eq("messageId", message._id))
        .first();
      if (!consumed) {
        evidence.push({
          createdAt: message.createdAt,
          messageId: String(message._id),
          rawText: message.content,
          submissionId: turn.submissionId ?? null,
        });
      }
    }
    return evidence;
  },
});

export const createSignal = mutation({
  args: {
    organizationId: v.id("organizations"),
    projectId: v.id("projects"),
    messageIds: v.array(v.string()),
    problemStatement: v.object({
      title: v.string(),
      summary: v.string(),
      desiredOutcome: v.string(),
      constraints: v.array(v.string()),
    }),
    processedByAgentInstanceId: v.string(),
    token: v.string(),
  },
  handler: async (ctx, args) => {
    requireAgent(args.token);
    const project = await ctx.db.get(args.projectId);
    if (!project || project.organizationId !== args.organizationId) {
      throw new ConvexError("Project not found");
    }
    const conversation = await ctx.db
      .query("conversations")
      .withIndex("by_organizationId", (q) =>
        q.eq("organizationId", args.organizationId)
      )
      .unique();
    if (!conversation || args.messageIds.length === 0) {
      throw new ConvexError("Signal evidence is required");
    }
    const sourceKey = JSON.stringify(args.messageIds);
    const existing = await ctx.db
      .query("signals")
      .withIndex("by_organization_and_sourceKey", (q) =>
        q.eq("organizationId", args.organizationId).eq("sourceKey", sourceKey)
      )
      .unique();
    if (existing) {
      return { signalId: existing._id };
    }

    const messages = [];
    for (const messageId of args.messageIds) {
      const message = await ctx.db.get(messageId as Id<"conversationMessages">);
      if (
        !message ||
        message.conversationId !== conversation._id ||
        message.role !== "user"
      ) {
        throw new ConvexError(`Source message not found: ${messageId}`);
      }
      const turn = await ctx.db.get(message.turnId);
      if (!turn || !["processing", "completed"].includes(turn.status)) {
        throw new ConvexError(`Source message is not admitted: ${messageId}`);
      }
      messages.push(message);
    }

    const signalId = await ctx.db.insert("signals", {
      conversationId: conversation._id,
      createdAt: Date.now(),
      desiredOutcome: args.problemStatement.desiredOutcome,
      organizationId: args.organizationId,
      processedByAgentInstanceId: args.processedByAgentInstanceId,
      processedByAgentName: "zopu",
      projectId: args.projectId,
      sourceKey,
      summary: args.problemStatement.summary,
      title: args.problemStatement.title,
    });
    for (const [
      ordinal,
      constraint,
    ] of args.problemStatement.constraints.entries()) {
      await ctx.db.insert("signalConstraints", {
        ordinal,
        signalId,
        value: constraint,
      });
    }
    for (const [ordinal, message] of messages.entries()) {
      await ctx.db.insert("signalSources", {
        messageId: message._id,
        ordinal,
        rawTextSnapshot: message.content,
        signalId,
        sourceCreatedAt: message.createdAt,
      });
    }
    return { signalId };
  },
});