import { composeSignal, SignalValidationError } from "@code/primitives/signal"; import { ConvexError, v } from "convex/values"; import { Effect } from "effect"; import type { Doc, Id } from "./_generated/dataModel"; import { type MutationCtx, mutation, type QueryCtx, query, } from "./_generated/server"; import { requireOrganizationMember, requireProjectMember } from "./authz"; const PROBLEM_STATEMENT = v.object({ title: v.string(), summary: v.string(), desiredOutcome: v.string(), constraints: v.array(v.string()), }); interface SignalSourceView { readonly _id: Id<"signalSources">; readonly _creationTime: number; readonly signalId: Id<"signals">; readonly organizationId: Id<"organizations">; readonly conversationId: string; readonly messageId: string; readonly ordinal: number; readonly rawTextSnapshot: string; readonly sourceCreatedAt: number; readonly submissionId: string | null; } interface SignalView { readonly _id: Id<"signals">; readonly _creationTime: number; readonly organizationId: Id<"organizations">; readonly projectId: Id<"projects"> | null; readonly conversationId: string; readonly problemStatement: { readonly title: string; readonly summary: string; readonly desiredOutcome: string; readonly constraints: readonly string[]; }; readonly processedByAgentName: string; readonly processedByAgentInstanceId: string; readonly createdAt: number; } interface ComposedSignal { readonly signal: SignalView; readonly sources: readonly SignalSourceView[]; } /** * An admitted user conversation message presented to the agent as candidate * Signal evidence. The agent selects from these by `messageId`; raw text and * timestamps are copied server-side and never accepted from the agent. */ interface ConversationMessageSourceView { readonly messageId: string; readonly clientRequestId: string; readonly rawText: string; readonly submissionId: string | null; readonly createdAt: number; } const toSignalView = (doc: Doc<"signals">): SignalView => ({ _creationTime: doc._creationTime, _id: doc._id, conversationId: doc.conversationId, createdAt: doc.createdAt, organizationId: doc.organizationId, problemStatement: doc.problemStatement, processedByAgentInstanceId: doc.processedByAgentInstanceId, processedByAgentName: doc.processedByAgentName, projectId: doc.projectId ?? null, }); const toSourceView = (doc: Doc<"signalSources">): SignalSourceView => ({ _creationTime: doc._creationTime, _id: doc._id, conversationId: doc.conversationId, messageId: doc.messageId, ordinal: doc.ordinal, organizationId: doc.organizationId, rawTextSnapshot: doc.rawTextSnapshot, signalId: doc.signalId, sourceCreatedAt: doc.sourceCreatedAt, submissionId: doc.submissionId ?? null, }); const toConversationMessageSourceView = ( doc: Doc<"conversationMessages"> ): ConversationMessageSourceView => ({ clientRequestId: doc.clientRequestId, createdAt: doc.createdAt, messageId: doc.messageId, rawText: doc.rawText, submissionId: doc.submissionId ?? null, }); /** * Build a deterministic, collision-safe key from the organization, the * conversation, and the ordered message selection. The key is the JSON * encoding of `[organizationId, conversationId, ...messageIds]`, so no * delimiter inside arbitrary message IDs can ever cause a collision and the * caller's selection order is part of the identity. */ const buildSourceKey = ( organizationId: Id<"organizations">, conversationId: string, messageIds: readonly string[] ): string => JSON.stringify([organizationId, conversationId, ...messageIds]); /** * Resolve all selected conversation messages server-side. Each message must * belong to the same organization and conversation, be a `user` message, be * `admitted`, and appear exactly once in the ordered selection. Raw text, * timestamps, and submission ids are copied from the stored rows; the agent * never supplies them. Returns the messages in the caller's requested order. */ const resolveSources = async ( ctx: MutationCtx, organizationId: Id<"organizations">, conversationId: string, messageIds: readonly string[] ): Promise[]> => { if (messageIds.length === 0) { throw new ConvexError("At least one source message is required"); } if (new Set(messageIds).size !== messageIds.length) { throw new ConvexError("Source message ids must be unique"); } const resolved: Doc<"conversationMessages">[] = []; for (const messageId of messageIds) { const doc = await ctx.db .query("conversationMessages") .withIndex("by_organization_and_message", (q) => q .eq("organizationId", organizationId) .eq("conversationId", conversationId) .eq("messageId", messageId) ) .unique(); if (!doc) { throw new ConvexError(`Source message not found: ${messageId}`); } if (doc.role !== "user") { throw new ConvexError( `Source message is not a user message: ${messageId}` ); } if (doc.status !== "admitted") { throw new ConvexError(`Source message is not admitted: ${messageId}`); } resolved.push(doc); } return resolved; }; /** * Verify a project belongs to the same organization. Projects now carry an * explicit `organizationId`; the caller is already proven to be a member, so * this is a direct invariant check with no cross-tenant leakage. */ const authorizeProject = async ( ctx: MutationCtx, organizationId: Id<"organizations">, projectId: Id<"projects"> ): Promise => { const project = await ctx.db.get(projectId); if (!project) { throw new ConvexError("Project not found"); } if (project.organizationId !== organizationId) { throw new ConvexError("Project does not belong to the target organization"); } }; /** * Atomically create one idempotent Signal from an ordered selection of user * messages plus a structured problem statement produced by the agent. * * Authorization: the authenticated identity must be a member of the target * organization. When a project is supplied it must belong to the same * organization. Every message id is resolved server-side from * `conversationMessages` under the same organization and conversation; raw * text, timestamps, and submission ids are copied from those rows and never * accepted from the agent. * * Idempotency: if a Signal with the same `(organizationId, sourceKey)` already * exists, it is returned as-is without creating duplicate sources. The whole * lookup-or-insert runs in one Convex mutation transaction, so retries cannot * produce duplicates. * * Returns the new (or existing) signal id. */ export const createFromMessages = mutation({ args: { organizationId: v.id("organizations"), projectId: v.optional(v.id("projects")), conversationId: v.string(), messageIds: v.array(v.string()), problemStatement: PROBLEM_STATEMENT, processedBy: v.object({ agentName: v.string(), agentInstanceId: v.string(), }), }, handler: async (ctx, args): Promise<{ signalId: Id<"signals"> }> => { await requireOrganizationMember(ctx, args.organizationId); if (args.projectId) { await authorizeProject(ctx, args.organizationId, args.projectId); } // The global conversation id equals the organization id; the caller's // conversation id must match. if (args.conversationId !== args.organizationId) { throw new ConvexError("Conversation does not belong to organization"); } const sourceKey = buildSourceKey( args.organizationId, args.conversationId, 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 sources = await resolveSources( ctx, args.organizationId, args.conversationId, args.messageIds ); // Compose the pure Effect Signal before writes, using a prospective id. // `composeSignal` decodes an `unknown` input, so pass plain objects. const now = Date.now(); const prospectiveId = crypto.randomUUID(); const scope = args.projectId === undefined ? { _tag: "Organization" as const, organizationId: args.organizationId } : { _tag: "Project" as const, organizationId: args.organizationId, projectId: args.projectId, }; const signal = await Effect.runPromise( composeSignal({ createdAt: now, id: prospectiveId, problemStatement: args.problemStatement, processedBy: args.processedBy, scope, sourceMessages: sources.map((doc) => ({ conversationId: doc.conversationId, createdAt: doc.createdAt, messageId: doc.messageId, rawText: doc.rawText, })), }) ).catch((error: unknown) => { if (error instanceof SignalValidationError) { throw new ConvexError(`Invalid signal: ${error.message}`); } throw new ConvexError( `Invalid signal: ${error instanceof Error ? error.message : "unknown"}` ); }); const signalId = await ctx.db.insert("signals", { conversationId: args.conversationId, createdAt: now, organizationId: args.organizationId, problemStatement: { constraints: [...signal.problemStatement.constraints], desiredOutcome: signal.problemStatement.desiredOutcome, summary: signal.problemStatement.summary, title: signal.problemStatement.title, }, processedByAgentInstanceId: args.processedBy.agentInstanceId, processedByAgentName: args.processedBy.agentName, ...(args.projectId === undefined ? {} : { projectId: args.projectId }), sourceKey, }); await Promise.all( sources.map((doc, ordinal) => ctx.db.insert("signalSources", { conversationId: doc.conversationId, messageId: doc.messageId, ordinal, organizationId: args.organizationId, rawTextSnapshot: doc.rawText, signalId, ...(doc.submissionId === undefined ? {} : { submissionId: doc.submissionId }), sourceCreatedAt: doc.createdAt, }) ) ); return { signalId }; }, }); /** * Load the composed source rows for a signal in ordinal order. */ const loadComposedSources = async ( ctx: QueryCtx, signal: Doc<"signals"> ): Promise => { const rows = await ctx.db .query("signalSources") .withIndex("by_signal_and_ordinal", (q) => q.eq("signalId", signal._id)) .collect(); // `collect` does not guarantee index order across all backends; sort by // ordinal to be deterministic. rows.sort((a, b) => a.ordinal - b.ordinal); return rows.map(toSourceView); }; /** * List the Signals for the authenticated user's organization, newest first. * Membership is required; cross-organization access is denied. Each entry * includes its composed source records in ordinal order. */ export const list = query({ args: { organizationId: v.id("organizations"), }, handler: async (ctx, args): Promise => { await requireOrganizationMember(ctx, args.organizationId); const signals = await ctx.db .query("signals") .withIndex("by_organization_and_createdAt", (q) => q.eq("organizationId", args.organizationId) ) .order("desc") .take(50); return Promise.all( signals.map(async (signal) => ({ signal: toSignalView(signal), sources: await loadComposedSources(ctx, signal), })) ); }, }); /** * List the bounded set of admitted user messages in the organization's * conversation that have not yet been consumed as a Signal source, newest * first. These are the candidate messages an agent should select from when * composing a new Signal. * * Authorization: the authenticated identity must be a member of the target * organization. The conversation id equals the organization id; the caller * never supplies tenancy. Cross-organization access is denied. * * Bounded: capped at 100 rows to keep agent evidence selection tractable. */ export const listAdmittedUnused = query({ args: { organizationId: v.id("organizations"), }, handler: async (ctx, args): Promise => { await requireOrganizationMember(ctx, args.organizationId); const conversationId = args.organizationId; const admitted = await ctx.db .query("conversationMessages") .withIndex("by_organization_and_message", (q) => q .eq("organizationId", args.organizationId) .eq("conversationId", conversationId) ) .order("desc") .take(100); const candidates = admitted.filter( (message) => message.role === "user" && message.status === "admitted" ); const unused: ConversationMessageSourceView[] = []; for (const message of candidates) { const consumed = await ctx.db .query("signalSources") .withIndex("by_organization_and_message", (q) => q .eq("organizationId", args.organizationId) .eq("conversationId", conversationId) .eq("messageId", message.messageId) ) .first(); if (!consumed) { unused.push(toConversationMessageSourceView(message)); } } return unused; }, }); /** * Get one Signal by id with its composed sources. Membership in the signal's * organization is required; access to another organization's signal is denied. */ export const get = query({ args: { signalId: v.id("signals"), }, handler: async (ctx, args): Promise => { const signal = await ctx.db.get(args.signalId); if (!signal) { return null; } await requireOrganizationMember(ctx, signal.organizationId); return { signal: toSignalView(signal), sources: await loadComposedSources(ctx, signal), }; }, }); // --------------------------------------------------------------------------- // Linked Signal query for the Work OS surface. // --------------------------------------------------------------------------- export interface LinkedSignalView { readonly signalId: Id<"signals">; readonly title: string; readonly summary: string; readonly sourceCount: number; readonly createdAt: number; } /** * List all Signal-issue attachments for a project, grouped by issue id. * User-authenticated: the caller must be a member of the project organization. * Used by the Work OS card list to show per-issue linked Signal counts. */ export const listLinkedSignalsForProject = query({ args: { projectId: v.id("projects"), }, handler: async (ctx, args): Promise> => { const { organizationId } = await requireProjectMember(ctx, args.projectId); const issues = await ctx.db .query("projectIssues") .withIndex("by_project_and_number", (q) => q.eq("projectId", args.projectId) ) .take(100); const result: Record = {}; for (const issue of issues) { const attachments = await ctx.db .query("signalIssueAttachments") .withIndex("by_issue", (q) => q.eq("issueId", issue._id)) .collect(); if (attachments.length === 0) { continue; } const signals = await Promise.all( attachments.map(async (attachment) => { const signal = await ctx.db.get(attachment.signalId); if (!signal || signal.organizationId !== organizationId) { return null; } const sources = await ctx.db .query("signalSources") .withIndex("by_signal_and_ordinal", (q) => q.eq("signalId", signal._id) ) .collect(); return { createdAt: signal.createdAt, signalId: signal._id, sourceCount: sources.length, summary: signal.problemStatement.summary, title: signal.problemStatement.title, } satisfies LinkedSignalView; }) ); result[String(issue._id)] = signals .filter((entry): entry is LinkedSignalView => entry !== null) // Deterministic order: newest linked Signal first. .toSorted((a, b) => b.createdAt - a.createdAt); } return result; }, });