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
This commit is contained in:
-Puter
2026-07-24 20:48:18 +05:30
parent 1324316fee
commit 03fc32bd6d
5 changed files with 1658 additions and 6 deletions

View File

@@ -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).
// -----------------------------------------------------------------

View File

@@ -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<string, () => Promise<unknown>>;
}
}
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<typeof schema>) => {
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<typeof schema>,
orgId: Id<"organizations">,
clientRequestId: string,
rawText: string
): Promise<string> => {
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<typeof schema>,
ownerId: string,
repoName: string
): Promise<Id<"projects">> => {
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<typeof schema>,
orgId: Id<"organizations">,
projectId: Id<"projects">,
messageIds: string[]
): Promise<Id<"signals">> => {
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);
});
});

View File

@@ -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<Doc<"projects">> => {
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<Doc<"signals">> => {
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<EvidenceMessage[]> => {
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<Doc<"conversationMessages">[]> => {
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<SignalSummaryView[]> => {
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<ActiveIssueView[]> => {
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<Id<"projectIssues">> => {
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<ProjectSummaryView[]> => {
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,
}));
},
});