From d3381f978f73af2d5969b5d910c9149e49b556bf Mon Sep 17 00:00:00 2001 From: -Puter <22245429+puterhimself@users.noreply.github.com> Date: Thu, 23 Jul 2026 17:16:59 +0530 Subject: [PATCH] feat(agents): add signal tools to zopu agent --- packages/agents/src/agents/zopu.ts | 12 ++- packages/agents/src/tools/signals.ts | 153 +++++++++++++++++++++++++++ packages/backend/convex/signals.ts | 81 ++++++++++++++ 3 files changed, 244 insertions(+), 2 deletions(-) create mode 100644 packages/agents/src/tools/signals.ts diff --git a/packages/agents/src/agents/zopu.ts b/packages/agents/src/agents/zopu.ts index dbf795d..757f79f 100644 --- a/packages/agents/src/agents/zopu.ts +++ b/packages/agents/src/agents/zopu.ts @@ -18,8 +18,16 @@ export default defineAgent(({ env }) => { return { cwd: repositoryRoot, description: "A simple conversational agent with persistent history.", - instructions: - "You are Zopu, a concise and helpful assistant. Answer the user's request directly. Ask a clarifying question only when the request cannot be completed without one.", + instructions: `You are Zopu, a concise and helpful assistant. Answer the user's request directly. Ask a clarifying question only when the request cannot be completed without one. + +## Signals +A Signal is your structured interpretation of the user's exact messages into a clear problem statement. Capture them precisely. + +- Call list_signal_source_messages to see admitted user messages not yet used as evidence. +- Do NOT create a Signal for greetings, chitchat, or unclear exploration. Just respond. +- When the user's request is coherent and describes a real problem or goal, compose exactly one Signal: call create_signal once with the ordered messageIds and a structured problem statement {title, summary, desiredOutcome, constraints}. +- Never call create_signal more than once for the same request. It is idempotent on the ordered message selection, but you should still create at most one. +- After creating, report the structured problem statement back to the user (title, summary, desired outcome, constraints) so they can confirm or correct it.`, model: `${AGENT_MODEL_PROVIDER}/${AGENT_MODEL_NAME}`, // sandbox: agentos(), sandbox: local({ cwd: repositoryRoot }), diff --git a/packages/agents/src/tools/signals.ts b/packages/agents/src/tools/signals.ts new file mode 100644 index 0000000..498af7d --- /dev/null +++ b/packages/agents/src/tools/signals.ts @@ -0,0 +1,153 @@ +/** + * Zopu Signal tools: candidate-evidence listing and structured Signal creation. + * + * SECURITY MODEL + * -------------- + * Both tools call Convex functions (`signals:listAdmittedUnused`, + * `signals:createFromMessages`) that authorize via the authenticated caller + * identity (`requireOrganizationMember` → `ctx.auth.getUserIdentity()`). They + * therefore require a request-scoped {@link ConvexHttpClient} carrying the + * caller's Bearer JWT — exactly the client `authenticatedAgentRoute` builds + * per request. We never use a shared/global token or a service token here, + * because those carry no user identity and cannot satisfy membership checks. + * + * The factory closes over an authenticated client plus the resolved + * organization (agent instance) id. The agent instance id equals the + * organization id (the hard tenancy boundary enforced by the route), so + * `organizationId` and `conversationId` are never trusted from the agent. + */ +import type { Id } from "@code/backend/convex/_generated/dataModel"; +import { defineTool } from "@flue/runtime"; +import type { ConvexHttpClient } from "convex/browser"; +import { makeFunctionReference } from "convex/server"; +import * as v from "valibot"; + +// --------------------------------------------------------------------------- +// Typed function references (no generated backend bindings imported here, to +// keep the agents package decoupled from backend codegen at type-check time). +// --------------------------------------------------------------------------- + +interface ConversationMessageSourceView { + readonly messageId: string; + readonly clientRequestId: string; + readonly rawText: string; + readonly submissionId: string | null; + readonly createdAt: number; +} + +const listAdmittedUnused = makeFunctionReference< + "query", + { readonly organizationId: Id<"organizations"> }, + ConversationMessageSourceView[] +>("signals:listAdmittedUnused"); + +const createFromMessages = makeFunctionReference< + "mutation", + { + readonly organizationId: Id<"organizations">; + readonly projectId?: Id<"projects">; + readonly conversationId: string; + readonly messageIds: readonly string[]; + readonly problemStatement: { + readonly title: string; + readonly summary: string; + readonly desiredOutcome: string; + readonly constraints: readonly string[]; + }; + readonly processedBy: { + readonly agentName: string; + readonly agentInstanceId: string; + }; + }, + { readonly signalId: Id<"signals"> } +>("signals:createFromMessages"); + +/** + * Input/output schemas shared with the agent instructions. + */ +const problemStatementSchema = v.object({ + title: v.pipe(v.string(), v.maxLength(200)), + summary: v.pipe(v.string(), v.maxLength(2000)), + desiredOutcome: v.pipe(v.string(), v.maxLength(2000)), + constraints: v.pipe(v.array(v.pipe(v.string(), v.maxLength(500))), v.maxLength(20)), +}); + +/** + * Build the Zopu Signal tools bound to one authenticated, organization-scoped + * agent instance. + * + * @param client A request-scoped authenticated Convex client carrying + * the caller's JWT. Built fresh per request; never shared. + * @param organizationId The caller's organization id. Equals the agent + * instance id and the global conversation id. + */ +export const createSignalTools = ( + client: ConvexHttpClient, + organizationId: Id<"organizations"> +) => { + // The conversation id is the organization id (global Zopu conversation). + const conversationId: string = organizationId; + const processedBy = { + agentInstanceId: organizationId, + agentName: "zopu", + }; + + return [ + defineTool({ + description: + "List the admitted user messages in this conversation that have not yet been used as Signal evidence, newest first. Use these messageIds as the ordered source selection when composing a Signal.", + name: "list_signal_source_messages", + output: v.object({ + messages: v.array( + v.object({ + messageId: v.string(), + rawText: v.string(), + createdAt: v.number(), + }) + ), + }), + async run() { + const messages = await client.query(listAdmittedUnused, { + organizationId, + }); + return { + messages: messages.map((message) => ({ + createdAt: message.createdAt, + messageId: message.messageId, + rawText: message.rawText, + })), + }; + }, + }), + defineTool({ + description: + "Create exactly one Signal from an ordered selection of admitted user messages plus a structured problem statement. Idempotent: the same ordered message selection returns the existing Signal. Call this at most once per coherent request, never for greetings or unclear exploration.", + input: v.object({ + messageIds: v.pipe( + v.array(v.string()), + v.minLength(1), + v.maxLength(50) + ), + problemStatement: problemStatementSchema, + projectId: v.optional(v.pipe(v.string(), v.minLength(1))), + }), + name: "create_signal", + output: v.object({ + signalId: v.string(), + }), + async run({ input }) { + const result = await client.mutation(createFromMessages, { + conversationId, + messageIds: [...input.messageIds], + organizationId, + problemStatement: input.problemStatement, + processedBy, + ...(input.projectId === undefined + ? {} + : { projectId: input.projectId as Id<"projects"> }), + }); + return { signalId: result.signalId }; + }, + }), + ]; +}; diff --git a/packages/backend/convex/signals.ts b/packages/backend/convex/signals.ts index 2046478..b3f9915 100644 --- a/packages/backend/convex/signals.ts +++ b/packages/backend/convex/signals.ts @@ -53,6 +53,19 @@ interface ComposedSignal { readonly sources: readonly SignalSourceView[]; } +/** + * An admitted user conversation message presented to the agent as candidate + * Signal evidence. The agent selects from these by `messageId`; raw text and + * timestamps are copied server-side and never accepted from the agent. + */ +interface ConversationMessageSourceView { + readonly messageId: string; + readonly clientRequestId: string; + readonly rawText: string; + readonly submissionId: string | null; + readonly createdAt: number; +} + const toSignalView = (doc: Doc<"signals">): SignalView => ({ _creationTime: doc._creationTime, _id: doc._id, @@ -78,6 +91,16 @@ const toSourceView = (doc: Doc<"signalSources">): SignalSourceView => ({ submissionId: doc.submissionId ?? null, }); +const toConversationMessageSourceView = ( + doc: Doc<"conversationMessages"> +): ConversationMessageSourceView => ({ + clientRequestId: doc.clientRequestId, + createdAt: doc.createdAt, + messageId: doc.messageId, + rawText: doc.rawText, + submissionId: doc.submissionId ?? null, +}); + /** * Build a deterministic, collision-safe key from the organization, the * conversation, and the ordered message selection. The key is the JSON @@ -348,6 +371,64 @@ export const list = query({ }, }); +/** + * List the bounded set of admitted user messages in the organization's + * conversation that have not yet been consumed as a Signal source, newest + * first. These are the candidate messages an agent should select from when + * composing a new Signal. + * + * Authorization: the authenticated identity must be a member of the target + * organization. The conversation id equals the organization id; the caller + * never supplies tenancy. Cross-organization access is denied. + * + * Bounded: capped at 100 rows to keep agent evidence selection tractable. + */ +export const listAdmittedUnused = query({ + args: { + organizationId: v.id("organizations"), + }, + handler: async ( + ctx, + args + ): Promise => { + await requireOrganizationMember(ctx, args.organizationId); + const conversationId = args.organizationId; + + const admitted = await ctx.db + .query("conversationMessages") + .withIndex("by_organization_and_message", (q) => + q + .eq("organizationId", args.organizationId) + .eq("conversationId", conversationId) + ) + .order("desc") + .take(100); + + const candidates = admitted.filter( + (message) => + message.role === "user" && message.status === "admitted" + ); + + const unused: ConversationMessageSourceView[] = []; + for (const message of candidates) { + const consumed = await ctx.db + .query("signalSources") + .withIndex("by_organization_and_message", (q) => + q + .eq("organizationId", args.organizationId) + .eq("conversationId", conversationId) + .eq("messageId", message.messageId) + ) + .first(); + if (!consumed) { + unused.push(toConversationMessageSourceView(message)); + } + } + + return unused; + }, +}); + /** * Get one Signal by id with its composed sources. Membership in the signal's * organization is required; access to another organization's signal is denied.