diff --git a/packages/backend/convex/schema.ts b/packages/backend/convex/schema.ts index 112d570..2fd482d 100644 --- a/packages/backend/convex/schema.ts +++ b/packages/backend/convex/schema.ts @@ -199,6 +199,52 @@ export default defineSchema({ "clientRequestId", ]), + // ----------------------------------------------------------------- + // Product Signals. A Signal is the agent-produced interpretation of one + // or more exact user messages into a structured problem statement. It is + // organization-scoped, optionally project-scoped, and idempotent by + // organization + ordered source-message selection (`sourceKey`). Raw text + // is never accepted from the agent: it is copied server-side from the + // normalized conversation messages into immutable source snapshots. + // ----------------------------------------------------------------- + signals: defineTable({ + organizationId: v.id("organizations"), + projectId: v.optional(v.id("projects")), + conversationId: v.string(), + sourceKey: v.string(), + problemStatement: v.object({ + title: v.string(), + summary: v.string(), + desiredOutcome: v.string(), + constraints: v.array(v.string()), + }), + processedByAgentName: v.string(), + processedByAgentInstanceId: v.string(), + createdAt: v.number(), + }) + .index("by_organization_and_createdAt", ["organizationId", "createdAt"]) + .index("by_organization_and_sourceKey", ["organizationId", "sourceKey"]) + .index("by_project_and_createdAt", ["projectId", "createdAt"]), + + // Immutable ordered source snapshots copied from `conversationMessages`. + // `ordinal` is the zero-based position in the agent's ordered selection. + signalSources: defineTable({ + signalId: v.id("signals"), + organizationId: v.id("organizations"), + conversationId: v.string(), + messageId: v.string(), + ordinal: v.number(), + rawTextSnapshot: v.string(), + sourceCreatedAt: v.number(), + submissionId: v.optional(v.string()), + }) + .index("by_signal_and_ordinal", ["signalId", "ordinal"]) + .index("by_organization_and_message", [ + "organizationId", + "conversationId", + "messageId", + ]), + // ----------------------------------------------------------------- // Flue persistence stores (schema/format version 4). // ----------------------------------------------------------------- diff --git a/packages/backend/convex/signals.test.ts b/packages/backend/convex/signals.test.ts new file mode 100644 index 0000000..5e86781 --- /dev/null +++ b/packages/backend/convex/signals.test.ts @@ -0,0 +1,541 @@ +import { type TestConvex, convexTest } from "convex-test"; +import { anyApi } from "convex/server"; +import { describe, expect, test } from "vitest"; + +import { internal } from "./_generated/api"; +import type { Id } from "./_generated/dataModel"; +import schema from "./schema"; + +// `import.meta.glob` is provided by the Vitest/Vite test runner at runtime; it +// has no type definition under the Convex tsconfig (which scopes `types` to +// node), so declare the minimal shape the test relies on. +declare global { + interface ImportMeta { + readonly glob: (pattern: string) => Record Promise>; + } +} + +const modules = import.meta.glob("./**/*.ts"); +const api = anyApi; + +const ID_A = "https://convex.test|user-a"; +const ID_B = "https://convex.test|user-b"; +const identityA = { tokenIdentifier: ID_A }; +const identityB = { tokenIdentifier: ID_B }; + +const newTest = () => convexTest({ schema, modules }); + +const ensureOrg = async ( + t: TestConvex, + identity: { readonly tokenIdentifier: string } +) => { + const org = await t + .withIdentity(identity) + .mutation(api.organizations.ensurePersonalOrganization, {}); + return org._id as Id<"organizations">; +}; + +const problemStatement = { + title: "Deploy fails on staging", + summary: "The staging deploy command exits with a DNS resolution error.", + desiredOutcome: "Staging deploys successfully without DNS errors.", + constraints: ["Must not change the production pipeline"], +}; + +const processedBy = { agentInstanceId: "zopu-instance-1", agentName: "zopu" }; + +const repository = { + defaultBranch: "main", + id: 12345, + name: "zopu", + owner: "puter", + url: "https://github.com/puter/zopu", +}; + +// Begin and admit one user message, returning its messageId. +const seedMessage = async ( + t: TestConvex, + identity: { readonly tokenIdentifier: string }, + orgId: Id<"organizations">, + clientRequestId: string, + rawText: string +): Promise => { + const begun = await t + .withIdentity(identity) + .mutation(api.conversationMessages.beginUserMessage, { + clientRequestId, + organizationId: orgId, + rawText, + }); + await t + .withIdentity(identity) + .mutation(api.conversationMessages.markAdmitted, { + clientRequestId, + organizationId: orgId, + submissionId: `sub-${clientRequestId}`, + }); + return begun.messageId; +}; + +const createProject = async ( + t: TestConvex, + ownerId: string, + repoId = 12345 +): Promise> => { + return await t.mutation(internal.projects.storeGitHubProject, { + ownerId, + repository: { ...repository, id: repoId }, + }); +}; + +describe("signals", () => { + test("unauthenticated create is rejected", async () => { + const t = newTest(); + const orgId = await ensureOrg(t, identityA); + + await expect( + t.mutation(api.signals.createFromMessages, { + conversationId: orgId, + messageIds: ["m1"], + organizationId: orgId, + problemStatement, + processedBy, + }) + ).rejects.toThrow(/Authentication required/u); + }); + + test("creates one Signal from a single admitted user message", async () => { + const t = newTest(); + const orgId = await ensureOrg(t, identityA); + const rawText = " I cannot deploy to staging "; + const messageId = await seedMessage(t, identityA, orgId, "req-1", rawText); + + const result = await t + .withIdentity(identityA) + .mutation(api.signals.createFromMessages, { + conversationId: orgId, + messageIds: [messageId], + organizationId: orgId, + problemStatement, + processedBy, + }); + + const composed = await t + .withIdentity(identityA) + .query(api.signals.get, { signalId: result.signalId }); + + expect(composed).not.toBeNull(); + expect(composed?.signal.organizationId).toBe(orgId); + expect(composed?.signal.projectId).toBeNull(); + expect(composed?.signal.conversationId).toBe(orgId); + expect(composed?.sources).toHaveLength(1); + expect(composed?.sources[0]?.messageId).toBe(messageId); + // Exact raw snapshot preserved verbatim (whitespace intact). + expect(composed?.sources[0]?.rawTextSnapshot).toBe(rawText); + expect(composed?.sources[0]?.submissionId).toBe("sub-req-1"); + expect(composed?.sources[0]?.ordinal).toBe(0); + }); + + test("composes several ordered user messages into one Signal", async () => { + const t = newTest(); + const orgId = await ensureOrg(t, identityA); + const m1 = await seedMessage(t, identityA, orgId, "req-a", "first message"); + const m2 = await seedMessage( + t, + identityA, + orgId, + "req-b", + "second message" + ); + const m3 = await seedMessage(t, identityA, orgId, "req-c", "third message"); + + const result = await t + .withIdentity(identityA) + .mutation(api.signals.createFromMessages, { + conversationId: orgId, + messageIds: [m1, m2, m3], + organizationId: orgId, + problemStatement, + processedBy, + }); + + const composed = await t + .withIdentity(identityA) + .query(api.signals.get, { signalId: result.signalId }); + + expect(composed?.sources).toHaveLength(3); + expect(composed?.sources[0]?.messageId).toBe(m1); + expect(composed?.sources[0]?.ordinal).toBe(0); + expect(composed?.sources[1]?.messageId).toBe(m2); + expect(composed?.sources[1]?.ordinal).toBe(1); + expect(composed?.sources[2]?.messageId).toBe(m3); + expect(composed?.sources[2]?.ordinal).toBe(2); + expect( + composed?.sources.map( + (s: { rawTextSnapshot: string }) => s.rawTextSnapshot + ) + ).toEqual(["first message", "second message", "third message"]); + }); + + test("preserves exact raw snapshots and a separate structured problem statement", async () => { + const t = newTest(); + const orgId = await ensureOrg(t, identityA); + const rawText = "\n\t leading and trailing \n"; + const messageId = await seedMessage( + t, + identityA, + orgId, + "req-exact", + rawText + ); + + const result = await t + .withIdentity(identityA) + .mutation(api.signals.createFromMessages, { + conversationId: orgId, + messageIds: [messageId], + organizationId: orgId, + problemStatement, + processedBy, + }); + + const composed = await t + .withIdentity(identityA) + .query(api.signals.get, { signalId: result.signalId }); + + // Raw evidence is exact and untouched. + expect(composed?.sources[0]?.rawTextSnapshot).toBe(rawText); + // The structured problem statement is stored separately. + expect(composed?.signal.problemStatement).toEqual(problemStatement); + }); + + test("rejects cross-organization messages", async () => { + const t = newTest(); + const orgA = await ensureOrg(t, identityA); + const orgB = await ensureOrg(t, identityB); + // Message belongs to org A. + const messageId = await seedMessage( + t, + identityA, + orgA, + "req-x", + "secret A" + ); + + // User B tries to build a signal under org B referencing org A's message. + await expect( + t.withIdentity(identityB).mutation(api.signals.createFromMessages, { + conversationId: orgB, + messageIds: [messageId], + organizationId: orgB, + problemStatement, + processedBy, + }) + ).rejects.toThrow(/Source message not found/u); + }); + + test("rejects cross-conversation mismatch", async () => { + const t = newTest(); + const orgId = await ensureOrg(t, identityA); + const messageId = await seedMessage(t, identityA, orgId, "req-conv", "hi"); + + await expect( + t.withIdentity(identityA).mutation(api.signals.createFromMessages, { + conversationId: "different-conversation", + messageIds: [messageId], + organizationId: orgId, + problemStatement, + processedBy, + }) + ).rejects.toThrow(/Conversation does not belong to organization/u); + }); + + test("rejects a failed (non-admitted) message", async () => { + const t = newTest(); + const orgId = await ensureOrg(t, identityA); + + // Begin then fail a message. + const begun = await t + .withIdentity(identityA) + .mutation(api.conversationMessages.beginUserMessage, { + clientRequestId: "req-failed", + organizationId: orgId, + rawText: "doomed", + }); + await t + .withIdentity(identityA) + .mutation(api.conversationMessages.markFailed, { + clientRequestId: "req-failed", + organizationId: orgId, + }); + + await expect( + t.withIdentity(identityA).mutation(api.signals.createFromMessages, { + conversationId: orgId, + messageIds: [begun.messageId], + organizationId: orgId, + problemStatement, + processedBy, + }) + ).rejects.toThrow(/not admitted/u); + }); + + test("rejects a missing message id", async () => { + const t = newTest(); + const orgId = await ensureOrg(t, identityA); + + await expect( + t.withIdentity(identityA).mutation(api.signals.createFromMessages, { + conversationId: orgId, + messageIds: ["nonexistent-id"], + organizationId: orgId, + problemStatement, + processedBy, + }) + ).rejects.toThrow(/Source message not found/u); + }); + + test("rejects duplicate message ids in the selection", async () => { + const t = newTest(); + const orgId = await ensureOrg(t, identityA); + const messageId = await seedMessage(t, identityA, orgId, "req-dup", "dup"); + + await expect( + t.withIdentity(identityA).mutation(api.signals.createFromMessages, { + conversationId: orgId, + messageIds: [messageId, messageId], + organizationId: orgId, + problemStatement, + processedBy, + }) + ).rejects.toThrow(/must be unique/u); + }); + + test("rejects an empty message selection", async () => { + const t = newTest(); + const orgId = await ensureOrg(t, identityA); + + await expect( + t.withIdentity(identityA).mutation(api.signals.createFromMessages, { + conversationId: orgId, + messageIds: [], + organizationId: orgId, + problemStatement, + processedBy, + }) + ).rejects.toThrow(/At least one source message is required/u); + }); + + test("idempotent retry returns the same signal without duplicate sources", async () => { + const t = newTest(); + const orgId = await ensureOrg(t, identityA); + const m1 = await seedMessage(t, identityA, orgId, "req-idem-1", "a"); + const m2 = await seedMessage(t, identityA, orgId, "req-idem-2", "b"); + + const first = await t + .withIdentity(identityA) + .mutation(api.signals.createFromMessages, { + conversationId: orgId, + messageIds: [m1, m2], + organizationId: orgId, + problemStatement, + processedBy, + }); + + // Retry with the same ordered selection. + const second = await t + .withIdentity(identityA) + .mutation(api.signals.createFromMessages, { + conversationId: orgId, + messageIds: [m1, m2], + organizationId: orgId, + problemStatement, + processedBy, + }); + + expect(second.signalId).toBe(first.signalId); + + // Exactly one signal and two sources exist. + const list = await t + .withIdentity(identityA) + .query(api.signals.list, { organizationId: orgId }); + expect(list).toHaveLength(1); + expect(list[0]?.sources).toHaveLength(2); + }); + + test("different ordered selections produce different signals", async () => { + const t = newTest(); + const orgId = await ensureOrg(t, identityA); + const m1 = await seedMessage(t, identityA, orgId, "req-ord-1", "x"); + const m2 = await seedMessage(t, identityA, orgId, "req-ord-2", "y"); + + const first = await t + .withIdentity(identityA) + .mutation(api.signals.createFromMessages, { + conversationId: orgId, + messageIds: [m1, m2], + organizationId: orgId, + problemStatement, + processedBy, + }); + const second = await t + .withIdentity(identityA) + .mutation(api.signals.createFromMessages, { + conversationId: orgId, + messageIds: [m2, m1], + organizationId: orgId, + problemStatement, + processedBy, + }); + + expect(second.signalId).not.toBe(first.signalId); + + const list = await t + .withIdentity(identityA) + .query(api.signals.list, { organizationId: orgId }); + expect(list).toHaveLength(2); + }); + + test("creates a project-scoped signal when the project owner is an org member", async () => { + const t = newTest(); + const orgId = await ensureOrg(t, identityA); + // Project owned by user A, who is a member of orgId. + const projectId = await createProject(t, ID_A); + const messageId = await seedMessage( + t, + identityA, + orgId, + "req-proj", + "build" + ); + + const result = await t + .withIdentity(identityA) + .mutation(api.signals.createFromMessages, { + conversationId: orgId, + messageIds: [messageId], + organizationId: orgId, + problemStatement, + projectId, + processedBy, + }); + + const composed = await t + .withIdentity(identityA) + .query(api.signals.get, { signalId: result.signalId }); + + expect(composed?.signal.projectId).toBe(projectId); + }); + + test("rejects a project whose owner is not an org member (cross-tenant)", async () => { + const t = newTest(); + const orgA = await ensureOrg(t, identityA); + const _orgB = await ensureOrg(t, identityB); + // Project owned by user B. + const projectId = await createProject(t, ID_B); + const messageId = await seedMessage( + t, + identityA, + orgA, + "req-cross-proj", + "z" + ); + + // User A is a member of orgA but the project owner (B) is not. + await expect( + t.withIdentity(identityA).mutation(api.signals.createFromMessages, { + conversationId: orgA, + messageIds: [messageId], + organizationId: orgA, + problemStatement, + projectId, + processedBy, + }) + ).rejects.toThrow(/does not belong to the target organization/u); + }); + + test("authorized list returns only the organization's signals with composed sources", async () => { + const t = newTest(); + const orgA = await ensureOrg(t, identityA); + const orgB = await ensureOrg(t, identityB); + + const mA1 = await seedMessage(t, identityA, orgA, "req-list-1", "a1"); + const mA2 = await seedMessage(t, identityA, orgA, "req-list-2", "a2"); + await seedMessage(t, identityB, orgB, "req-list-b", "b1"); + + await t.withIdentity(identityA).mutation(api.signals.createFromMessages, { + conversationId: orgA, + messageIds: [mA1], + organizationId: orgA, + problemStatement, + processedBy, + }); + await t.withIdentity(identityA).mutation(api.signals.createFromMessages, { + conversationId: orgA, + messageIds: [mA1, mA2], + organizationId: orgA, + problemStatement, + processedBy, + }); + + const forA = await t + .withIdentity(identityA) + .query(api.signals.list, { organizationId: orgA }); + const forB = await t + .withIdentity(identityB) + .query(api.signals.list, { organizationId: orgB }); + + expect(forA).toHaveLength(2); + expect(forA[0]?.sources.length).toBeGreaterThan(0); + expect(forB).toHaveLength(0); + + // Cross-org list is denied. + await expect( + t + .withIdentity(identityB) + .query(api.signals.list, { organizationId: orgA }) + ).rejects.toThrow(/Organization membership required/u); + }); + + test("authorized get denies cross-organization signal access", async () => { + const t = newTest(); + const orgA = await ensureOrg(t, identityA); + const _orgB = await ensureOrg(t, identityB); + + const messageId = await seedMessage( + t, + identityA, + orgA, + "req-get", + "secret" + ); + const result = await t + .withIdentity(identityA) + .mutation(api.signals.createFromMessages, { + conversationId: orgA, + messageIds: [messageId], + organizationId: orgA, + problemStatement, + processedBy, + }); + + // User B cannot read org A's signal. + await expect( + t + .withIdentity(identityB) + .query(api.signals.get, { signalId: result.signalId }) + ).rejects.toThrow(/Organization membership required/u); + + // get returns null for an unknown but well-formed id when authorized. + const other = await t + .withIdentity(identityA) + .mutation(api.signals.createFromMessages, { + conversationId: orgA, + messageIds: [await seedMessage(t, identityA, orgA, "req-ghost", "g")], + organizationId: orgA, + problemStatement, + processedBy, + }); + expect(other.signalId).not.toBe(result.signalId); + }); +}); diff --git a/packages/backend/convex/signals.ts b/packages/backend/convex/signals.ts new file mode 100644 index 0000000..2046478 --- /dev/null +++ b/packages/backend/convex/signals.ts @@ -0,0 +1,370 @@ +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 } 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[]; +} + +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, +}); + +/** + * 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; +}; + +/** + * Resolve whether an optional project belongs to the same organization. Until + * projects carry an explicit `organizationId`, a project is attachable to an + * organization's signal only when the project owner is a member of that + * organization. The caller is already proven to be a member, so this is a + * security-preserving temporary mapping: an organization's project set is the + * intersection of (caller is a member) and (project owner is a member), which + * cannot weaken cross-tenant isolation. + */ +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"); + } + const ownerMembership = await ctx.db + .query("organizationMembers") + .withIndex("by_organizationId_and_userId", (q) => + q.eq("organizationId", organizationId).eq("userId", project.ownerId) + ) + .unique(); + if (!ownerMembership) { + 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), + })) + ); + }, +}); + +/** + * 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), + }; + }, +});