From 1324316feeca7e613eafefde378a949bb792ed37 Mon Sep 17 00:00:00 2001 From: -Puter <22245429+puterhimself@users.noreply.github.com> Date: Fri, 24 Jul 2026 20:32:48 +0530 Subject: [PATCH 1/2] fix(signals): source order follows agent selection, not chronological timestamps Remove the OutOfOrderSource constraint from composeSignal. The agent's selection order is authoritative for source ordinals; timestamps remain as per-source provenance metadata. This resolves the contract conflict where reversed message selections (valid agent compositions) were rejected by the primitive. Each source still retains its original createdAt, so chronological provenance is preserved exactly. The sourceKey includes ordered message IDs, so different orderings produce distinct signals with deterministic idempotency. Coverage added: - Primitive: non-chronological order accepted, timestamps preserved - Backend: reversed selection preserves ordinals and exact raw text --- packages/backend/convex/signals.test.ts | 63 +++++++++++++++++++++++++ packages/primitives/src/signal.test.ts | 17 +++++-- packages/primitives/src/signal.ts | 19 ++------ 3 files changed, 82 insertions(+), 17 deletions(-) diff --git a/packages/backend/convex/signals.test.ts b/packages/backend/convex/signals.test.ts index 6d1e508..b14e092 100644 --- a/packages/backend/convex/signals.test.ts +++ b/packages/backend/convex/signals.test.ts @@ -412,6 +412,69 @@ describe("signals", () => { expect(list).toHaveLength(2); }); + test("reversed selection preserves agent ordinals and original timestamps", async () => { + const t = newTest(); + const orgId = await ensureOrg(t, identityA); + const m1 = await seedMessage(t, identityA, orgId, "req-rev-1", "first"); + const m2 = await seedMessage(t, identityA, orgId, "req-rev-2", "second"); + + // Chronological selection: m1 then m2. + const chrono = await t + .withIdentity(identityA) + .mutation(api.signals.createFromMessages, { + conversationId: orgId, + messageIds: [m1, m2], + organizationId: orgId, + problemStatement, + processedBy, + }); + + // Reversed selection: m2 then m1 (non-chronological agent order). + const reversed = await t + .withIdentity(identityA) + .mutation(api.signals.createFromMessages, { + conversationId: orgId, + messageIds: [m2, m1], + organizationId: orgId, + problemStatement, + processedBy, + }); + + expect(reversed.signalId).not.toBe(chrono.signalId); + + const chronoSignal = await t + .withIdentity(identityA) + .query(api.signals.get, { signalId: chrono.signalId }); + const reversedSignal = await t + .withIdentity(identityA) + .query(api.signals.get, { signalId: reversed.signalId }); + + // Chronological: ordinals follow selection. + expect(chronoSignal?.sources[0]?.messageId).toBe(m1); + expect(chronoSignal?.sources[0]?.ordinal).toBe(0); + expect(chronoSignal?.sources[1]?.messageId).toBe(m2); + expect(chronoSignal?.sources[1]?.ordinal).toBe(1); + + // Reversed: ordinals follow the agent's reversed selection, and each + // source retains its original creation timestamp (unchanged by reordering). + expect(reversedSignal?.sources[0]?.messageId).toBe(m2); + expect(reversedSignal?.sources[0]?.ordinal).toBe(0); + expect(reversedSignal?.sources[1]?.messageId).toBe(m1); + expect(reversedSignal?.sources[1]?.ordinal).toBe(1); + + // Exact raw text is preserved in both orderings. + expect( + chronoSignal?.sources.map( + (s: { rawTextSnapshot: string }) => s.rawTextSnapshot + ) + ).toEqual(["first", "second"]); + expect( + reversedSignal?.sources.map( + (s: { rawTextSnapshot: string }) => s.rawTextSnapshot + ) + ).toEqual(["second", "first"]); + }); + test("creates a project-scoped signal when the project owner is an org member", async () => { const t = newTest(); const orgId = await ensureOrg(t, identityA); diff --git a/packages/primitives/src/signal.test.ts b/packages/primitives/src/signal.test.ts index e767463..0fad741 100644 --- a/packages/primitives/src/signal.test.ts +++ b/packages/primitives/src/signal.test.ts @@ -223,11 +223,22 @@ describe("composeSignal", () => { getValidationError(input, "MixedConversation"); }); - it("rejects decreasing source timestamps", () => { + it("accepts non-chronological source order with preserved timestamps", () => { const input = makeSignalInput(); - input.sourceMessages[1].createdAt = 99; + // Agent selects the later message first (logical, not chronological order). + const [earlier, later] = input.sourceMessages; + input.sourceMessages = [later, earlier]; - getValidationError(input, "OutOfOrderSource"); + const signal = runSignal(input); + + // Ordinal follows the agent's selection, not timestamps. + expect(signal.sourceMessages.map(({ messageId }) => messageId)).toEqual([ + "message-2", + "message-1", + ]); + // Original timestamps are preserved exactly on each source. + const timestamps = signal.sourceMessages.map(({ createdAt }) => createdAt); + expect(timestamps).toEqual([200, 100]); }); it("preserves source order and exact raw whitespace", () => { diff --git a/packages/primitives/src/signal.ts b/packages/primitives/src/signal.ts index ceb1f30..fc4c8ec 100644 --- a/packages/primitives/src/signal.ts +++ b/packages/primitives/src/signal.ts @@ -92,7 +92,6 @@ export const SignalValidationReason = Schema.Literals([ "InvalidInput", "DuplicateSource", "MixedConversation", - "OutOfOrderSource", ]); export type SignalValidationReason = typeof SignalValidationReason.Type; @@ -118,9 +117,11 @@ export const composeSignal = Effect.fn("Signal.compose")( ) ); - const [{ conversationId, createdAt: firstCreatedAt }] = - signal.sourceMessages; - let previousCreatedAt = firstCreatedAt; + // Source order follows the agent's logical selection, not chronological + // timestamp order. Each source retains its original createdAt so temporal + // provenance is preserved exactly; the ordinal array position reflects the + // agent's narrative composition of the problem statement. + const [{ conversationId }] = signal.sourceMessages; const messageIds = new Set(); for (const source of signal.sourceMessages) { @@ -142,16 +143,6 @@ export const composeSignal = Effect.fn("Signal.compose")( ); } messageIds.add(source.messageId); - - if (source.createdAt < previousCreatedAt) { - return yield* Effect.fail( - new SignalValidationError({ - message: "Signal source timestamps must be non-decreasing", - reason: "OutOfOrderSource", - }) - ); - } - previousCreatedAt = source.createdAt; } return signal; From 03fc32bd6db108c65651b3f32ea48302da9b473b Mon Sep 17 00:00:00 2001 From: -Puter <22245429+puterhimself@users.noreply.github.com> Date: Fri, 24 Jul 2026 20:48:18 +0530 Subject: [PATCH 2/2] feat(signals): project-scoped Zopu routing loop with secure tools Add the complete project-scoped Zopu chat, Signal extraction, and work-routing loop: Backend (signalRouting.ts): - Agent-token-gated functions for the full routing cycle: list evidence, create signal, list signals, list active issues, attach signal to issue, create issue from signal, begin issue, get project context, list projects - attachSignalToIssue requires signal.projectId to exist and equal issue.projectId (org equality alone is insufficient) - createIssueFromSignal is idempotent: queries existing attachments before creating, returns created/reused flag - Duplicate attach retries return existing relation without emitting a duplicate event - All functions deny-by-default with agent token validation Schema: - Add signalIssueAttachments table with indexes by signal, issue, and the composite (signalId, issueId) for idempotent lookups Agent tools (tools/signals.ts): - Nine Flue tool definitions bound to the organization-scoped agent instance ID, calling the agent-gated Convex functions - Organization ID never accepted as free input; always from the bound instance Zopu agent (zopu.ts): - Comprehensive routing-loop instructions: assess actionability, identify project, create signal only when actionable, route via attach or create, explain outcome, optionally begin - Wiring of routing tools into the agent definition Tests (signalRouting.test.ts): - Authentication: invalid token rejected on every function - Project scoping: evidence, signals, issues scoped correctly - Idempotency: duplicate attach, duplicate create-from-signal, duplicate signal creation - Attach vs create: both paths verified - Cross-project rejection: signal and issue must be in same project - Org-scoped signal attach rejected (no projectId) - Provenance: exact raw text preserved through routing loop - Begin issue: transition and auth verification --- packages/agents/src/agents/zopu.ts | 57 +- packages/agents/src/tools/signals.ts | 317 ++++++++ packages/backend/convex/schema.ts | 17 + packages/backend/convex/signalRouting.test.ts | 552 ++++++++++++++ packages/backend/convex/signalRouting.ts | 721 ++++++++++++++++++ 5 files changed, 1658 insertions(+), 6 deletions(-) create mode 100644 packages/agents/src/tools/signals.ts create mode 100644 packages/backend/convex/signalRouting.test.ts create mode 100644 packages/backend/convex/signalRouting.ts diff --git a/packages/agents/src/agents/zopu.ts b/packages/agents/src/agents/zopu.ts index ff6c3b7..a670408 100644 --- a/packages/agents/src/agents/zopu.ts +++ b/packages/agents/src/agents/zopu.ts @@ -2,30 +2,75 @@ import path from "node:path"; import { parseAgentEnv } from "@code/env/agent"; import { defineAgent } from "@flue/runtime"; - // import { agentos } from "../sandboxes/agentos"; import { local } from "@flue/runtime/node"; import paseo from "../skills/paseo/SKILL.md" with { type: "skill" }; import { paseoCli } from "../tools/paseo"; +import { createSignalRoutingTools } from "../tools/signals"; const repositoryRoot = path.resolve(process.cwd(), "../.."); +const INSTRUCTIONS = `You are Zopu, the global planning and work-routing agent for the Zopu Work OS. + +## Your role + +You listen to user messages in the persistent global conversation and decide whether they contain actionable work. The control plane stores every user message as exact evidence before you process it; you never supply or rewrite raw message text. + +## Work routing loop + +When a user sends a message, follow this decision flow: + +1. **Assess actionability.** Does the message contain a concrete problem, request, blocker, opportunity, or decision that warrants a work unit? Greetings, questions about the system, casual conversation, and exploration do NOT create work. If the message is casual conversation, respond naturally and do nothing else. + +2. **Identify project context.** If the message references a project, call list_projects to find it. If the user has one project, use it. If the project is ambiguous, ask one focused clarification. + +3. **Create a Signal (only when actionable).** When the message is actionable: + a. Call list_signal_evidence to see the exact admitted user messages. + b. Select the message IDs that compose the problem statement. + c. Call create_signal with a structured problem statement (title, summary, desiredOutcome, constraints). The problem statement must faithfully represent the user's own intent. Do not invent scope they did not mention. + d. Include the projectId when the project is known. + +4. **Route the Signal.** After creating the Signal (or if one already exists): + a. Call list_active_issues for the relevant project. + b. Compare the Signal's problem to existing issues. + c. If the Signal clearly relates to an existing active issue, call attach_signal_to_issue. + d. If the Signal is new work that does not match any existing issue, call create_issue_from_signal. + e. If genuinely uncertain whether to attach or create, ask one focused question. + +5. **Explain the outcome.** Tell the user clearly what happened: + - "Created a Signal: [title] and attached it to issue #[number]: [issue title]." + - "Created a Signal: [title] and opened new issue #[number]: [issue title]." + - Include the problem statement title so the user can verify accuracy. + +6. **Optionally begin work.** Only when the user explicitly asks to start or work on the issue, call begin_issue. Do not begin work automatically. + +## Rules + +- Never supply or rewrite the raw source message text. The control plane copies it server-side. +- Never create work from casual chat. +- Ask at most one focused clarification when genuinely ambiguous. +- Preserve project and organization scope at all times. +- Repeated delivery of the same message must not create duplicate Signals or attachments. The backend is idempotent. +- Never claim you created a Signal until the tool call returns successfully. +- Do not create Candidate Work, Work Units, or Runs directly. Those are downstream concerns. +- When the user asks about existing work, use list_active_issues and list_recent_signals to answer.`; + export { authenticatedAgentRoute as route } from "../auth"; -export default defineAgent(({ env }) => { +export default defineAgent(({ env, id }) => { const { AGENT_MODEL_NAME, AGENT_MODEL_PROVIDER } = parseAgentEnv(env); return { cwd: repositoryRoot, - description: "A simple conversational agent with persistent history.", - instructions: - "You are Zopu, the global planning agent. Help the user clarify and plan work. User messages are durably captured as Signal evidence by the control plane; do not claim that you created a Signal until secure tool execution is wired.", + description: + "Project-scoped work-routing agent that turns conversation into Signals and routes them to issues.", + instructions: INSTRUCTIONS, model: `${AGENT_MODEL_PROVIDER}/${AGENT_MODEL_NAME}`, // sandbox: agentos(), sandbox: local({ cwd: repositoryRoot }), skills: [paseo], - tools: [paseoCli], + tools: [paseoCli, ...createSignalRoutingTools(id, env)], }; }); diff --git a/packages/agents/src/tools/signals.ts b/packages/agents/src/tools/signals.ts new file mode 100644 index 0000000..0b8a5bd --- /dev/null +++ b/packages/agents/src/tools/signals.ts @@ -0,0 +1,317 @@ +import { parseAgentEnv } from "@code/env/agent"; +import { defineTool } from "@flue/runtime"; +import { ConvexHttpClient } from "convex/browser"; +import { makeFunctionReference } from "convex/server"; +import * as v from "valibot"; + +// --------------------------------------------------------------------------- +// Agent-gated Convex function references for the routing loop. +// --------------------------------------------------------------------------- + +const listEvidenceRef = makeFunctionReference< + "query", + { readonly organizationId: string; readonly token: string }, + { + messageId: string; + rawText: string; + createdAt: number; + submissionId: string | null; + }[] +>("signalRouting:listEvidence"); + +const createSignalRef = makeFunctionReference< + "mutation", + { + readonly organizationId: string; + readonly projectId?: string; + messageIds: string[]; + readonly problemStatement: { + readonly title: string; + readonly summary: string; + readonly desiredOutcome: string; + constraints: string[]; + }; + readonly processedByAgentInstanceId: string; + readonly token: string; + }, + { signalId: string } +>("signalRouting:createSignal"); + +const listSignalsRef = makeFunctionReference< + "query", + { + readonly organizationId: string; + readonly projectId?: string; + readonly token: string; + }, + { + _id: string; + createdAt: number; + problemStatement: { + title: string; + summary: string; + desiredOutcome: string; + constraints: string[]; + }; + projectId: string | null; + }[] +>("signalRouting:listSignals"); + +const listActiveIssuesRef = makeFunctionReference< + "query", + { + readonly organizationId: string; + readonly projectId: string; + readonly token: string; + }, + { + _id: string; + number: number; + title: string; + body: string; + status: string; + projectId: string; + updatedAt: number; + }[] +>("signalRouting:listActiveIssues"); + +const attachSignalToIssueRef = makeFunctionReference< + "mutation", + { + readonly organizationId: string; + readonly signalId: string; + readonly issueId: string; + readonly token: string; + }, + { attachmentId: string; alreadyAttached: boolean } +>("signalRouting:attachSignalToIssue"); + +const createIssueFromSignalRef = makeFunctionReference< + "mutation", + { + readonly organizationId: string; + readonly signalId: string; + readonly token: string; + }, + { issueId: string; projectId: string } +>("signalRouting:createIssueFromSignal"); + +const beginIssueRef = makeFunctionReference< + "mutation", + { + readonly organizationId: string; + readonly issueId: string; + readonly token: string; + }, + "queued" | "working" +>("signalRouting:beginIssue"); + +const getProjectContextRef = makeFunctionReference< + "query", + { + readonly organizationId: string; + readonly projectId: string; + readonly token: string; + }, + { + project: { + _id: string; + name: string; + description: string | null; + organizationId: string; + }; + contextDocuments: { + kind: string; + path: string; + content: string; + revision: number; + }[]; + } +>("signalRouting:getProjectContext"); + +const listProjectsRef = makeFunctionReference< + "query", + { + readonly organizationId: string; + readonly token: string; + }, + { + _id: string; + name: string; + description: string | null; + organizationId: string; + }[] +>("signalRouting:listProjects"); + +// --------------------------------------------------------------------------- +// Tool factory: creates all routing tools bound to one agent instance. +// +// The `instanceId` is the organization ID (established by the authenticated +// route middleware). The tool never accepts organization ID as free input; +// it always comes from the bound instance. This is the hard tenancy boundary. +// --------------------------------------------------------------------------- + +export const createSignalRoutingTools = ( + instanceId: string, + runtimeEnv: Record +) => { + const env = parseAgentEnv(runtimeEnv); + const client = new ConvexHttpClient(env.CONVEX_URL); + const token = env.FLUE_DB_TOKEN; + // The instance id is the organization id; tools are scoped to it. + const organizationId = instanceId; + + return [ + defineTool({ + description: + "List projects in the current organization. Call this first to identify the project context the user is working in.", + name: "list_projects", + async run() { + return await client.query(listProjectsRef, { + organizationId, + token, + }); + }, + }), + + defineTool({ + description: + "Read the project name, description, and canonical context documents (README, product, design, tech, agents). Use this to understand the project before routing work.", + input: v.object({ + projectId: v.string(), + }), + name: "get_project_context", + async run({ input }) { + return await client.query(getProjectContextRef, { + organizationId, + projectId: input.projectId, + token, + }); + }, + }), + + defineTool({ + description: + "List admitted user messages in the current conversation that have not yet been consumed by a Signal. These are the candidate evidence messages. Raw text is exact — never modify it.", + name: "list_signal_evidence", + async run() { + return await client.query(listEvidenceRef, { + organizationId, + token, + }); + }, + }), + + defineTool({ + description: + "Create a Signal from one or more admitted user messages plus a structured problem statement. Use ONLY when the conversation contains an actionable problem, request, blocker, or decision. Do NOT create Signals for casual chat, greetings, or exploration without a concrete problem. The problemStatement must faithfully represent the user's own words — do not invent or rewrite their intent.", + input: v.object({ + messageIds: v.array(v.string()), + problemStatement: v.object({ + constraints: v.array(v.string()), + desiredOutcome: v.string(), + summary: v.string(), + title: v.string(), + }), + projectId: v.optional(v.string()), + }), + name: "create_signal", + async run({ input }) { + return await client.mutation(createSignalRef, { + messageIds: input.messageIds, + organizationId, + problemStatement: input.problemStatement, + ...(input.projectId === undefined + ? {} + : { projectId: input.projectId }), + processedByAgentInstanceId: organizationId, + token, + }); + }, + }), + + defineTool({ + description: + "List recent Signals for a project (or organization-wide if no projectId). Use this to see what has already been captured.", + input: v.object({ + projectId: v.optional(v.string()), + }), + name: "list_recent_signals", + async run({ input }) { + return await client.query(listSignalsRef, { + organizationId, + ...(input.projectId === undefined + ? {} + : { projectId: input.projectId }), + token, + }); + }, + }), + + defineTool({ + description: + "List active (open, queued, working, needs-input) ProjectIssues for a project. Use this to find existing issues that a new Signal might relate to before deciding whether to attach or create a new one.", + input: v.object({ + projectId: v.string(), + }), + name: "list_active_issues", + async run({ input }) { + return await client.query(listActiveIssuesRef, { + organizationId, + projectId: input.projectId, + token, + }); + }, + }), + + defineTool({ + description: + "Attach a Signal to an existing ProjectIssue. This links the signal's evidence to an open issue. Idempotent: repeating the same attachment is safe and returns the existing relation. Use when the signal's problem clearly relates to an existing active issue.", + input: v.object({ + issueId: v.string(), + signalId: v.string(), + }), + name: "attach_signal_to_issue", + async run({ input }) { + return await client.mutation(attachSignalToIssueRef, { + issueId: input.issueId, + organizationId, + signalId: input.signalId, + token, + }); + }, + }), + + defineTool({ + description: + "Create a new ProjectIssue from a Signal. The signal must be project-scoped. This also auto-attaches the signal to the new issue. Use when the signal's problem does not match any existing active issue.", + input: v.object({ + signalId: v.string(), + }), + name: "create_issue_from_signal", + async run({ input }) { + return await client.mutation(createIssueFromSignalRef, { + organizationId, + signalId: input.signalId, + token, + }); + }, + }), + + defineTool({ + description: + "Begin working on a ProjectIssue by transitioning it to queued. Use only after the user explicitly confirms they want to start the issue. Returns the new status.", + input: v.object({ + issueId: v.string(), + }), + name: "begin_issue", + async run({ input }) { + return await client.mutation(beginIssueRef, { + issueId: input.issueId, + organizationId, + token, + }); + }, + }), + ]; +}; diff --git a/packages/backend/convex/schema.ts b/packages/backend/convex/schema.ts index 63ec174..5e2cf2b 100644 --- a/packages/backend/convex/schema.ts +++ b/packages/backend/convex/schema.ts @@ -282,6 +282,23 @@ export default defineSchema({ "messageId", ]), + // ----------------------------------------------------------------- + // Signal-to-issue attachments. A proper relation that links one Signal + // to one ProjectIssue, idempotent by (signalId, issueId). Multiple + // Signals may attach to the same ProjectIssue. The relation is scoped + // by organization and project to prevent cross-tenant leakage. + // ----------------------------------------------------------------- + signalIssueAttachments: defineTable({ + signalId: v.id("signals"), + issueId: v.id("projectIssues"), + organizationId: v.id("organizations"), + projectId: v.id("projects"), + createdAt: v.number(), + }) + .index("by_signal", ["signalId"]) + .index("by_issue", ["issueId"]) + .index("by_signal_and_issue", ["signalId", "issueId"]), + // ----------------------------------------------------------------- // Flue persistence stores (schema/format version 4). // ----------------------------------------------------------------- diff --git a/packages/backend/convex/signalRouting.test.ts b/packages/backend/convex/signalRouting.test.ts new file mode 100644 index 0000000..2e294c3 --- /dev/null +++ b/packages/backend/convex/signalRouting.test.ts @@ -0,0 +1,552 @@ +import { env } from "@code/env/convex"; +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"; + +declare global { + interface ImportMeta { + readonly glob: (pattern: string) => Record Promise>; + } +} + +const modules = import.meta.glob("./**/*.ts"); +const api = anyApi; + +const TOKEN = env.FLUE_DB_TOKEN; +const BAD_TOKEN = "not-the-right-token"; + +const ID_A = "https://convex.test|user-a"; +const identityA = { tokenIdentifier: ID_A }; + +const newTest = () => convexTest({ schema, modules }); + +const ensureOrg = async (t: TestConvex) => { + const org = await t + .withIdentity(identityA) + .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 seedMessage = async ( + t: TestConvex, + orgId: Id<"organizations">, + clientRequestId: string, + rawText: string +): Promise => { + const begun = await t + .withIdentity(identityA) + .mutation(api.conversationMessages.beginUserMessage, { + clientRequestId, + organizationId: orgId, + rawText, + }); + await t + .withIdentity(identityA) + .mutation(api.conversationMessages.markAdmitted, { + clientRequestId, + organizationId: orgId, + submissionId: `sub-${clientRequestId}`, + }); + return begun.messageId; +}; + +const createProject = async ( + t: TestConvex, + ownerId: string, + repoName: string +): Promise> => { + const ownerIdentity = { tokenIdentifier: ownerId }; + await t + .withIdentity(ownerIdentity) + .mutation(api.organizations.ensurePersonalOrganization); + const outcome = await t + .withIdentity(ownerIdentity) + .mutation(internal.projects.persistPublicGitImport, { + userId: ownerId, + source: { + host: "github.com", + projectName: repoName, + repositoryPath: `owner/${repoName}`, + normalizedUrl: `https://github.com/owner/${repoName}`, + url: `https://github.com/owner/${repoName}`, + }, + remote: { + defaultBranch: "main", + documents: [ + { + kind: "readme" as const, + path: "README.md", + content: `# ${repoName}\n\nTest.\n`, + }, + ], + warnings: [], + }, + }); + return outcome.id as unknown as Id<"projects">; +}; + +// Create a project-scoped signal via the routing backend. +const createRoutingSignal = async ( + t: TestConvex, + orgId: Id<"organizations">, + projectId: Id<"projects">, + messageIds: string[] +): Promise> => { + const result = await t.mutation(api.signalRouting.createSignal, { + messageIds, + organizationId: orgId, + problemStatement, + processedByAgentInstanceId: orgId, + projectId, + token: TOKEN, + }); + return result.signalId as Id<"signals">; +}; + +describe("signalRouting authentication", () => { + test("invalid token is rejected on every function", async () => { + const t = newTest(); + const orgId = await ensureOrg(t); + const projectId = await createProject(t, ID_A, "auth-test"); + const messageId = await seedMessage(t, orgId, "req-auth", "test message"); + + await expect( + t.mutation(api.signalRouting.createSignal, { + messageIds: [messageId], + organizationId: orgId, + problemStatement, + processedByAgentInstanceId: orgId, + projectId, + token: BAD_TOKEN, + }) + ).rejects.toThrow(/Invalid agent control token/u); + + await expect( + t.query(api.signalRouting.listEvidence, { + organizationId: orgId, + token: BAD_TOKEN, + }) + ).rejects.toThrow(/Invalid agent control token/u); + + await expect( + t.query(api.signalRouting.listActiveIssues, { + organizationId: orgId, + projectId, + token: BAD_TOKEN, + }) + ).rejects.toThrow(/Invalid agent control token/u); + + await expect( + t.query(api.signalRouting.listProjects, { + organizationId: orgId, + token: BAD_TOKEN, + }) + ).rejects.toThrow(/Invalid agent control token/u); + }); +}); + +describe("signalRouting create and route", () => { + test("create signal, list evidence, and list signals", async () => { + const t = newTest(); + const orgId = await ensureOrg(t); + const projectId = await createProject(t, ID_A, "route-test"); + const messageId = await seedMessage( + t, + orgId, + "req-route", + "The work-unit card should expose the generated PR." + ); + + // Evidence appears before signal creation. + const evidenceBefore = await t.query(api.signalRouting.listEvidence, { + organizationId: orgId, + token: TOKEN, + }); + expect(evidenceBefore).toHaveLength(1); + expect(evidenceBefore[0]?.rawText).toBe( + "The work-unit card should expose the generated PR." + ); + + // Create the signal. + const signalId = await createRoutingSignal(t, orgId, projectId, [ + messageId, + ]); + + // Evidence is consumed after signal creation. + const evidenceAfter = await t.query(api.signalRouting.listEvidence, { + organizationId: orgId, + token: TOKEN, + }); + expect(evidenceAfter).toHaveLength(0); + + // Signal appears in the project list. + const signals = await t.query(api.signalRouting.listSignals, { + organizationId: orgId, + projectId, + token: TOKEN, + }); + expect(signals).toHaveLength(1); + expect(signals[0]?._id).toBe(signalId); + expect(signals[0]?.problemStatement.title).toBe("Deploy fails on staging"); + }); + + test("attach signal to existing issue", async () => { + const t = newTest(); + const orgId = await ensureOrg(t); + const projectId = await createProject(t, ID_A, "attach-test"); + const messageId = await seedMessage(t, orgId, "req-attach", "problem"); + const signalId = await createRoutingSignal(t, orgId, projectId, [ + messageId, + ]); + + // Create an issue via the existing projectIssues backend. + const issueId = await t + .withIdentity(identityA) + .mutation(api.projectIssues.create, { + body: "Existing issue body that is long enough.", + projectId, + title: "Existing UI issue", + }); + + const result = await t.mutation(api.signalRouting.attachSignalToIssue, { + issueId: issueId as string, + organizationId: orgId, + signalId: signalId as string, + token: TOKEN, + }); + + expect(result.alreadyAttached).toBe(false); + + // List active issues includes the attached issue. + const issues = await t.query(api.signalRouting.listActiveIssues, { + organizationId: orgId, + projectId, + token: TOKEN, + }); + expect(issues).toHaveLength(1); + expect(issues[0]?._id).toBe(issueId); + }); + + test("create issue from signal (attach-vs-create fallback)", async () => { + const t = newTest(); + const orgId = await ensureOrg(t); + const projectId = await createProject(t, ID_A, "create-test"); + const messageId = await seedMessage(t, orgId, "req-create", "new problem"); + const signalId = await createRoutingSignal(t, orgId, projectId, [ + messageId, + ]); + + const result = await t.mutation(api.signalRouting.createIssueFromSignal, { + organizationId: orgId, + signalId: signalId as string, + token: TOKEN, + }); + + expect(result.created).toBe(true); + expect(result.projectId).toBe(projectId); + + // The issue exists and is open. + const issues = await t.query(api.signalRouting.listActiveIssues, { + organizationId: orgId, + projectId, + token: TOKEN, + }); + expect(issues).toHaveLength(1); + expect(issues[0]?.title).toBe("Deploy fails on staging"); + }); +}); + +describe("signalRouting idempotency", () => { + test("duplicate attach retry returns existing without duplicate event", async () => { + const t = newTest(); + const orgId = await ensureOrg(t); + const projectId = await createProject(t, ID_A, "idem-attach"); + const messageId = await seedMessage(t, orgId, "req-idem-att", "problem"); + const signalId = await createRoutingSignal(t, orgId, projectId, [ + messageId, + ]); + + const issueId = await t + .withIdentity(identityA) + .mutation(api.projectIssues.create, { + body: "Existing issue body that is long enough for validation.", + projectId, + title: "Pre-existing issue", + }); + + // First attach. + const first = await t.mutation(api.signalRouting.attachSignalToIssue, { + issueId: issueId as string, + organizationId: orgId, + signalId: signalId as string, + token: TOKEN, + }); + expect(first.alreadyAttached).toBe(false); + + // Retry the same attach. + const second = await t.mutation(api.signalRouting.attachSignalToIssue, { + issueId: issueId as string, + organizationId: orgId, + signalId: signalId as string, + token: TOKEN, + }); + expect(second.alreadyAttached).toBe(true); + expect(second.attachmentId).toBe(first.attachmentId); + + // Verify only one signal.attached event was emitted (not duplicated). + const events = await t + .withIdentity(identityA) + .query(api.projectIssues.events, { projectId }); + const attached = events.filter((e: { kind: string }) => + e.kind.includes("signal.attached") + ); + expect(attached).toHaveLength(1); + }); + + test("duplicate create-from-signal retry returns existing issue", async () => { + const t = newTest(); + const orgId = await ensureOrg(t); + const projectId = await createProject(t, ID_A, "idem-create"); + const messageId = await seedMessage(t, orgId, "req-idem-crt", "problem"); + const signalId = await createRoutingSignal(t, orgId, projectId, [ + messageId, + ]); + + // First create. + const first = await t.mutation(api.signalRouting.createIssueFromSignal, { + organizationId: orgId, + signalId: signalId as string, + token: TOKEN, + }); + expect(first.created).toBe(true); + + // Retry — should return existing, not create a new issue. + const second = await t.mutation(api.signalRouting.createIssueFromSignal, { + organizationId: orgId, + signalId: signalId as string, + token: TOKEN, + }); + expect(second.created).toBe(false); + expect(second.issueId).toBe(first.issueId); + + // Only one issue exists. + const issues = await t.query(api.signalRouting.listActiveIssues, { + organizationId: orgId, + projectId, + token: TOKEN, + }); + expect(issues).toHaveLength(1); + }); + + test("duplicate signal creation is idempotent (sourceKey)", async () => { + const t = newTest(); + const orgId = await ensureOrg(t); + const projectId = await createProject(t, ID_A, "idem-signal"); + const messageId = await seedMessage(t, orgId, "req-idem-sig", "test"); + + const first = await t.mutation(api.signalRouting.createSignal, { + messageIds: [messageId], + organizationId: orgId, + problemStatement, + processedByAgentInstanceId: orgId, + projectId, + token: TOKEN, + }); + const second = await t.mutation(api.signalRouting.createSignal, { + messageIds: [messageId], + organizationId: orgId, + problemStatement, + processedByAgentInstanceId: orgId, + projectId, + token: TOKEN, + }); + + expect(second.signalId).toBe(first.signalId); + }); +}); + +describe("signalRouting cross-project rejection", () => { + test("attach rejects when signal and issue are in different projects", async () => { + const t = newTest(); + const orgId = await ensureOrg(t); + const projectA = await createProject(t, ID_A, "cross-a"); + const projectB = await createProject(t, ID_A, "cross-b"); + + const messageId = await seedMessage(t, orgId, "req-cross", "problem"); + // Signal is scoped to project A. + const signalId = await createRoutingSignal(t, orgId, projectA, [messageId]); + + // Issue is in project B. + const issueId = await t + .withIdentity(identityA) + .mutation(api.projectIssues.create, { + body: "Issue in project B with enough body text.", + projectId: projectB, + title: "Project B issue", + }); + + await expect( + t.mutation(api.signalRouting.attachSignalToIssue, { + issueId: issueId as string, + organizationId: orgId, + signalId: signalId as string, + token: TOKEN, + }) + ).rejects.toThrow(/must belong to the same project/u); + }); + + test("attach rejects org-scoped signal (no projectId)", async () => { + const t = newTest(); + const orgId = await ensureOrg(t); + const projectId = await createProject(t, ID_A, "org-scope-test"); + + const messageId = await seedMessage(t, orgId, "req-org", "problem"); + + // Create an org-scoped signal (no projectId). + const signalId = await t.mutation(api.signalRouting.createSignal, { + messageIds: [messageId], + organizationId: orgId, + problemStatement, + processedByAgentInstanceId: orgId, + token: TOKEN, + }); + + const issueId = await t + .withIdentity(identityA) + .mutation(api.projectIssues.create, { + body: "Issue body that is long enough for the validator.", + projectId, + title: "Some issue", + }); + + await expect( + t.mutation(api.signalRouting.attachSignalToIssue, { + issueId: issueId as string, + organizationId: orgId, + signalId: signalId.signalId as string, + token: TOKEN, + }) + ).rejects.toThrow(/not project-scoped/u); + }); +}); + +describe("signalRouting provenance", () => { + test("exact message raw text is preserved through the routing loop", async () => { + const t = newTest(); + const orgId = await ensureOrg(t); + const projectId = await createProject(t, ID_A, "prov-test"); + const rawText = " The work-unit card should expose the generated PR. "; + const messageId = await seedMessage(t, orgId, "req-prov", rawText); + + // Evidence shows exact raw text. + const evidence = await t.query(api.signalRouting.listEvidence, { + organizationId: orgId, + token: TOKEN, + }); + expect(evidence[0]?.rawText).toBe(rawText); + + // Signal creation preserves exact snapshot. + const signalId = await createRoutingSignal(t, orgId, projectId, [ + messageId, + ]); + + // Verify via the existing signals.get that the snapshot is exact. + const composed = await t + .withIdentity(identityA) + .query(api.signals.get, { signalId }); + expect(composed?.sources[0]?.rawTextSnapshot).toBe(rawText); + }); +}); + +describe("signalRouting begin issue", () => { + test("begin transitions an open issue to queued", async () => { + const t = newTest(); + const orgId = await ensureOrg(t); + const projectId = await createProject(t, ID_A, "begin-test"); + const messageId = await seedMessage(t, orgId, "req-begin", "start this"); + const signalId = await createRoutingSignal(t, orgId, projectId, [ + messageId, + ]); + + const result = await t.mutation(api.signalRouting.createIssueFromSignal, { + organizationId: orgId, + signalId: signalId as string, + token: TOKEN, + }); + + const status = await t.mutation(api.signalRouting.beginIssue, { + issueId: result.issueId as string, + organizationId: orgId, + token: TOKEN, + }); + + expect(status).toBe("queued"); + }); + + test("begin rejects invalid token", async () => { + const t = newTest(); + const orgId = await ensureOrg(t); + const projectId = await createProject(t, ID_A, "begin-auth"); + const messageId = await seedMessage(t, orgId, "req-begin-auth", "msg"); + const signalId = await createRoutingSignal(t, orgId, projectId, [ + messageId, + ]); + + const result = await t.mutation(api.signalRouting.createIssueFromSignal, { + organizationId: orgId, + signalId: signalId as string, + token: TOKEN, + }); + + await expect( + t.mutation(api.signalRouting.beginIssue, { + issueId: result.issueId as string, + organizationId: orgId, + token: BAD_TOKEN, + }) + ).rejects.toThrow(/Invalid agent control token/u); + }); +}); + +describe("signalRouting project context", () => { + test("get project context returns project and documents", async () => { + const t = newTest(); + const orgId = await ensureOrg(t); + const projectId = await createProject(t, ID_A, "ctx-test"); + + const context = await t.query(api.signalRouting.getProjectContext, { + organizationId: orgId, + projectId, + token: TOKEN, + }); + + expect(context.project._id).toBe(projectId); + expect(context.project.name).toBe("ctx-test"); + expect(context.contextDocuments.length).toBeGreaterThan(0); + }); + + test("list projects returns organization projects", async () => { + const t = newTest(); + const orgId = await ensureOrg(t); + const projectId = await createProject(t, ID_A, "list-proj-test"); + + const projects = await t.query(api.signalRouting.listProjects, { + organizationId: orgId, + token: TOKEN, + }); + + expect(projects).toHaveLength(1); + expect(projects[0]?._id).toBe(projectId); + }); +}); diff --git a/packages/backend/convex/signalRouting.ts b/packages/backend/convex/signalRouting.ts new file mode 100644 index 0000000..3dc4111 --- /dev/null +++ b/packages/backend/convex/signalRouting.ts @@ -0,0 +1,721 @@ +import { env } from "@code/env/convex"; +import { + projectIssueDraftFromSignal, + queueProjectIssue, + type ProjectIssueDraft, +} from "@code/primitives/project-issue"; +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 { + mutation, + type MutationCtx, + query, + type QueryCtx, +} from "./_generated/server"; + +// --------------------------------------------------------------------------- +// Agent token guard (service-level auth for Zopu tools). +// --------------------------------------------------------------------------- + +const requireAgent = (token: string) => { + if (token !== env.FLUE_DB_TOKEN) { + throw new ConvexError("Invalid agent control token"); + } +}; + +const PROBLEM_STATEMENT = v.object({ + title: v.string(), + summary: v.string(), + desiredOutcome: v.string(), + constraints: v.array(v.string()), +}); + +// --------------------------------------------------------------------------- +// View helpers +// --------------------------------------------------------------------------- + +interface EvidenceMessage { + readonly messageId: string; + readonly rawText: string; + readonly createdAt: number; + readonly submissionId: string | null; +} + +interface ActiveIssueView { + readonly _id: Id<"projectIssues">; + readonly number: number; + readonly title: string; + readonly body: string; + readonly status: string; + readonly projectId: Id<"projects">; + readonly updatedAt: number; +} + +interface ProjectSummaryView { + readonly _id: Id<"projects">; + readonly name: string; + readonly description: string | null; + readonly organizationId: Id<"organizations">; +} + +interface ContextDocumentView { + readonly kind: string; + readonly path: string; + readonly content: string; + readonly revision: number; +} + +interface SignalSummaryView { + readonly _id: Id<"signals">; + readonly problemStatement: { + readonly title: string; + readonly summary: string; + readonly desiredOutcome: string; + readonly constraints: readonly string[]; + }; + readonly projectId: Id<"projects"> | null; + readonly createdAt: number; +} + +// --------------------------------------------------------------------------- +// Organization + project authorization (agent-gated, no user identity). +// The agent instance id IS the organization id; the tool never supplies +// tenancy. We verify the organization exists and the project belongs to it. +// --------------------------------------------------------------------------- + +const authorizeProjectInOrg = async ( + ctx: MutationCtx | QueryCtx, + 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"); + } + return project; +}; + +const authorizeSignalInOrg = async ( + ctx: MutationCtx | QueryCtx, + organizationId: Id<"organizations">, + signalId: Id<"signals"> +): Promise> => { + const signal = await ctx.db.get(signalId); + if (!signal) { + throw new ConvexError("Signal not found"); + } + if (signal.organizationId !== organizationId) { + throw new ConvexError("Signal does not belong to the target organization"); + } + return signal; +}; + +// --------------------------------------------------------------------------- +// 1. List admitted user messages not yet consumed by a Signal. +// --------------------------------------------------------------------------- + +export const listEvidence = query({ + args: { + organizationId: v.id("organizations"), + token: v.string(), + }, + handler: async (ctx, args): Promise => { + requireAgent(args.token); + 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: EvidenceMessage[] = []; + 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({ + createdAt: message.createdAt, + messageId: message.messageId, + rawText: message.rawText, + submissionId: message.submissionId ?? null, + }); + } + } + + return unused; + }, +}); + +// --------------------------------------------------------------------------- +// 2. Create a Signal from selected messages + structured problem statement. +// --------------------------------------------------------------------------- + +const buildSourceKey = ( + organizationId: Id<"organizations">, + conversationId: string, + messageIds: readonly string[] +): string => JSON.stringify([organizationId, conversationId, ...messageIds]); + +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; +}; + +export const createSignal = mutation({ + args: { + organizationId: v.id("organizations"), + projectId: v.optional(v.id("projects")), + messageIds: v.array(v.string()), + problemStatement: PROBLEM_STATEMENT, + processedByAgentInstanceId: v.string(), + token: v.string(), + }, + handler: async (ctx, args): Promise<{ signalId: Id<"signals"> }> => { + requireAgent(args.token); + if (args.projectId) { + await authorizeProjectInOrg(ctx, args.organizationId, args.projectId); + } + + const conversationId = args.organizationId; + if (conversationId !== args.organizationId) { + throw new ConvexError("Conversation does not belong to organization"); + } + + const sourceKey = buildSourceKey( + args.organizationId, + 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, + conversationId, + args.messageIds + ); + + 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: { + agentInstanceId: args.processedByAgentInstanceId, + agentName: "zopu", + }, + 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, + createdAt: now, + organizationId: args.organizationId, + problemStatement: { + constraints: [...signal.problemStatement.constraints], + desiredOutcome: signal.problemStatement.desiredOutcome, + summary: signal.problemStatement.summary, + title: signal.problemStatement.title, + }, + processedByAgentInstanceId: args.processedByAgentInstanceId, + processedByAgentName: "zopu", + ...(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 }; + }, +}); + +// --------------------------------------------------------------------------- +// 3. List recent Signals for a project (or organization-wide when no project). +// --------------------------------------------------------------------------- + +export const listSignals = query({ + args: { + organizationId: v.id("organizations"), + projectId: v.optional(v.id("projects")), + token: v.string(), + }, + handler: async (ctx, args): Promise => { + requireAgent(args.token); + if (args.projectId) { + await authorizeProjectInOrg(ctx, args.organizationId, args.projectId); + const signals = await ctx.db + .query("signals") + .withIndex("by_project_and_createdAt", (q) => + q.eq("projectId", args.projectId as Id<"projects">) + ) + .order("desc") + .take(20); + return signals.map((s) => ({ + _id: s._id, + createdAt: s.createdAt, + problemStatement: s.problemStatement, + projectId: s.projectId ?? null, + })); + } + + const signals = await ctx.db + .query("signals") + .withIndex("by_organization_and_createdAt", (q) => + q.eq("organizationId", args.organizationId) + ) + .order("desc") + .take(20); + return signals.map((s) => ({ + _id: s._id, + createdAt: s.createdAt, + problemStatement: s.problemStatement, + projectId: s.projectId ?? null, + })); + }, +}); + +// --------------------------------------------------------------------------- +// 4. List active (open/queued/working/needs-input) ProjectIssues for a project. +// --------------------------------------------------------------------------- + +export const listActiveIssues = query({ + args: { + organizationId: v.id("organizations"), + projectId: v.id("projects"), + token: v.string(), + }, + handler: async (ctx, args): Promise => { + requireAgent(args.token); + await authorizeProjectInOrg(ctx, args.organizationId, args.projectId); + + const issues = await ctx.db + .query("projectIssues") + .withIndex("by_project_and_status", (q) => + q.eq("projectId", args.projectId) + ) + .take(100); + + return issues + .filter( + (issue) => + issue.status === "open" || + issue.status === "queued" || + issue.status === "working" || + issue.status === "needs-input" + ) + .map((issue) => ({ + _id: issue._id, + body: issue.body, + number: issue.number, + projectId: issue.projectId, + status: issue.status, + title: issue.title, + updatedAt: issue.updatedAt, + })) + .sort((a, b) => b.updatedAt - a.updatedAt); + }, +}); + +// --------------------------------------------------------------------------- +// 5. Attach a Signal to an existing ProjectIssue (idempotent). +// --------------------------------------------------------------------------- + +export const attachSignalToIssue = mutation({ + args: { + organizationId: v.id("organizations"), + signalId: v.id("signals"), + issueId: v.id("projectIssues"), + token: v.string(), + }, + handler: async ( + ctx, + args + ): Promise<{ + attachmentId: Id<"signalIssueAttachments">; + alreadyAttached: boolean; + }> => { + requireAgent(args.token); + const signal = await authorizeSignalInOrg( + ctx, + args.organizationId, + args.signalId + ); + const issue = await ctx.db.get(args.issueId); + if (!issue) { + throw new ConvexError("Issue not found"); + } + await authorizeProjectInOrg(ctx, args.organizationId, issue.projectId); + + // Cross-project rejection: the signal must be project-scoped and its + // project must match the issue's project. Organization equality alone is + // insufficient — it would permit attaching across different projects in + // the same org. + if (!signal.projectId) { + throw new ConvexError("Signal is not project-scoped"); + } + if (signal.projectId !== issue.projectId) { + throw new ConvexError("Signal and issue must belong to the same project"); + } + + // Idempotent: if the attachment already exists, return it without + // emitting a duplicate event. + const existing = await ctx.db + .query("signalIssueAttachments") + .withIndex("by_signal_and_issue", (q) => + q.eq("signalId", args.signalId).eq("issueId", args.issueId) + ) + .unique(); + if (existing) { + return { + alreadyAttached: true, + attachmentId: existing._id, + }; + } + + const now = Date.now(); + const attachmentId = await ctx.db.insert("signalIssueAttachments", { + createdAt: now, + issueId: args.issueId, + organizationId: args.organizationId, + projectId: issue.projectId, + signalId: args.signalId, + }); + + await ctx.db.insert("projectEvents", { + createdAt: now, + data: { + signalProblemTitle: signal.problemStatement.title, + signalId: String(signal._id), + }, + issueId: args.issueId, + kind: "signal.attached", + projectId: issue.projectId, + }); + + return { alreadyAttached: false, attachmentId }; + }, +}); + +// --------------------------------------------------------------------------- +// 6. Create a new ProjectIssue from a Signal. +// --------------------------------------------------------------------------- + +const insertIssue = async ( + ctx: MutationCtx, + projectId: Id<"projects">, + draft: ProjectIssueDraft +): Promise> => { + const latest = await ctx.db + .query("projectIssues") + .withIndex("by_project_and_number", (q) => q.eq("projectId", projectId)) + .order("desc") + .first(); + const timestamp = Date.now(); + const number = (latest?.number ?? 0) + 1; + const issueId = await ctx.db.insert("projectIssues", { + body: draft.body, + createdAt: timestamp, + number, + projectId, + status: "open", + title: draft.title, + updatedAt: timestamp, + }); + await ctx.db.insert("projectEvents", { + createdAt: timestamp, + data: { number, source: "signal", title: draft.title }, + issueId, + kind: "issue.created", + projectId, + }); + return issueId; +}; + +export const createIssueFromSignal = mutation({ + args: { + organizationId: v.id("organizations"), + signalId: v.id("signals"), + token: v.string(), + }, + handler: async ( + ctx, + args + ): Promise<{ + created: boolean; + issueId: Id<"projectIssues">; + projectId: Id<"projects">; + }> => { + requireAgent(args.token); + const signal = await authorizeSignalInOrg( + ctx, + args.organizationId, + args.signalId + ); + if (!signal.projectId) { + throw new ConvexError("Signal is not project-scoped"); + } + await authorizeProjectInOrg(ctx, args.organizationId, signal.projectId); + + // Idempotent: if any attachment already exists for this signal, return + // the existing issue without creating a duplicate. This prevents Flue + // retries from producing duplicate issues/attachments. + const existingAttachment = await ctx.db + .query("signalIssueAttachments") + .withIndex("by_signal", (q) => q.eq("signalId", args.signalId)) + .first(); + if (existingAttachment) { + return { + created: false, + issueId: existingAttachment.issueId, + projectId: existingAttachment.projectId, + }; + } + + const draft = await Effect.runPromise( + projectIssueDraftFromSignal({ + problemStatement: signal.problemStatement, + signalId: String(signal._id), + }) + ).catch((error: unknown) => { + throw new ConvexError( + error instanceof Error ? error.message : "Invalid project issue" + ); + }); + + const issueId = await insertIssue(ctx, signal.projectId, draft); + + // Auto-attach the signal to the new issue. + const now = Date.now(); + await ctx.db.insert("signalIssueAttachments", { + createdAt: now, + issueId, + organizationId: args.organizationId, + projectId: signal.projectId, + signalId: signal._id, + }); + + return { created: true, issueId, projectId: signal.projectId }; + }, +}); + +// --------------------------------------------------------------------------- +// 7. Begin a ProjectIssue (transition to queued/working). +// --------------------------------------------------------------------------- + +export const beginIssue = mutation({ + args: { + organizationId: v.id("organizations"), + issueId: v.id("projectIssues"), + token: v.string(), + }, + handler: async (ctx, args): Promise<"queued" | "working"> => { + requireAgent(args.token); + const issue = await ctx.db.get(args.issueId); + if (!issue) { + throw new ConvexError("Issue not found"); + } + await authorizeProjectInOrg(ctx, args.organizationId, issue.projectId); + + const status = await Effect.runPromise( + queueProjectIssue(issue.status) + ).catch((error: unknown) => { + throw new ConvexError( + error instanceof Error ? error.message : "Invalid issue status" + ); + }); + if (status === issue.status) { + return status; + } + + const timestamp = Date.now(); + await ctx.db.patch(issue._id, { + status, + updatedAt: timestamp, + }); + await ctx.db.insert("projectEvents", { + createdAt: timestamp, + data: { number: issue.number }, + issueId: issue._id, + kind: "issue.queued", + projectId: issue.projectId, + }); + return status; + }, +}); + +// --------------------------------------------------------------------------- +// 8. Get project summary + context documents for routing decisions. +// --------------------------------------------------------------------------- + +export const getProjectContext = query({ + args: { + organizationId: v.id("organizations"), + projectId: v.id("projects"), + token: v.string(), + }, + handler: async ( + ctx, + args + ): Promise<{ + project: ProjectSummaryView; + contextDocuments: ContextDocumentView[]; + }> => { + requireAgent(args.token); + const project = await authorizeProjectInOrg( + ctx, + args.organizationId, + args.projectId + ); + + const docs = await ctx.db + .query("projectContextDocuments") + .withIndex("by_project", (q) => q.eq("projectId", args.projectId)) + .take(25); + + return { + contextDocuments: docs.map((doc) => ({ + content: doc.content, + kind: doc.kind, + path: doc.path, + revision: doc.revision, + })), + project: { + _id: project._id, + description: project.description ?? null, + name: project.name, + organizationId: project.organizationId, + }, + }; + }, +}); + +// --------------------------------------------------------------------------- +// 9. List projects in the organization (for selecting a project context). +// --------------------------------------------------------------------------- + +export const listProjects = query({ + args: { + organizationId: v.id("organizations"), + token: v.string(), + }, + handler: async (ctx, args): Promise => { + requireAgent(args.token); + const projects = await ctx.db + .query("projects") + .withIndex("by_organization_and_createdAt", (q) => + q.eq("organizationId", args.organizationId) + ) + .take(50); + return projects.map((p) => ({ + _id: p._id, + description: p.description ?? null, + name: p.name, + organizationId: p.organizationId, + })); + }, +});