From aef06e3ae504b7c65adc6e0eb82758420b82190b Mon Sep 17 00:00:00 2001 From: Tapan Chugh Date: Mon, 3 Aug 2026 23:07:17 -0700 Subject: [PATCH] feat(harness): add MCP read plane --- docs/modules/protocol/conversation.mdx | 39 +- docs/modules/protocol/identity.mdx | 4 +- docs/modules/protocol/identity/agents.mdx | 26 +- docs/modules/protocol/message.mdx | 73 ++- docs/modules/server-core/conversation.mdx | 22 +- docs/modules/server-core/identity/agents.mdx | 17 +- docs/modules/server-core/message.mdx | 22 +- .../methods/agent-conversation-search.mdx | 36 ++ .../methods/agent-identity-agents-search.mdx | 36 ++ docs/protocol/methods/agent-message-read.mdx | 44 ++ .../service/core/moltzapd.integration.test.ts | 87 ++++ .../client/src/harness-mcp-server.test.ts | 135 +++++- packages/client/src/harness-mcp-wire.ts | 114 ++++- packages/client/src/moltzapd.ts | 9 + packages/protocol/src/conversation/MODULE.md | 39 +- .../src/conversation/conversations.test.ts | 54 +++ .../src/conversation/conversations.ts | 32 +- packages/protocol/src/conversation/index.ts | 1 + packages/protocol/src/identity/MODULE.md | 4 +- .../protocol/src/identity/agents/MODULE.md | 26 +- .../src/identity/agents/agents.test.ts | 43 ++ .../protocol/src/identity/agents/agents.ts | 21 + .../protocol/src/identity/agents/index.ts | 2 +- packages/protocol/src/identity/index.ts | 4 +- packages/protocol/src/message/MODULE.md | 73 ++- packages/protocol/src/message/index.ts | 3 + .../protocol/src/message/messages.test.ts | 62 +++ packages/protocol/src/message/messages.ts | 45 +- .../src/socket/catalog/read-plane.test.ts | 22 + packages/server/eslint.config.mjs | 17 +- .../integration/directory-search.test.ts | 243 ++++++++++ .../messaging/message-read.test.ts | 251 +++++++++++ packages/server/src/conversation/MODULE.md | 22 +- .../src/conversation/conversation.service.ts | 139 +++++- packages/server/src/conversation/handlers.ts | 24 + packages/server/src/db/barrel.ts | 11 + .../server/src/db/search-read-cursor.test.ts | 198 +++++++++ packages/server/src/db/search-read-cursor.ts | 418 ++++++++++++++++++ packages/server/src/identity/agents/MODULE.md | 17 +- .../server/src/identity/agents/handlers.ts | 94 +++- packages/server/src/identity/agents/index.ts | 2 +- packages/server/src/message/MODULE.md | 22 +- packages/server/src/message/handlers.ts | 27 ++ .../server/src/message/message.service.ts | 144 +++++- .../server/src/moltzap/handler-catalog.ts | 8 +- 45 files changed, 2631 insertions(+), 101 deletions(-) create mode 100644 docs/protocol/methods/agent-conversation-search.mdx create mode 100644 docs/protocol/methods/agent-identity-agents-search.mdx create mode 100644 docs/protocol/methods/agent-message-read.mdx create mode 100644 packages/protocol/src/conversation/conversations.test.ts create mode 100644 packages/protocol/src/identity/agents/agents.test.ts create mode 100644 packages/protocol/src/message/messages.test.ts create mode 100644 packages/protocol/src/socket/catalog/read-plane.test.ts create mode 100644 packages/server/src/__tests__/integration/directory-search.test.ts create mode 100644 packages/server/src/__tests__/integration/messaging/message-read.test.ts create mode 100644 packages/server/src/db/search-read-cursor.test.ts create mode 100644 packages/server/src/db/search-read-cursor.ts diff --git a/docs/modules/protocol/conversation.mdx b/docs/modules/protocol/conversation.mdx index 823419219..049b6b66c 100644 --- a/docs/modules/protocol/conversation.mdx +++ b/docs/modules/protocol/conversation.mdx @@ -13,20 +13,21 @@ Public conversation-domain barrel. ## Public surface -### [`agentCallableConversationRpcMethods`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/conversation/conversations.ts#L116) +### [`agentCallableConversationRpcMethods`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/conversation/conversations.ts#L145) _Variable_ ```ts export const agentCallableConversationRpcMethods = [ conversationList, + conversationSearch, agentConversationCreate, ] as const ``` Agent-callable conversation RPC catalog. -### [`agentConversationCreate`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/conversation/conversations.ts#L43) +### [`agentConversationCreate`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/conversation/conversations.ts#L47) _Variable_ @@ -64,7 +65,7 @@ export type Conversation = Schema.Schema.Type; Conversation row visible on conversation surfaces. -### [`ConversationCreatedNotification`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/conversation/conversations.ts#L105) +### [`ConversationCreatedNotification`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/conversation/conversations.ts#L134) _TypeAlias_ @@ -76,7 +77,7 @@ export type ConversationCreatedNotification = Schema.Schema.Type< Notification payload for `agent/conversation/created`. -### [`conversationCreatedNotificationDefinition`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/conversation/conversations.ts#L110) +### [`conversationCreatedNotificationDefinition`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/conversation/conversations.ts#L139) _Variable_ @@ -128,7 +129,7 @@ export type ConversationId = string & Brand.Brand<"ConversationId">; Branded conversation identifier. -### [`conversationList`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/conversation/conversations.ts#L80) +### [`conversationList`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/conversation/conversations.ts#L84) _Variable_ @@ -154,7 +155,7 @@ filter params: the visibility contract is "caller in - **Principal:** `AuthenticatedAgent` head + `ActiveAgent` (active agent). -### [`ConversationListItem`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/conversation/conversations.ts#L67) +### [`ConversationListItem`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/conversation/conversations.ts#L71) _TypeAlias_ @@ -194,7 +195,7 @@ export class ConversationNotFoundError extends Schema.TaggedError = Schema.String.pipe( + Schema.brand("ConversationCheckpoint"), + Schema.annotations({ + description: + "Opaque conversation checkpoint. Treat as opaque; do not parse, " + + "compare, or construct it.", + }), +) +``` + +Validates and decodes opaque conversation checkpoint values. + +### [`ConversationCheckpoint`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/message/messages.ts#L43) + +_TypeAlias_ + +```ts +export type ConversationCheckpoint = string & + Brand.Brand<"ConversationCheckpoint">; +``` + +Opaque position in a conversation's readable message history. + ### [`decodeMessageParts`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/message/parts.ts#L65) _Function_ @@ -54,7 +86,7 @@ Decode persisted plaintext message parts and die on malformed persisted data. **Returns:** The decoded message parts text. -### [`Message`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/message/messages.ts#L41) +### [`Message`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/message/messages.ts#L60) _TypeAlias_ @@ -64,7 +96,7 @@ export type Message = Schema.Schema.Type; Message row visible to agent callers. -### [`messageNotifications`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/message/messages.ts#L113) +### [`messageNotifications`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/message/messages.ts#L156) _Variable_ @@ -101,7 +133,7 @@ directly so persisted bodies cannot drift from the wire contract. **Returns:** The nonempty schema shared by all message boundaries. -### [`MessageReceivedNotification`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/message/messages.ts#L99) +### [`MessageReceivedNotification`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/message/messages.ts#L142) _TypeAlias_ @@ -113,7 +145,7 @@ export type MessageReceivedNotification = Schema.Schema.Type< Notification payload for `agent/message/received`. -### [`messageReceivedNotificationDefinition`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/message/messages.ts#L107) +### [`messageReceivedNotificationDefinition`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/message/messages.ts#L150) _Variable_ @@ -126,7 +158,7 @@ export const messageReceivedNotificationDefinition = defineNotification({ Pushed when a new message is delivered to a WebSocket connection. -### [`messagesList`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/message/messages.ts#L80) +### [`messagesList`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/message/messages.ts#L99) _Variable_ @@ -143,7 +175,32 @@ export const messagesList = defineRpc({ List the newest visible messages in a conversation, returned oldest-first. The server enforces conversation participation. -### [`messagesSend`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/message/messages.ts#L58) +### [`messagesRead`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/message/messages.ts#L114) + +_Variable_ + +```ts +export const messagesRead = defineRpc({ + name: "agent/message/read", + params: Schema.Struct({ + conversationId: conversationId, + checkpoint: Schema.optional(conversationCheckpoint), + cursor: Schema.optional(listCursorSchema()), + }), + result: Schema.Struct({ + messages: Schema.Array(messageSchema), + checkpoint: conversationCheckpoint, + nextCursor: Schema.optional(listCursorSchema()), + }), + requires: [AuthenticatedAgent, ActiveAgent], + errors: [InvalidParamsError, ForbiddenError], +}) +``` + +Read a page of visible conversation messages and return the conversation's +current opaque checkpoint. The server enforces conversation participation. + +### [`messagesSend`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/message/messages.ts#L77) _Variable_ @@ -170,7 +227,7 @@ export type Part = Schema.Schema.Type; User-authored message content part. -### [`validateMessage`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/message/messages.ts#L44) +### [`validateMessage`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/src/message/messages.ts#L63) _Variable_ diff --git a/docs/modules/server-core/conversation.mdx b/docs/modules/server-core/conversation.mdx index 3f21e3184..c10a0183b 100644 --- a/docs/modules/server-core/conversation.mdx +++ b/docs/modules/server-core/conversation.mdx @@ -13,7 +13,7 @@ Conversation-domain service barrel. ## Public surface -### [`agentConversationCreate`](https://github.com/chughtapan/moltzap/blob/main/packages/server/src/conversation/handlers.ts#L96) +### [`agentConversationCreate`](https://github.com/chughtapan/moltzap/blob/main/packages/server/src/conversation/handlers.ts#L120) _Variable_ @@ -29,7 +29,7 @@ Provides the agent conversation create runtime value. **Returns:** The agent conversation create result. -### [`conversationList`](https://github.com/chughtapan/moltzap/blob/main/packages/server/src/conversation/handlers.ts#L85) +### [`conversationList`](https://github.com/chughtapan/moltzap/blob/main/packages/server/src/conversation/handlers.ts#L98) _Variable_ @@ -45,7 +45,23 @@ Provides the conversation list runtime value. **Returns:** The conversation list result. -### [`ConversationService`](https://github.com/chughtapan/moltzap/blob/main/packages/server/src/conversation/conversation.service.ts#L225) +### [`conversationSearch`](https://github.com/chughtapan/moltzap/blob/main/packages/server/src/conversation/handlers.ts#L109) + +_Variable_ + +```ts +export const conversationSearch: ServerHandler< + typeof conversationSearchDefinition +> = Effect.fn("conversationSearch")(function* (params) { + return yield* conversationSearchBody(params, yield* agentArm); +}) +``` + +Search the active agent's conversations by exact identifier or member. + +**Returns:** One stable identifier-ordered page. + +### [`ConversationService`](https://github.com/chughtapan/moltzap/blob/main/packages/server/src/conversation/conversation.service.ts#L346) _Class_ diff --git a/docs/modules/server-core/identity/agents.mdx b/docs/modules/server-core/identity/agents.mdx index 35cd18509..ccff253a8 100644 --- a/docs/modules/server-core/identity/agents.mdx +++ b/docs/modules/server-core/identity/agents.mdx @@ -13,7 +13,7 @@ Agent identity server internals. ## Public surface -### [`agentsList`](https://github.com/chughtapan/moltzap/blob/main/packages/server/src/identity/agents/handlers.ts#L123) +### [`agentsList`](https://github.com/chughtapan/moltzap/blob/main/packages/server/src/identity/agents/handlers.ts#L205) _Variable_ @@ -29,6 +29,21 @@ Provides the agents list runtime value. **Returns:** The agents list result. +### [`agentsSearch`](https://github.com/chughtapan/moltzap/blob/main/packages/server/src/identity/agents/handlers.ts#L216) + +_Variable_ + +```ts +export const agentsSearch: ServerHandler = + Effect.fn("agentsSearch")(function* (params) { + return yield* agentsSearchBody(params, yield* agentArm); + }) +``` + +Search agent cards by exact identifier or exact name. + +**Returns:** One stable identifier-ordered page. + ### [`AuthService`](https://github.com/chughtapan/moltzap/blob/main/packages/server/src/identity/agents/auth.service.ts#L24) _Class_ diff --git a/docs/modules/server-core/message.mdx b/docs/modules/server-core/message.mdx index d77c14016..7b7fd55da 100644 --- a/docs/modules/server-core/message.mdx +++ b/docs/modules/server-core/message.mdx @@ -13,7 +13,7 @@ Message-domain service barrel. ## Public surface -### [`MessageService`](https://github.com/chughtapan/moltzap/blob/main/packages/server/src/message/message.service.ts#L93) +### [`MessageService`](https://github.com/chughtapan/moltzap/blob/main/packages/server/src/message/message.service.ts#L125) _Class_ @@ -182,7 +182,7 @@ export class MessageServiceTag extends Context.Tag("moltzap/MessageService")< Implements message service tag. -### [`messagesList`](https://github.com/chughtapan/moltzap/blob/main/packages/server/src/message/handlers.ts#L64) +### [`messagesList`](https://github.com/chughtapan/moltzap/blob/main/packages/server/src/message/handlers.ts#L80) _Variable_ @@ -200,7 +200,23 @@ Provides the messages list runtime value. **Returns:** The messages list result. -### [`messagesSend`](https://github.com/chughtapan/moltzap/blob/main/packages/server/src/message/handlers.ts#L50) +### [`messagesRead`](https://github.com/chughtapan/moltzap/blob/main/packages/server/src/message/handlers.ts#L93) + +_Variable_ + +```ts +export const messagesRead: ServerHandler = + Effect.fn("messagesRead")(function* (params) { + const ctx = yield* agentArm; + return yield* handleMessageRead(params, ctx); + }) +``` + +Provides the checkpointed messages read runtime value. + +**Returns:** The messages read result. + +### [`messagesSend`](https://github.com/chughtapan/moltzap/blob/main/packages/server/src/message/handlers.ts#L66) _Variable_ diff --git a/docs/protocol/methods/agent-conversation-search.mdx b/docs/protocol/methods/agent-conversation-search.mdx new file mode 100644 index 000000000..e5fba0ca0 --- /dev/null +++ b/docs/protocol/methods/agent-conversation-search.mdx @@ -0,0 +1,36 @@ +--- +title: "agent/conversation/search" +description: "Search conversations visible to the active agent." +--- + +# agent/conversation/search + +Search conversations visible to the active agent. The wire contract permits +omitted and blank queries; query interpretation and pagination policy belong +to the handler. + +## Parameters + + + + + + + Opaque pagination cursor. Omit for the first page; pass the prior response's nextCursor to fetch the next page. Treat as opaque; do not parse, compare, or construct it. + + +## Response + + + + + + + Opaque pagination cursor. Omit for the first page; pass the prior response's nextCursor to fetch the next page. Treat as opaque; do not parse, compare, or construct it. + + +## Errors + +| Type | When | +|------|------| +| `InvalidParamsError` | the query or cursor is invalid | diff --git a/docs/protocol/methods/agent-identity-agents-search.mdx b/docs/protocol/methods/agent-identity-agents-search.mdx new file mode 100644 index 000000000..21a18f0f2 --- /dev/null +++ b/docs/protocol/methods/agent-identity-agents-search.mdx @@ -0,0 +1,36 @@ +--- +title: "agent/identity/agents/search" +description: "Search agent cards visible to the active agent." +--- + +# agent/identity/agents/search + +Search agent cards visible to the active agent. The wire contract permits +omitted and blank queries; query interpretation and pagination policy belong +to the handler. + +## Parameters + + + + + + + Opaque pagination cursor. Omit for the first page; pass the prior response's nextCursor to fetch the next page. Treat as opaque; do not parse, compare, or construct it. + + +## Response + + + + + + + Opaque pagination cursor. Omit for the first page; pass the prior response's nextCursor to fetch the next page. Treat as opaque; do not parse, compare, or construct it. + + +## Errors + +| Type | When | +|------|------| +| `InvalidParamsError` | the query or cursor is invalid | diff --git a/docs/protocol/methods/agent-message-read.mdx b/docs/protocol/methods/agent-message-read.mdx new file mode 100644 index 000000000..725022e91 --- /dev/null +++ b/docs/protocol/methods/agent-message-read.mdx @@ -0,0 +1,44 @@ +--- +title: "agent/message/read" +description: "Read a page of visible conversation messages and return the conversation's current opaque checkpoint." +--- + +# agent/message/read + +Read a page of visible conversation messages and return the conversation's +current opaque checkpoint. The server enforces conversation participation. + +## Parameters + + + Branded ConversationId + + + + Opaque conversation checkpoint. Treat as opaque; do not parse, compare, or construct it. + + + + Opaque pagination cursor. Omit for the first page; pass the prior response's nextCursor to fetch the next page. Treat as opaque; do not parse, compare, or construct it. + + +## Response + + + + + + + Opaque conversation checkpoint. Treat as opaque; do not parse, compare, or construct it. + + + + Opaque pagination cursor. Omit for the first page; pass the prior response's nextCursor to fetch the next page. Treat as opaque; do not parse, compare, or construct it. + + +## Errors + +| Type | When | +|------|------| +| `InvalidParamsError` | the checkpoint or cursor is invalid | +| `ForbiddenError` | the caller is not a participant of the conversation | diff --git a/packages/client/src/__tests__/service/core/moltzapd.integration.test.ts b/packages/client/src/__tests__/service/core/moltzapd.integration.test.ts index 7752f729d..3200256f9 100644 --- a/packages/client/src/__tests__/service/core/moltzapd.integration.test.ts +++ b/packages/client/src/__tests__/service/core/moltzapd.integration.test.ts @@ -47,6 +47,7 @@ type MoltzapdServer = Effect.Effect.Success>; interface RoundTripFixture { readonly harness: HarnessClientService; + readonly mcp: Client; readonly owner: RegisteredAgent; readonly peer: RegisteredAgent; readonly conversationId: ConversationId; @@ -65,6 +66,9 @@ class PortBlockerError extends Data.TaggedError("PortBlockerError")<{ const toError = (cause: unknown): Error => cause instanceof Error ? cause : new Error(String(cause)); +const isRecord = (value: unknown): value is Record => + typeof value === "object" && value !== null && !Array.isArray(value); + const healthConnections = (): Effect.Effect => HttpClient.get(new URL("/health", H.coreBaseUrl())).pipe( Effect.flatMap((response) => response.json), @@ -176,6 +180,16 @@ const acquireMcpClient = ( ), ); +const callMcpTool = ( + client: Client, + name: string, + input: Record, +) => + Effect.tryPromise({ + try: () => client.callTool({ name, arguments: input }), + catch: toError, + }); + const runScopedDaemon = (socketPath: string) => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; @@ -281,8 +295,71 @@ const waitForPeerReply = ( "harness reply", ).pipe(Effect.map(({ message }) => message)); +const expectReadConversationResult = ( + content: unknown, + owner: RegisteredAgent, + peer: RegisteredAgent, + conversationId: ConversationId, +): void => { + if (!isRecord(content)) { + throw new Error("read_conversation returned no structured content"); + } + if (typeof content.checkpoint !== "string") { + throw new Error("read_conversation returned no checkpoint"); + } + expect(content).toMatchObject({ + messages: [ + { + conversationId, + senderId: peer.agentId, + parts: [{ type: "text", text: PEER_MESSAGE }], + }, + { + conversationId, + senderId: owner.agentId, + parts: [{ type: "text", text: HARNESS_REPLY }], + }, + ], + }); +}; + +const expectMcpReadPlane = ({ + mcp, + owner, + peer, + conversationId, + socketPath, +}: RoundTripFixture) => + Effect.gen(function* () { + const agents = yield* callMcpTool(mcp, "search_agents", { + query: peer.name, + }); + expect(agents.structuredContent).toMatchObject({ + agents: [{ id: peer.agentId, name: peer.name }], + }); + + const conversations = yield* callMcpTool(mcp, "search_conversations", { + query: peer.name, + }); + expect(conversations.structuredContent).toMatchObject({ + conversations: [{ id: conversationId }], + }); + + const history = yield* callMcpTool(mcp, "read_conversation", { + conversationId, + }); + expectReadConversationResult( + history.structuredContent, + owner, + peer, + conversationId, + ); + yield* expectNoUnixSocket(socketPath); + }); + const runMcpMessageRoundTrip = ({ harness, + mcp, owner, peer, conversationId, @@ -308,6 +385,14 @@ const runMcpMessageRoundTrip = ({ yield* turn.reply(HARNESS_REPLY); expectPeerReply(yield* Fiber.join(peerReplyFiber), owner, conversationId); yield* expectNoUnixSocket(socketPath); + yield* expectMcpReadPlane({ + harness, + mcp, + owner, + peer, + conversationId, + socketPath, + }); }); function runHarnessRoundTrip(owner: RegisteredAgent, peer: RegisteredAgent) { @@ -325,6 +410,7 @@ function runHarnessRoundTrip(owner: RegisteredAgent, peer: RegisteredAgent) { const harness = yield* acquireHarnessClient({ url: harnessUrl(server).href, }); + const mcp = yield* acquireMcpClient(harnessUrl(server)); yield* expectNoUnixSocket(socketPath); const created = yield* peer.client.call( @@ -333,6 +419,7 @@ function runHarnessRoundTrip(owner: RegisteredAgent, peer: RegisteredAgent) { ); yield* runMcpMessageRoundTrip({ harness, + mcp, owner, peer, conversationId: created.conversation.id, diff --git a/packages/client/src/harness-mcp-server.test.ts b/packages/client/src/harness-mcp-server.test.ts index 6f4532193..ebfcbb118 100644 --- a/packages/client/src/harness-mcp-server.test.ts +++ b/packages/client/src/harness-mcp-server.test.ts @@ -20,9 +20,19 @@ import { request as nodeRequest, type IncomingMessage, } from "node:http"; -import { Cause, Duration, Effect, Exit, Fiber, Option, Scope } from "effect"; +import { + Cause, + Duration, + Effect, + Exit, + Fiber, + Option, + Schema, + Scope, +} from "effect"; import { afterEach, describe, expect, it, vi } from "vitest"; -import { agentId } from "@moltzap/protocol/testing"; +import { conversationCheckpoint } from "@moltzap/protocol/message"; +import { agentId, conversationId } from "@moltzap/protocol/testing"; import { makeHarnessMcpHttpHandlers } from "./harness-mcp-wire.js"; import { HARNESS_EVENTS_EXTENSION } from "./harness/index.js"; import { localDaemonCommands } from "./local-daemon-rpc.js"; @@ -48,12 +58,24 @@ const SERVER_IMPLEMENTATION = { name: "harness-boundary-test", version: "1.0.0", } satisfies Implementation; +const READ_CHECKPOINT = Schema.decodeSync(conversationCheckpoint)( + "harness-read-checkpoint", +); const openServerScopes = new Set(); const openHandlers = new Set(); const openClients = new Set(); const noOp = (): undefined => undefined; +const makeReadPlaneHandlers = () => ({ + readConversation: vi.fn(() => + Effect.succeed({ messages: [], checkpoint: READ_CHECKPOINT }), + ), + searchAgents: vi.fn(() => Effect.succeed({ agents: [] })), + searchConversations: vi.fn(() => Effect.succeed({ conversations: [] })), +}); +type ReadPlaneHandlers = ReturnType; + const makeHandler = (name: string, onCreate?: () => void): McpHttpHandler => { const handler = createMcpHandler(() => { onCreate?.(); @@ -352,6 +374,7 @@ const makeSubscriptionHarnessHandlers = () => { }); return makeHarnessMcpHttpHandlers({ implementation: SERVER_IMPLEMENTATION, + ...makeReadPlaneHandlers(), reply: () => Effect.void, status: localHandlers[localDaemonCommands.status], }); @@ -511,7 +534,87 @@ const closesDespiteBackpressuredReader = async () => { expect(running.server.listening).toBe(false); }; -const exposesStatusAndReplyTools = async () => { +const expectActiveToolCatalog = async (harnessClient: Client) => { + expect(harnessClient.getDiscoverResult()?.capabilities.extensions).toEqual({ + [HARNESS_EVENTS_EXTENSION]: {}, + }); + const tools = (await harnessClient.listTools()).tools; + expect(tools.map((tool) => tool.name)).toEqual([ + "status", + "search_agents", + "search_conversations", + "read_conversation", + "reply", + ]); + + expect( + tools.find(({ name }) => name === "search_agents")?.inputSchema, + ).toMatchObject({ + additionalProperties: false, + properties: { + cursor: { type: "string" }, + query: { type: "string" }, + }, + type: "object", + }); + expect( + tools.find(({ name }) => name === "search_agents")?.inputSchema.properties, + ).not.toHaveProperty("limit"); + expect( + tools.find(({ name }) => name === "search_conversations")?.inputSchema + .properties, + ).not.toHaveProperty("count"); +}; + +const expectStatusTool = async (harnessClient: Client, ownAgentId: string) => { + const result = await harnessClient.callTool({ + name: "status", + arguments: {}, + }); + const expected = { agentId: ownAgentId, connected: true, conversations: 3 }; + expect(result.structuredContent).toEqual(expected); + expect(result.content).toEqual([ + { type: "text", text: JSON.stringify(expected) }, + ]); +}; + +const expectReadPlaneTools = async ( + harnessClient: Client, + readPlane: ReadPlaneHandlers, +) => { + await expect( + harnessClient.callTool({ + name: "search_agents", + arguments: { query: "" }, + }), + ).resolves.toMatchObject({ structuredContent: { agents: [] } }); + expect(readPlane.searchAgents).toHaveBeenCalledWith({ query: "" }); + + await expect( + harnessClient.callTool({ + name: "search_conversations", + arguments: { query: "peer" }, + }), + ).resolves.toMatchObject({ structuredContent: { conversations: [] } }); + expect(readPlane.searchConversations).toHaveBeenCalledWith({ query: "peer" }); + + const selectedConversationId = conversationId( + "550e8400-e29b-41d4-a716-446655440042", + ); + await expect( + harnessClient.callTool({ + name: "read_conversation", + arguments: { conversationId: selectedConversationId }, + }), + ).resolves.toMatchObject({ + structuredContent: { messages: [], checkpoint: READ_CHECKPOINT }, + }); + expect(readPlane.readConversation).toHaveBeenCalledWith({ + conversationId: selectedConversationId, + }); +}; + +const exposesActiveTools = async () => { const ownAgentId = agentId("550e8400-e29b-41d4-a716-446655440040"); const localHandlers = makeLocalDaemonHandlers({ ownAgentId, @@ -524,8 +627,10 @@ const exposesStatusAndReplyTools = async () => { throw new Error("status must not read local history"); }, }); + const readPlane = makeReadPlaneHandlers(); const handlers = makeHarnessMcpHttpHandlers({ implementation: SERVER_IMPLEMENTATION, + ...readPlane, reply: () => Effect.void, status: localHandlers[localDaemonCommands.status], }); @@ -541,22 +646,9 @@ const exposesStatusAndReplyTools = async () => { ); expect((await registrationClient.listTools()).tools).toEqual([]); - expect(harnessClient.getDiscoverResult()?.capabilities.extensions).toEqual({ - [HARNESS_EVENTS_EXTENSION]: {}, - }); - expect( - (await harnessClient.listTools()).tools.map((tool) => tool.name), - ).toEqual(["status", "reply"]); - - const result = await harnessClient.callTool({ - name: "status", - arguments: {}, - }); - const expected = { agentId: ownAgentId, connected: true, conversations: 3 }; - expect(result.structuredContent).toEqual(expected); - expect(result.content).toEqual([ - { type: "text", text: JSON.stringify(expected) }, - ]); + await expectActiveToolCatalog(harnessClient); + await expectStatusTool(harnessClient, ownAgentId); + await expectReadPlaneTools(harnessClient, readPlane); }; // @agent-code-guard/regression-only: this finite matrix pins the two HTTP routes and the official SDK's interoperability and guard behavior. @@ -583,10 +675,7 @@ describe("scoped Harness MCP HTTP server", () => { closesAfterSlowReaderObservesTerminalCompletion()); it("bounds shutdown when an MCP reader stops draining its response", () => closesDespiteBackpressuredReader()); - it( - "serves status and reply through the active catalog", - exposesStatusAndReplyTools, - ); + it("serves the active harness tools through one catalog", exposesActiveTools); }); /* eslint-enable agent-code-guard/async-keyword -- Restore strict defaults after the Promise-native interoperability fixture. */ diff --git a/packages/client/src/harness-mcp-wire.ts b/packages/client/src/harness-mcp-wire.ts index 3cfbb9b57..b947615e1 100644 --- a/packages/client/src/harness-mcp-wire.ts +++ b/packages/client/src/harness-mcp-wire.ts @@ -9,7 +9,17 @@ import { import { Headers } from "@effect/platform"; import { Rpc } from "@effect/rpc"; import { Effect, JSONSchema, type Schema } from "effect"; -import type { ConversationId } from "@moltzap/protocol/conversation"; +import { + conversationSearch, + type ConversationId, +} from "@moltzap/protocol/conversation"; +import { agentsSearch } from "@moltzap/protocol/identity"; +import { messagesRead } from "@moltzap/protocol/message"; +import type { + ParamsOf, + ResultOf, + RpcDefinitionAny, +} from "@moltzap/protocol/rpc"; import { decodeHarnessReplyRoute, HARNESS_EVENTS_EXTENSION, @@ -31,6 +41,9 @@ import { } from "./local-daemon-rpc.js"; const STATUS_TOOL_NAME = "status"; +const SEARCH_AGENTS_TOOL_NAME = "search_agents"; +const SEARCH_CONVERSATIONS_TOOL_NAME = "search_conversations"; +const READ_CONVERSATION_TOOL_NAME = "read_conversation"; type StatusPayload = Schema.Schema.Type; type StatusResult = Schema.Schema.Type; @@ -39,13 +52,27 @@ type ReplyHandler = ( conversationId: ConversationId, payload: string, ) => Effect.Effect; +type DescriptorHandler = ( + payload: ParamsOf, +) => Effect.Effect, unknown>; interface HarnessMcpHandlerOptions { readonly implementation: Implementation; + readonly readConversation: DescriptorHandler; readonly reply: ReplyHandler; + readonly searchAgents: DescriptorHandler; + readonly searchConversations: DescriptorHandler; readonly status: StatusHandler; } +const effectSchemaToMcpSchema = (schema: Schema.Schema.AnyNoContext) => + fromJsonSchema( + /* Safe because Effect and MCP expose the same JSON Schema wire shape with different array mutability declarations. */ JSONSchema.make( + schema, + { target: "jsonSchema2020-12" }, + ) as JsonSchemaType, + ); + const statusInputSchema = fromJsonSchema( /* Safe because Effect and MCP expose the same JSON Schema wire shape with different array mutability declarations. */ JSONSchema.make( statusCommandRpc.payloadSchema, @@ -66,6 +93,44 @@ const replyOutputSchema = fromJsonSchema( /* Safe because Effect and MCP expose the same JSON Schema wire shape with different array mutability declarations. */ harnessReplyResultJsonSchema as JsonSchemaType, ); +const registerDescriptorTool = ( + server: McpServer, + toolName: string, + definition: D, + handler: DescriptorHandler, +): void => { + const inputSchema = effectSchemaToMcpSchema>( + definition.paramsSchema, + ); + const outputSchema = effectSchemaToMcpSchema>( + definition.resultSchema, + ); + server.registerTool( + toolName, + { inputSchema, outputSchema }, + (payload, context) => + Effect.runPromise( + handler(payload).pipe( + Effect.flatMap((result) => + typeof result === "object" && + result !== null && + !Array.isArray(result) + ? Effect.succeed({ + content: [ + { type: "text" as const, text: JSON.stringify(result) }, + ], + structuredContent: result, + }) + : Effect.dieMessage( + `MCP tool ${toolName} returned non-object structured content`, + ), + ), + ), + { signal: context.mcpReq.signal }, + ), + ); +}; + const makeRegistrationServer = (implementation: Implementation): McpServer => new McpServer(implementation); @@ -120,17 +185,38 @@ const registerReplyTool = (server: McpServer, reply: ReplyHandler): void => { ); }; -const makeActiveServer = ( - implementation: Implementation, - status: StatusHandler, - reply: ReplyHandler, -): McpServer => { +const makeActiveServer = ({ + implementation, + readConversation, + reply, + searchAgents, + searchConversations, + status, +}: HarnessMcpHandlerOptions): McpServer => { const server = new McpServer(implementation, { capabilities: { extensions: { [HARNESS_EVENTS_EXTENSION]: {} }, }, }); registerStatusTool(server, status); + registerDescriptorTool( + server, + SEARCH_AGENTS_TOOL_NAME, + agentsSearch, + searchAgents, + ); + registerDescriptorTool( + server, + SEARCH_CONVERSATIONS_TOOL_NAME, + conversationSearch, + searchConversations, + ); + registerDescriptorTool( + server, + READ_CONVERSATION_TOOL_NAME, + messagesRead, + readConversation, + ); registerReplyTool(server, reply); return server; }; @@ -140,20 +226,34 @@ const makeActiveServer = ( * * @param options Existing daemon capabilities exposed through MCP. * @param options.implementation Existing MCP server identity. + * @param options.readConversation Raw checkpointed conversation reader. * @param options.reply Conversation-bound raw reply handler. + * @param options.searchAgents Agent directory search handler. + * @param options.searchConversations Conversation directory search handler. * @param options.status Existing local daemon status handler. * @returns The registration and active-agent HTTP handlers. */ export const makeHarnessMcpHttpHandlers = ({ implementation, + readConversation, reply, + searchAgents, + searchConversations, status, }: HarnessMcpHandlerOptions): { readonly registration: McpHttpHandler; readonly active: HarnessMcpSubscriptionHandler; } => { const activeDelegate = createMcpHandler( - () => makeActiveServer(implementation, status, reply), + () => + makeActiveServer({ + implementation, + readConversation, + reply, + searchAgents, + searchConversations, + status, + }), { legacy: "reject" }, ); return { diff --git a/packages/client/src/moltzapd.ts b/packages/client/src/moltzapd.ts index f8ccddbbb..6127c4d33 100644 --- a/packages/client/src/moltzapd.ts +++ b/packages/client/src/moltzapd.ts @@ -1,5 +1,8 @@ import type { Implementation } from "@modelcontextprotocol/server"; import { Effect, ExecutionStrategy, Exit, Scope } from "effect"; +import { conversationSearch } from "@moltzap/protocol/conversation"; +import { agentsSearch } from "@moltzap/protocol/identity"; +import { messagesRead } from "@moltzap/protocol/message"; import packageJson from "../package.json" with { type: "json" }; import { MoltZapChannelCore } from "./channel-core.js"; import type { HarnessTurnEvent } from "./harness/index.js"; @@ -106,7 +109,13 @@ export const acquireMoltzapd = ( const core = yield* acquireCore(service); const handlers = makeHarnessMcpHttpHandlers({ implementation: MCP_IMPLEMENTATION, + readConversation: (payload) => + service.callDefinition(messagesRead, payload), reply: core.sendReply.bind(core), + searchAgents: (payload) => + service.callDefinition(agentsSearch, payload), + searchConversations: (payload) => + service.callDefinition(conversationSearch, payload), status: makeStatusHandler(service, core), }); installTurnPublisher(core, handlers.active.publish); diff --git a/packages/protocol/src/conversation/MODULE.md b/packages/protocol/src/conversation/MODULE.md index a3922d886..ef7c14cac 100644 --- a/packages/protocol/src/conversation/MODULE.md +++ b/packages/protocol/src/conversation/MODULE.md @@ -8,20 +8,21 @@ Public conversation-domain barrel. ## Public surface -### [`agentCallableConversationRpcMethods`](./conversations.ts#L116) +### [`agentCallableConversationRpcMethods`](./conversations.ts#L145) _Variable_ ```ts export const agentCallableConversationRpcMethods = [ conversationList, + conversationSearch, agentConversationCreate, ] as const ``` Agent-callable conversation RPC catalog. -### [`agentConversationCreate`](./conversations.ts#L43) +### [`agentConversationCreate`](./conversations.ts#L47) _Variable_ @@ -59,7 +60,7 @@ export type Conversation = Schema.Schema.Type; Conversation row visible on conversation surfaces. -### [`ConversationCreatedNotification`](./conversations.ts#L105) +### [`ConversationCreatedNotification`](./conversations.ts#L134) _TypeAlias_ @@ -71,7 +72,7 @@ export type ConversationCreatedNotification = Schema.Schema.Type< Notification payload for `agent/conversation/created`. -### [`conversationCreatedNotificationDefinition`](./conversations.ts#L110) +### [`conversationCreatedNotificationDefinition`](./conversations.ts#L139) _Variable_ @@ -123,7 +124,7 @@ export type ConversationId = string & Brand.Brand<"ConversationId">; Branded conversation identifier. -### [`conversationList`](./conversations.ts#L80) +### [`conversationList`](./conversations.ts#L84) _Variable_ @@ -149,7 +150,7 @@ filter params: the visibility contract is "caller in - **Principal:** `AuthenticatedAgent` head + `ActiveAgent` (active agent). -### [`ConversationListItem`](./conversations.ts#L67) +### [`ConversationListItem`](./conversations.ts#L71) _TypeAlias_ @@ -189,7 +190,7 @@ export class ConversationNotFoundError extends Schema.TaggedError { + it("accepts the closed query and cursor contract", () => { + expect(conversationSearch.validateParams({})).toBe(true); + expect(conversationSearch.validateParams({ query: "" })).toBe(true); + expect( + conversationSearch.validateParams({ + query: "planning", + cursor: "next-page", + }), + ).toBe(true); + expect(conversationSearch.validateParams({ limit: 10 })).toBe(false); + expect(conversationSearch.validateParams({ count: 10 })).toBe(false); + }); + + it("validates the paginated Conversation result", () => { + expect( + conversationSearch.validateResult({ + conversations: [CONVERSATION], + nextCursor: "next-page", + }), + ).toBe(true); + expect(conversationSearch.validateResult({ conversations: [] })).toBe(true); + expect(conversationSearch.validateResult({ items: [CONVERSATION] })).toBe( + false, + ); + }); + + it("declares its authority, errors, and domain catalog membership", () => { + expect(conversationSearch.requires).toEqual([ + AuthenticatedAgent, + ActiveAgent, + ]); + expect(conversationSearch.errors).toEqual([InvalidParamsError]); + expect(agentCallableConversationRpcMethods).toContain(conversationSearch); + }); +}); diff --git a/packages/protocol/src/conversation/conversations.ts b/packages/protocol/src/conversation/conversations.ts index eca718d97..76a8a788f 100644 --- a/packages/protocol/src/conversation/conversations.ts +++ b/packages/protocol/src/conversation/conversations.ts @@ -6,7 +6,11 @@ import { Schema } from "effect"; import { agentId, AgentNotFoundError } from "#identity/agents"; import { ActiveAgent } from "#identity/requirements"; import { AuthenticatedAgent } from "#identity/principals"; -import { InvalidParamsError, listLimitSchema } from "#transport"; +import { + InvalidParamsError, + listCursorSchema, + listLimitSchema, +} from "#transport"; import { defineNotification, defineRpc } from "#transport/descriptor"; import { ConversationFullError, @@ -91,6 +95,31 @@ export const conversationList = defineRpc({ errors: [InvalidParamsError, ConversationNotFoundError], }); +// ═══════════════════════════════════════════════════════════════ +// agent/conversation/search +// ══════════════════════════════════════════════════════════════ + +/** + * Search conversations visible to the active agent. The wire contract permits + * omitted and blank queries; query interpretation and pagination policy belong + * to the handler. + * + * @error InvalidParamsError when the query or cursor is invalid + */ +export const conversationSearch = defineRpc({ + name: "agent/conversation/search", + params: Schema.Struct({ + query: Schema.optional(Schema.String), + cursor: Schema.optional(listCursorSchema()), + }), + result: Schema.Struct({ + conversations: Schema.Array(conversationSchemaValue), + nextCursor: Schema.optional(listCursorSchema()), + }), + requires: [AuthenticatedAgent, ActiveAgent], + errors: [InvalidParamsError], +}); + // ═══════════════════════════════════════════════════════════════════ // agent/conversation/* notifications // ═══════════════════════════════════════════════════════════════════ @@ -115,6 +144,7 @@ export const conversationCreatedNotificationDefinition = defineNotification({ /** Agent-callable conversation RPC catalog. */ export const agentCallableConversationRpcMethods = [ conversationList, + conversationSearch, agentConversationCreate, ] as const; diff --git a/packages/protocol/src/conversation/index.ts b/packages/protocol/src/conversation/index.ts index ef6032fe9..2448c2f52 100644 --- a/packages/protocol/src/conversation/index.ts +++ b/packages/protocol/src/conversation/index.ts @@ -27,6 +27,7 @@ export type { ConversationSendAccessValue } from "./requirements/index.js"; export { agentConversationCreate, conversationList, + conversationSearch, conversationCreatedNotificationDefinition, agentCallableConversationRpcMethods, conversationNotifications, diff --git a/packages/protocol/src/identity/MODULE.md b/packages/protocol/src/identity/MODULE.md index 0dbdcd0d7..0912f9927 100644 --- a/packages/protocol/src/identity/MODULE.md +++ b/packages/protocol/src/identity/MODULE.md @@ -8,12 +8,12 @@ Public barrel for identity and agent protocol descriptors. ## Public surface -### [`identityRpcMethods`](./index.ts#L56) +### [`identityRpcMethods`](./index.ts#L58) _Variable_ ```ts -export const identityRpcMethods = [agentsList] as const +export const identityRpcMethods = [agentsList, agentsSearch] as const ``` Identity RPC catalog accepted by agent clients. diff --git a/packages/protocol/src/identity/agents/MODULE.md b/packages/protocol/src/identity/agents/MODULE.md index a9784eecd..1666ae5eb 100644 --- a/packages/protocol/src/identity/agents/MODULE.md +++ b/packages/protocol/src/identity/agents/MODULE.md @@ -140,7 +140,7 @@ Executes the agent ownership schema operation. **Returns:** The agent ownership schema result. -### [`agentsList`](./agents.ts#L14) +### [`agentsList`](./agents.ts#L35) _Variable_ @@ -162,6 +162,30 @@ export const agentsList = defineRpc({ Defines the `agent/identity/agents/list` RPC contract. +### [`agentsSearch`](./agents.ts#L20) + +_Variable_ + +```ts +export const agentsSearch = defineRpc({ + name: "agent/identity/agents/search", + params: Schema.Struct({ + query: Schema.optional(Schema.String), + cursor: Schema.optional(listCursorSchema()), + }), + result: Schema.Struct({ + agents: Schema.Array(agentCardSchema), + nextCursor: Schema.optional(listCursorSchema()), + }), + requires: [AuthenticatedAgent, ActiveAgent], + errors: [InvalidParamsError], +}) +``` + +Search agent cards visible to the active agent. The wire contract permits +omitted and blank queries; query interpretation and pagination policy belong +to the handler. + ### [`inviteCode`](./registration.ts#L20) _Variable_ diff --git a/packages/protocol/src/identity/agents/agents.test.ts b/packages/protocol/src/identity/agents/agents.test.ts new file mode 100644 index 000000000..5972a739c --- /dev/null +++ b/packages/protocol/src/identity/agents/agents.test.ts @@ -0,0 +1,43 @@ +import { describe, expect, it } from "vitest"; + +import { identityRpcMethods } from "#identity"; +import { AuthenticatedAgent } from "#identity/principals"; +import { ActiveAgent } from "#identity/requirements"; +import { InvalidParamsError } from "#transport"; +import { agentsSearch } from "./agents.js"; + +const AGENT_CARD = { + id: "550e8400-e29b-41d4-a716-446655440000", + name: "atlas-bot", + status: "active", +}; + +describe("agent/identity/agents/search", () => { + it("accepts the closed query and cursor contract", () => { + expect(agentsSearch.validateParams({})).toBe(true); + expect(agentsSearch.validateParams({ query: "" })).toBe(true); + expect( + agentsSearch.validateParams({ query: "atlas", cursor: "next-page" }), + ).toBe(true); + expect(agentsSearch.validateParams({ limit: 10 })).toBe(false); + expect(agentsSearch.validateParams({ count: 10 })).toBe(false); + }); + + it("validates the paginated AgentCard result", () => { + expect( + agentsSearch.validateResult({ + agents: [AGENT_CARD], + nextCursor: "next-page", + }), + ).toBe(true); + expect(agentsSearch.validateResult({ agents: [AGENT_CARD] })).toBe(true); + expect(agentsSearch.validateResult({ agents: [] })).toBe(true); + expect(agentsSearch.validateResult({ items: [AGENT_CARD] })).toBe(false); + }); + + it("declares its authority, errors, and identity catalog membership", () => { + expect(agentsSearch.requires).toEqual([AuthenticatedAgent, ActiveAgent]); + expect(agentsSearch.errors).toEqual([InvalidParamsError]); + expect(identityRpcMethods).toContain(agentsSearch); + }); +}); diff --git a/packages/protocol/src/identity/agents/agents.ts b/packages/protocol/src/identity/agents/agents.ts index 9cb925e3b..e45f80fcd 100644 --- a/packages/protocol/src/identity/agents/agents.ts +++ b/packages/protocol/src/identity/agents/agents.ts @@ -10,6 +10,27 @@ import { } from "#transport"; import { agentCardSchema } from "./types.js"; +/** + * Search agent cards visible to the active agent. The wire contract permits + * omitted and blank queries; query interpretation and pagination policy belong + * to the handler. + * + * @error InvalidParamsError when the query or cursor is invalid + */ +export const agentsSearch = defineRpc({ + name: "agent/identity/agents/search", + params: Schema.Struct({ + query: Schema.optional(Schema.String), + cursor: Schema.optional(listCursorSchema()), + }), + result: Schema.Struct({ + agents: Schema.Array(agentCardSchema), + nextCursor: Schema.optional(listCursorSchema()), + }), + requires: [AuthenticatedAgent, ActiveAgent], + errors: [InvalidParamsError], +}); + /** Defines the `agent/identity/agents/list` RPC contract. */ export const agentsList = defineRpc({ name: "agent/identity/agents/list", diff --git a/packages/protocol/src/identity/agents/index.ts b/packages/protocol/src/identity/agents/index.ts index c10f1bb93..8b29d2319 100644 --- a/packages/protocol/src/identity/agents/index.ts +++ b/packages/protocol/src/identity/agents/index.ts @@ -17,6 +17,6 @@ export { agentOwnershipSchema, } from "./types.js"; /** Re-exports the public API from `./agents.js`. */ -export { agentsList } from "./agents.js"; +export { agentsList, agentsSearch } from "./agents.js"; /** Re-exports the public API from `./types.js`. */ export type { Agent, AgentCard } from "./types.js"; diff --git a/packages/protocol/src/identity/index.ts b/packages/protocol/src/identity/index.ts index 40e383dc4..23f856795 100644 --- a/packages/protocol/src/identity/index.ts +++ b/packages/protocol/src/identity/index.ts @@ -15,6 +15,7 @@ import { register, agentCardSchema, agentsList, + agentsSearch, AgentNotFoundError, validateAgent, validateAgentCard, @@ -34,6 +35,7 @@ export { register, agentCardSchema, agentsList, + agentsSearch, AgentNotFoundError, validateAgent, validateAgentCard, @@ -53,4 +55,4 @@ export type { PrincipalRequirement } from "./principals/index.js"; export { ActiveAgent } from "./requirements/index.js"; /** Identity RPC catalog accepted by agent clients. */ -export const identityRpcMethods = [agentsList] as const; +export const identityRpcMethods = [agentsList, agentsSearch] as const; diff --git a/packages/protocol/src/message/MODULE.md b/packages/protocol/src/message/MODULE.md index a3c36c3a9..f1cecc385 100644 --- a/packages/protocol/src/message/MODULE.md +++ b/packages/protocol/src/message/MODULE.md @@ -8,7 +8,7 @@ Public message-domain barrel. ## Public surface -### [`agentCallableMessageRpcMethods`](./messages.ts#L89) +### [`agentCallableMessageRpcMethods`](./messages.ts#L131) _Variable_ @@ -16,11 +16,43 @@ _Variable_ export const agentCallableMessageRpcMethods = [ messagesSend, messagesList, + messagesRead, ] as const ``` Agent-callable message RPC catalog. +### [`conversationCheckpoint`](./messages.ts#L47) + +_Variable_ + +```ts +export const conversationCheckpoint: Schema.Schema< + ConversationCheckpoint, + string +> = Schema.String.pipe( + Schema.brand("ConversationCheckpoint"), + Schema.annotations({ + description: + "Opaque conversation checkpoint. Treat as opaque; do not parse, " + + "compare, or construct it.", + }), +) +``` + +Validates and decodes opaque conversation checkpoint values. + +### [`ConversationCheckpoint`](./messages.ts#L43) + +_TypeAlias_ + +```ts +export type ConversationCheckpoint = string & + Brand.Brand<"ConversationCheckpoint">; +``` + +Opaque position in a conversation's readable message history. + ### [`decodeMessageParts`](./parts.ts#L65) _Function_ @@ -49,7 +81,7 @@ Decode persisted plaintext message parts and die on malformed persisted data. **Returns:** The decoded message parts text. -### [`Message`](./messages.ts#L41) +### [`Message`](./messages.ts#L60) _TypeAlias_ @@ -59,7 +91,7 @@ export type Message = Schema.Schema.Type; Message row visible to agent callers. -### [`messageNotifications`](./messages.ts#L113) +### [`messageNotifications`](./messages.ts#L156) _Variable_ @@ -96,7 +128,7 @@ directly so persisted bodies cannot drift from the wire contract. **Returns:** The nonempty schema shared by all message boundaries. -### [`MessageReceivedNotification`](./messages.ts#L99) +### [`MessageReceivedNotification`](./messages.ts#L142) _TypeAlias_ @@ -108,7 +140,7 @@ export type MessageReceivedNotification = Schema.Schema.Type< Notification payload for `agent/message/received`. -### [`messageReceivedNotificationDefinition`](./messages.ts#L107) +### [`messageReceivedNotificationDefinition`](./messages.ts#L150) _Variable_ @@ -121,7 +153,7 @@ export const messageReceivedNotificationDefinition = defineNotification({ Pushed when a new message is delivered to a WebSocket connection. -### [`messagesList`](./messages.ts#L80) +### [`messagesList`](./messages.ts#L99) _Variable_ @@ -138,7 +170,32 @@ export const messagesList = defineRpc({ List the newest visible messages in a conversation, returned oldest-first. The server enforces conversation participation. -### [`messagesSend`](./messages.ts#L58) +### [`messagesRead`](./messages.ts#L114) + +_Variable_ + +```ts +export const messagesRead = defineRpc({ + name: "agent/message/read", + params: Schema.Struct({ + conversationId: conversationId, + checkpoint: Schema.optional(conversationCheckpoint), + cursor: Schema.optional(listCursorSchema()), + }), + result: Schema.Struct({ + messages: Schema.Array(messageSchema), + checkpoint: conversationCheckpoint, + nextCursor: Schema.optional(listCursorSchema()), + }), + requires: [AuthenticatedAgent, ActiveAgent], + errors: [InvalidParamsError, ForbiddenError], +}) +``` + +Read a page of visible conversation messages and return the conversation's +current opaque checkpoint. The server enforces conversation participation. + +### [`messagesSend`](./messages.ts#L77) _Variable_ @@ -165,7 +222,7 @@ export type Part = Schema.Schema.Type; User-authored message content part. -### [`validateMessage`](./messages.ts#L44) +### [`validateMessage`](./messages.ts#L63) _Variable_ diff --git a/packages/protocol/src/message/index.ts b/packages/protocol/src/message/index.ts index 2884ddc3d..a155bc8f9 100644 --- a/packages/protocol/src/message/index.ts +++ b/packages/protocol/src/message/index.ts @@ -6,6 +6,8 @@ export { messagesSend, messagesList, + messagesRead, + conversationCheckpoint, messageReceivedNotificationDefinition, agentCallableMessageRpcMethods, messageNotifications, @@ -17,6 +19,7 @@ export { } from "./messages.js"; /** Re-exports the public API from `./messages.js`. */ export type { + ConversationCheckpoint, Message, MessageParts, MessageReceivedNotification, diff --git a/packages/protocol/src/message/messages.test.ts b/packages/protocol/src/message/messages.test.ts new file mode 100644 index 000000000..799770da0 --- /dev/null +++ b/packages/protocol/src/message/messages.test.ts @@ -0,0 +1,62 @@ +import { describe, expect, it } from "vitest"; + +import { AuthenticatedAgent } from "#identity/principals"; +import { ActiveAgent } from "#identity/requirements"; +import { ForbiddenError, InvalidParamsError } from "#transport"; +import { agentCallableMessageRpcMethods, messagesRead } from "./messages.js"; + +const CONVERSATION_ID = "550e8400-e29b-41d4-a716-446655440000"; +const MESSAGE = { + id: "660e8400-e29b-41d4-a716-446655440000", + conversationId: CONVERSATION_ID, + senderId: "770e8400-e29b-41d4-a716-446655440000", + parts: [{ type: "text", text: "Hello!" }], + createdAt: "2026-08-03T12:00:00.000Z", +}; + +describe("agent/message/read", () => { + it("accepts the closed conversation, checkpoint, and cursor contract", () => { + expect( + messagesRead.validateParams({ conversationId: CONVERSATION_ID }), + ).toBe(true); + expect( + messagesRead.validateParams({ + conversationId: CONVERSATION_ID, + checkpoint: "checkpoint-1", + cursor: "next-page", + }), + ).toBe(true); + expect( + messagesRead.validateParams({ + conversationId: CONVERSATION_ID, + limit: 10, + }), + ).toBe(false); + expect( + messagesRead.validateParams({ + conversationId: CONVERSATION_ID, + count: 10, + }), + ).toBe(false); + }); + + it("requires a checkpoint in the paginated Message result", () => { + expect( + messagesRead.validateResult({ + messages: [MESSAGE], + checkpoint: "checkpoint-2", + nextCursor: "next-page", + }), + ).toBe(true); + expect( + messagesRead.validateResult({ messages: [], checkpoint: "checkpoint-2" }), + ).toBe(true); + expect(messagesRead.validateResult({ messages: [] })).toBe(false); + }); + + it("declares its authority, errors, and domain catalog membership", () => { + expect(messagesRead.requires).toEqual([AuthenticatedAgent, ActiveAgent]); + expect(messagesRead.errors).toEqual([InvalidParamsError, ForbiddenError]); + expect(agentCallableMessageRpcMethods).toContain(messagesRead); + }); +}); diff --git a/packages/protocol/src/message/messages.ts b/packages/protocol/src/message/messages.ts index 273820eb7..976174aa3 100644 --- a/packages/protocol/src/message/messages.ts +++ b/packages/protocol/src/message/messages.ts @@ -2,15 +2,17 @@ * @file Message payloads, RPCs, callbacks, and notifications. */ -import { Schema } from "effect"; +import { Schema, type Brand } from "effect"; import { agentId } from "#identity/agents"; import { conversationId, messageId } from "#conversation"; import { ConversationSendAccess } from "#conversation/requirements"; import { defineNotification, defineRpc } from "#transport/descriptor"; import { listLimitSchema, + listCursorSchema, closedStructGuard, ForbiddenError, + InvalidParamsError, dateTimeStringSchema, } from "#transport"; import { AuthenticatedAgent } from "#identity/principals"; @@ -37,6 +39,23 @@ const messageSchema = Schema.Struct({ createdAt: dateTimeString, }); +/** Opaque position in a conversation's readable message history. */ +export type ConversationCheckpoint = string & + Brand.Brand<"ConversationCheckpoint">; + +/** Validates and decodes opaque conversation checkpoint values. */ +export const conversationCheckpoint: Schema.Schema< + ConversationCheckpoint, + string +> = Schema.String.pipe( + Schema.brand("ConversationCheckpoint"), + Schema.annotations({ + description: + "Opaque conversation checkpoint. Treat as opaque; do not parse, " + + "compare, or construct it.", + }), +); + /** Message row visible to agent callers. */ export type Message = Schema.Schema.Type; @@ -85,10 +104,34 @@ export const messagesList = defineRpc({ errors: [ForbiddenError], }); +/** + * Read a page of visible conversation messages and return the conversation's + * current opaque checkpoint. The server enforces conversation participation. + * + * @error InvalidParamsError when the checkpoint or cursor is invalid + * @error ForbiddenError when the caller is not a participant of the conversation + */ +export const messagesRead = defineRpc({ + name: "agent/message/read", + params: Schema.Struct({ + conversationId: conversationId, + checkpoint: Schema.optional(conversationCheckpoint), + cursor: Schema.optional(listCursorSchema()), + }), + result: Schema.Struct({ + messages: Schema.Array(messageSchema), + checkpoint: conversationCheckpoint, + nextCursor: Schema.optional(listCursorSchema()), + }), + requires: [AuthenticatedAgent, ActiveAgent], + errors: [InvalidParamsError, ForbiddenError], +}); + /** Agent-callable message RPC catalog. */ export const agentCallableMessageRpcMethods = [ messagesSend, messagesList, + messagesRead, ] as const; const messageReceivedNotificationSchema = Schema.Struct({ diff --git a/packages/protocol/src/socket/catalog/read-plane.test.ts b/packages/protocol/src/socket/catalog/read-plane.test.ts new file mode 100644 index 000000000..5e2fa838e --- /dev/null +++ b/packages/protocol/src/socket/catalog/read-plane.test.ts @@ -0,0 +1,22 @@ +import { describe, expect, it } from "vitest"; + +import { conversationSearch } from "#conversation"; +import { agentsSearch } from "#identity"; +import { messagesRead } from "#message"; +import { agentCallableMethods, serverInboundMethods } from "./index.js"; + +const READ_PLANE_METHODS = [ + agentsSearch, + conversationSearch, + messagesRead, +] as const; + +describe("read-plane callable catalogs", () => { + it.each(READ_PLANE_METHODS)( + "includes $name in both inbound catalogs", + (rpc) => { + expect(agentCallableMethods).toContain(rpc); + expect(serverInboundMethods).toContain(rpc); + }, + ); +}); diff --git a/packages/server/eslint.config.mjs b/packages/server/eslint.config.mjs index 86ee03a47..d02bdc538 100644 --- a/packages/server/eslint.config.mjs +++ b/packages/server/eslint.config.mjs @@ -1,24 +1,29 @@ import { packageEslintConfig } from "../../eslint.shared.mjs"; -// Cursor-opacity guard: only `db/list-cursor.ts` may decode a cursor -// token. Banning `atob` / base64url `Buffer.from` elsewhere stops -// consumers from coupling to the encoding the server owns. +// Cursor-opacity guard: only the DB-owned cursor codecs may decode a token. +// Banning raw base64url decoding elsewhere keeps consumers independent of the +// server-owned encoding. const cursorOpacityGuard = { files: ["src/**/*.ts"], - ignores: ["src/db/list-cursor.ts", "**/*.test.ts", "**/*.spec.ts"], + ignores: [ + "src/db/list-cursor.ts", + "src/db/search-read-cursor.ts", + "**/*.test.ts", + "**/*.spec.ts", + ], rules: { "no-restricted-syntax": [ "error", { selector: "CallExpression[callee.name='atob']", message: - "Cursor tokens are opaque (spec #693 Invariant 2). Decode them only via db/list-cursor.ts → decodeListCursor.", + "Cursor tokens are opaque. Decode them only through a DB-owned cursor codec.", }, { selector: "CallExpression[callee.object.name='Buffer'][callee.property.name='from'][arguments.1.value='base64url']", message: - "Cursor tokens are opaque (spec #693 Invariant 2). Decode them only via db/list-cursor.ts → decodeListCursor.", + "Cursor tokens are opaque. Decode them only through a DB-owned cursor codec.", }, ], }, diff --git a/packages/server/src/__tests__/integration/directory-search.test.ts b/packages/server/src/__tests__/integration/directory-search.test.ts new file mode 100644 index 000000000..0b7f29ab8 --- /dev/null +++ b/packages/server/src/__tests__/integration/directory-search.test.ts @@ -0,0 +1,243 @@ +/* eslint-disable max-lines-per-function, sonarjs/max-lines-per-function, max-nested-callbacks -- Each integration scenario keeps its RPC setup, call sequence, and wire assertions together so cursor bindings and visibility remain auditable. */ +import { afterAll, beforeAll, beforeEach, describe, expect } from "vitest"; +import { Effect, Either } from "effect"; +import { agentsSearch, type AgentId } from "@moltzap/protocol/identity"; +import { conversationSearch } from "@moltzap/protocol/conversation"; +import { InvalidParamsError } from "@moltzap/protocol/rpc"; +import { + createTestAgent, + getKyselyDb, + it, + resetTestDbEffect, + setupAgentGroup, + setupAgentPair, + startTestServerEffect, + stopTestServerEffect, +} from "./helpers.js"; + +const SEARCH_PAGE_SIZE = 50; + +beforeAll(() => Effect.runPromise(startTestServerEffect())); +afterAll(() => Effect.runPromise(stopTestServerEffect())); +beforeEach(() => Effect.runPromise(resetTestDbEffect())); + +function sorted(values: readonly string[]): string[] { + return [...values].sort((left, right) => left.localeCompare(right)); +} + +function expectInvalidParams(result: Either.Either): void { + Either.match(result, { + onLeft: (error) => { + expect(error).toBeInstanceOf(InvalidParamsError); + }, + onRight: () => { + expect.fail("Expected InvalidParamsError"); + }, + }); +} + +function insertConversation(creator: AgentId, members: readonly AgentId[]) { + return Effect.gen(function* () { + const db = getKyselyDb(); + const created = yield* db + .insertInto("conversations") + .values({ created_by_id: creator }) + .returning("id"); + const row = created[0]; + if (row === undefined) { + return yield* Effect.die("Conversation insert returned no row"); + } + yield* db.insertInto("conversation_participants").values( + [...new Set([creator, ...members])].map((agentId) => ({ + conversation_id: row.id, + agent_id: agentId, + })), + ); + return row.id; + }); +} + +function insertConversations( + creator: AgentId, + members: readonly AgentId[], + count: number, +) { + return Effect.forEach( + [...Array(count).keys()], + () => insertConversation(creator, members), + { concurrency: 1 }, + ); +} + +describe(agentsSearch.name, () => { + it("browses on blank queries and matches exact ids and names", () => + Effect.gen(function* () { + const { agents } = yield* setupAgentGroup(3); + const [alice, bob, carol] = agents; + if (alice === undefined || bob === undefined || carol === undefined) { + return yield* Effect.die("Expected three connected agents"); + } + + const browse = yield* alice.client.sendRpc(agentsSearch, {}); + const whitespace = yield* alice.client.sendRpc(agentsSearch, { + query: " \t ", + }); + const byId = yield* alice.client.sendRpc(agentsSearch, { + query: carol.agentId, + }); + const byName = yield* alice.client.sendRpc(agentsSearch, { + query: bob.name, + }); + const unknown = yield* alice.client.sendRpc(agentsSearch, { + query: "unknown-agent", + }); + + const expectedIds = sorted(agents.map((agent) => agent.agentId)); + expect(browse.agents.map((agent) => agent.id)).toEqual(expectedIds); + expect(whitespace.agents.map((agent) => agent.id)).toEqual(expectedIds); + expect(byId.agents.map((agent) => agent.id)).toEqual([carol.agentId]); + expect(byName.agents.map((agent) => agent.id)).toEqual([bob.agentId]); + expect(unknown.agents).toEqual([]); + })); + + it("pages in stable id order and rejects cursor binding mismatches", () => + Effect.gen(function* () { + const { alice, bob } = yield* setupAgentPair(); + const extra = yield* Effect.forEach( + [...Array(SEARCH_PAGE_SIZE - 1).keys()], + (index) => + createTestAgent(`search-extra-${String(index).padStart(2, "0")}`), + { concurrency: 1 }, + ); + + const first = yield* alice.client.sendRpc(agentsSearch, {}); + expect(first.agents).toHaveLength(SEARCH_PAGE_SIZE); + expect(first.nextCursor).toBeDefined(); + const cursor = first.nextCursor; + if (cursor === undefined) { + return yield* Effect.die("Expected overflowing agent search page"); + } + const second = yield* alice.client.sendRpc(agentsSearch, { cursor }); + const allIds = [ + alice.agentId, + bob.agentId, + ...extra.map((agent) => agent.agentId), + ]; + expect([ + ...first.agents.map((agent) => agent.id), + ...second.agents.map((agent) => agent.id), + ]).toEqual(sorted(allIds)); + expect(second.nextCursor).toBeUndefined(); + + expectInvalidParams( + yield* Effect.either( + alice.client.sendRpc(agentsSearch, { + query: alice.name, + cursor, + }), + ), + ); + expectInvalidParams( + yield* Effect.either(bob.client.sendRpc(agentsSearch, { cursor })), + ); + expectInvalidParams( + yield* Effect.either( + alice.client.sendRpc(conversationSearch, { cursor }), + ), + ); + })); +}); +describe(conversationSearch.name, () => { + it("matches exact conversation and current-member tokens within visibility", () => + Effect.gen(function* () { + const { agents } = yield* setupAgentGroup(3); + const [alice, bob, carol] = agents; + if (alice === undefined || bob === undefined || carol === undefined) { + return yield* Effect.die("Expected three connected agents"); + } + const aliceBob = yield* insertConversation(alice.agentId, [bob.agentId]); + const aliceCarol = yield* insertConversation(alice.agentId, [ + carol.agentId, + ]); + const bobCarol = yield* insertConversation(bob.agentId, [carol.agentId]); + const group = yield* insertConversation(alice.agentId, [ + bob.agentId, + carol.agentId, + ]); + + const browse = yield* alice.client.sendRpc(conversationSearch, {}); + const whitespace = yield* alice.client.sendRpc(conversationSearch, { + query: " ", + }); + const byConversation = yield* alice.client.sendRpc(conversationSearch, { + query: aliceCarol, + }); + const byMemberId = yield* alice.client.sendRpc(conversationSearch, { + query: bob.agentId, + }); + const byMemberName = yield* alice.client.sendRpc(conversationSearch, { + query: bob.name, + }); + const hidden = yield* alice.client.sendRpc(conversationSearch, { + query: bobCarol, + }); + const unknown = yield* alice.client.sendRpc(conversationSearch, { + query: "unknown-member", + }); + + const visible = sorted([aliceBob, aliceCarol, group]); + const withBob = sorted([aliceBob, group]); + expect( + browse.conversations.map((conversation) => conversation.id), + ).toEqual(visible); + expect( + whitespace.conversations.map((conversation) => conversation.id), + ).toEqual(visible); + expect( + byConversation.conversations.map((conversation) => conversation.id), + ).toEqual([aliceCarol]); + expect( + byMemberId.conversations.map((conversation) => conversation.id), + ).toEqual(withBob); + expect( + byMemberName.conversations.map((conversation) => conversation.id), + ).toEqual(withBob); + expect(hidden.conversations).toEqual([]); + expect(unknown.conversations).toEqual([]); + })); + + it("pages visible conversations by id and binds the cursor to the caller", () => + Effect.gen(function* () { + const { alice, bob } = yield* setupAgentPair(); + const ids = yield* insertConversations( + alice.agentId, + [bob.agentId], + SEARCH_PAGE_SIZE + 1, + ); + + const first = yield* alice.client.sendRpc(conversationSearch, {}); + expect(first.conversations).toHaveLength(SEARCH_PAGE_SIZE); + expect(first.nextCursor).toBeDefined(); + const cursor = first.nextCursor; + if (cursor === undefined) { + return yield* Effect.die( + "Expected overflowing conversation search page", + ); + } + const second = yield* alice.client.sendRpc(conversationSearch, { + cursor, + }); + expect([ + ...first.conversations.map((conversation) => conversation.id), + ...second.conversations.map((conversation) => conversation.id), + ]).toEqual(sorted(ids)); + expect(second.nextCursor).toBeUndefined(); + + expectInvalidParams( + yield* Effect.either( + bob.client.sendRpc(conversationSearch, { cursor }), + ), + ); + })); +}); +/* eslint-enable max-lines-per-function, sonarjs/max-lines-per-function, max-nested-callbacks -- Restore strict defaults after the integration scenarios. */ diff --git a/packages/server/src/__tests__/integration/messaging/message-read.test.ts b/packages/server/src/__tests__/integration/messaging/message-read.test.ts new file mode 100644 index 000000000..debf2064e --- /dev/null +++ b/packages/server/src/__tests__/integration/messaging/message-read.test.ts @@ -0,0 +1,251 @@ +import { afterAll, beforeAll, beforeEach, expect } from "vitest"; +import { Chunk, Duration, Effect, Either, Fiber, Stream } from "effect"; + +import { + type ConversationCheckpoint, + messageReceivedNotificationDefinition, + messagesRead, + messagesSend, + type Message, +} from "@moltzap/protocol/message"; +import { + agentConversationCreate, + type ConversationId, +} from "@moltzap/protocol/conversation"; +import type { ListCursor } from "@moltzap/protocol/rpc"; +import { WIRE_ERROR_TAG } from "@moltzap/protocol/testing"; +import { ConversationService } from "#conversation"; +import { MessageService } from "#message"; +import { + getKyselyDb, + getTestCoreApp, + it, + registerAndConnect, + resetTestDbEffect, + setupAgentPair, + startTestServerEffect, + stopTestServerEffect, + type ConnectedAgent, +} from "../helpers.js"; + +const READ_PAGE_SIZE = 50; +const OVERFLOW_MESSAGE_COUNT = READ_PAGE_SIZE + 1; +const READ_SETTLE_MS = 100; +const SUBSCRIBE_SETTLE = "10 millis"; +const TEST_TIMEOUT_MS = 30_000; + +beforeAll(() => Effect.runPromise(startTestServerEffect()), 60_000); + +afterAll(() => Effect.runPromise(stopTestServerEffect())); + +beforeEach(() => Effect.runPromise(resetTestDbEffect())); + +function expectWireErrorTag( + outcome: Either.Either, + tag: string, +): void { + Either.match(outcome, { + onLeft: (error) => { + expect( + /* Safe because wire errors are a tagged union asserted by discriminant. */ + (error as { readonly _tag?: string })._tag, + ).toBe(tag); + }, + onRight: () => expect.fail(`expected ${tag}`), + }); +} + +interface ReadFixture { + readonly alice: ConnectedAgent; + readonly intruder: ConnectedAgent; + readonly conversationId: ConversationId; + readonly otherConversationId: ConversationId; + readonly sent: readonly Message[]; +} + +interface ReadPosition { + readonly checkpoint: ConversationCheckpoint; + readonly cursor: ListCursor; + readonly nextCheckpoint: ConversationCheckpoint; +} + +function sendSourceMessages( + alice: ConnectedAgent, + conversationId: ConversationId, +) { + return Effect.gen(function* () { + const sent: Message[] = []; + for (let index = 1; index <= OVERFLOW_MESSAGE_COUNT; index++) { + const result = yield* alice.client.sendRpc(messagesSend, { + conversationId, + parts: [{ type: "text", text: `source-${index}` }], + }); + sent.push(result.message); + } + return sent; + }); +} + +function setupReadFixture() { + return Effect.gen(function* () { + const { alice, bob } = yield* setupAgentPair(); + const intruder = yield* registerAndConnect("message-read-intruder"); + const created = yield* alice.client.sendRpc(agentConversationCreate, { + participants: [bob.agentId], + }); + const other = yield* alice.client.sendRpc(agentConversationCreate, { + participants: [bob.agentId], + }); + const sent = yield* sendSourceMessages(alice, created.conversation.id); + return { + alice, + intruder, + conversationId: created.conversation.id, + otherConversationId: other.conversation.id, + sent, + } satisfies ReadFixture; + }); +} + +function readFrozenPages(fixture: ReadFixture) { + return Effect.gen(function* () { + const firstPage = yield* fixture.alice.client.sendRpc(messagesRead, { + conversationId: fixture.conversationId, + }); + expect(firstPage.messages.map((message) => message.id)).toEqual( + fixture.sent.slice(0, READ_PAGE_SIZE).map((message) => message.id), + ); + const cursor = firstPage.nextCursor; + expect(cursor).toBeDefined(); + if (cursor === undefined) { + return yield* Effect.dieMessage("first read page must have a cursor"); + } + + const inserted = yield* fixture.alice.client.sendRpc(messagesSend, { + conversationId: fixture.conversationId, + parts: [{ type: "text", text: "after-frozen-window" }], + }); + const secondPage = yield* fixture.alice.client.sendRpc(messagesRead, { + conversationId: fixture.conversationId, + cursor, + }); + expect(secondPage.messages.map((message) => message.id)).toEqual( + fixture.sent.slice(READ_PAGE_SIZE).map((message) => message.id), + ); + expect(secondPage.checkpoint).toBe(firstPage.checkpoint); + expect(secondPage.nextCursor).toBeUndefined(); + expect(secondPage.messages.map((message) => message.id)).not.toContain( + inserted.message.id, + ); + return { firstPage, cursor, inserted }; + }); +} + +function readNextCheckpoint( + fixture: ReadFixture, + frozen: Effect.Effect.Success>, +) { + return Effect.gen(function* () { + const app = getTestCoreApp(); + const db = getKyselyDb(); + const restartedService = new MessageService({ + db, + conversations: new ConversationService(db, app.connections), + networkSend: app.networkSendService, + }); + const nextWindow = yield* restartedService.read({ + conversationId: fixture.conversationId, + requesterAgentId: fixture.alice.agentId, + checkpoint: frozen.firstPage.checkpoint, + }); + expect(nextWindow.messages.map((message) => message.id)).toEqual([ + frozen.inserted.message.id, + ]); + const noChange = yield* fixture.alice.client.sendRpc(messagesRead, { + conversationId: fixture.conversationId, + checkpoint: nextWindow.checkpoint, + }); + expect(noChange.messages).toEqual([]); + expect(noChange.checkpoint).toBe(nextWindow.checkpoint); + return { + checkpoint: frozen.firstPage.checkpoint, + cursor: frozen.cursor, + nextCheckpoint: nextWindow.checkpoint, + } satisfies ReadPosition; + }); +} + +function assertPositionValidation( + fixture: ReadFixture, + position: ReadPosition, +) { + return Effect.gen(function* () { + const crossCheckpoint = yield* Effect.either( + fixture.alice.client.sendRpc(messagesRead, { + conversationId: fixture.otherConversationId, + checkpoint: position.checkpoint, + }), + ); + expectWireErrorTag(crossCheckpoint, WIRE_ERROR_TAG.InvalidParams); + const crossCursor = yield* Effect.either( + fixture.alice.client.sendRpc(messagesRead, { + conversationId: fixture.otherConversationId, + cursor: position.cursor, + }), + ); + expectWireErrorTag(crossCursor, WIRE_ERROR_TAG.InvalidParams); + const conflictingPosition = yield* Effect.either( + fixture.alice.client.sendRpc(messagesRead, { + conversationId: fixture.conversationId, + checkpoint: position.checkpoint, + cursor: position.cursor, + }), + ); + expectWireErrorTag(conflictingPosition, WIRE_ERROR_TAG.InvalidParams); + + // Authorization precedes token validation, so an outsider learns nothing + // about the mutually exclusive positions supplied with the request. + const inaccessible = yield* Effect.either( + fixture.intruder.client.sendRpc(messagesRead, { + conversationId: fixture.conversationId, + checkpoint: position.checkpoint, + cursor: position.cursor, + }), + ); + expectWireErrorTag(inaccessible, WIRE_ERROR_TAG.Forbidden); + }); +} + +function assertReadDoesNotNotify( + fixture: ReadFixture, + checkpoint: ConversationCheckpoint, +) { + return Effect.gen(function* () { + const notifications = yield* fixture.alice.client + .subscribe(messageReceivedNotificationDefinition) + .pipe( + Stream.interruptAfter(Duration.millis(READ_SETTLE_MS)), + Stream.runCollect, + Effect.fork, + ); + yield* Effect.sleep(SUBSCRIBE_SETTLE); + yield* fixture.alice.client.sendRpc(messagesRead, { + conversationId: fixture.conversationId, + checkpoint, + }); + expect(Chunk.toReadonlyArray(yield* Fiber.join(notifications))).toEqual([]); + }); +} + +it( + "reads a frozen checkpoint window in source order without dispatch side effects", + () => + Effect.gen(function* () { + const fixture = yield* setupReadFixture(); + const frozen = yield* readFrozenPages(fixture); + const position = yield* readNextCheckpoint(fixture, frozen); + yield* assertPositionValidation(fixture, position); + yield* assertReadDoesNotNotify(fixture, position.nextCheckpoint); + }), + TEST_TIMEOUT_MS, +); diff --git a/packages/server/src/conversation/MODULE.md b/packages/server/src/conversation/MODULE.md index 85f3d65a1..abf8bd9fd 100644 --- a/packages/server/src/conversation/MODULE.md +++ b/packages/server/src/conversation/MODULE.md @@ -8,7 +8,7 @@ Conversation-domain service barrel. ## Public surface -### [`agentConversationCreate`](./handlers.ts#L96) +### [`agentConversationCreate`](./handlers.ts#L120) _Variable_ @@ -24,7 +24,7 @@ Provides the agent conversation create runtime value. **Returns:** The agent conversation create result. -### [`conversationList`](./handlers.ts#L85) +### [`conversationList`](./handlers.ts#L98) _Variable_ @@ -40,7 +40,23 @@ Provides the conversation list runtime value. **Returns:** The conversation list result. -### [`ConversationService`](./conversation.service.ts#L225) +### [`conversationSearch`](./handlers.ts#L109) + +_Variable_ + +```ts +export const conversationSearch: ServerHandler< + typeof conversationSearchDefinition +> = Effect.fn("conversationSearch")(function* (params) { + return yield* conversationSearchBody(params, yield* agentArm); +}) +``` + +Search the active agent's conversations by exact identifier or member. + +**Returns:** One stable identifier-ordered page. + +### [`ConversationService`](./conversation.service.ts#L346) _Class_ diff --git a/packages/server/src/conversation/conversation.service.ts b/packages/server/src/conversation/conversation.service.ts index 5b252fe4a..c243fef58 100644 --- a/packages/server/src/conversation/conversation.service.ts +++ b/packages/server/src/conversation/conversation.service.ts @@ -2,8 +2,12 @@ // safer-arch-ignore folder-explicit-api-required: ConversationService is the deliberate concrete service boundary paired with the public conversation index. import { type Db, + READ_PLANE_PAGE_SIZE, sql, catchSqlErrorAsDefect, + decodeSearchCursor, + normalizeSearchQuery, + paginateSearchRows, rawQuery, takeFirstOption, takeFirstOrFail, @@ -12,20 +16,23 @@ import { import { type Conversation, type ConversationId, + conversationId as conversationIdSchema, ConversationFullError, ConversationNotFoundError, } from "@moltzap/protocol/conversation"; import { type AgentId, type UserId, + agentId as agentIdSchema, AgentNotFoundError, } from "@moltzap/protocol/identity"; import type { SqlError } from "@effect/sql/SqlError"; -import { Effect, Option } from "effect"; +import { Effect, Option, Schema } from "effect"; import { InvalidParamsError, DEFAULT_PAGE_LIMIT, ForbiddenError, + type ListCursor, } from "@moltzap/protocol/rpc"; import type { ConnectionManager } from "#socket"; @@ -53,6 +60,12 @@ interface ListConversationsInput { readonly cursor?: string; } +interface SearchConversationsInput { + readonly agentId: AgentId; + readonly query?: string; + readonly cursor?: string; +} + function mapConversation(row: ConversationColumns): Conversation { return { id: row.id, @@ -75,6 +88,114 @@ interface ConversationPage { readonly cursor?: string; } +interface ConversationSearchPage { + readonly conversations: readonly Conversation[]; + readonly nextCursor?: ListCursor; +} + +function conversationSearchTokenFilter(normalizedQuery: string) { + if (normalizedQuery === "") { + return sql``; + } + const searchId = Schema.decodeOption(agentIdSchema)(normalizedQuery); + if (Option.isSome(searchId)) { + return sql` + AND ( + conversation.id = ${searchId.value}::uuid + OR EXISTS ( + SELECT 1 + FROM conversation_participants matching_membership + WHERE matching_membership.conversation_id = conversation.id + AND matching_membership.agent_id = ${searchId.value}::uuid + ) + ) + `; + } + return sql` + AND EXISTS ( + SELECT 1 + FROM conversation_participants matching_membership + JOIN agents matching_agent + ON matching_agent.id = matching_membership.agent_id + WHERE matching_membership.conversation_id = conversation.id + AND matching_agent.name = ${normalizedQuery} + ) + `; +} + +function conversationSearchCursorFilter(lastId?: ConversationId) { + return lastId === undefined + ? sql`` + : sql`AND conversation.id > ${lastId}::uuid`; +} + +function queryConversationSearchRows( + db: Db, + input: { + readonly agentId: AgentId; + readonly normalizedQuery: string; + readonly lastId?: ConversationId; + }, +): Effect.Effect { + return rawQuery( + db, + sql` + SELECT + conversation.id, + conversation.name, + conversation.created_by_id, + conversation.created_at, + conversation.updated_at + FROM conversation_participants caller_membership + JOIN conversations conversation + ON conversation.id = caller_membership.conversation_id + WHERE caller_membership.agent_id = ${input.agentId} + ${conversationSearchTokenFilter(input.normalizedQuery)} + ${conversationSearchCursorFilter(input.lastId)} + ORDER BY conversation.id ASC + LIMIT ${READ_PLANE_PAGE_SIZE + 1} + `, + ); +} + +function searchConversations( + db: Db, + input: SearchConversationsInput, +): Effect.Effect { + return catchSqlErrorAsDefect( + Effect.gen(function* () { + const normalizedQuery = normalizeSearchQuery(input.query); + const binding = { + kind: "conversations" as const, + query: normalizedQuery, + agentId: input.agentId, + }; + const cursorPosition = + input.cursor === undefined + ? undefined + : yield* decodeSearchCursor(input.cursor, binding); + const lastId = + cursorPosition === undefined + ? undefined + : Schema.decodeSync(conversationIdSchema)(cursorPosition.lastId); + const rows = yield* queryConversationSearchRows(db, { + agentId: input.agentId, + normalizedQuery, + ...(lastId === undefined ? {} : { lastId }), + }); + const { page, nextCursor } = paginateSearchRows( + rows, + binding, + (row) => row.id, + ); + return { + conversations: page.map(mapConversation), + ...(nextCursor === undefined ? {} : { nextCursor }), + }; + }), + ).pipe(Effect.withSpan("searchConversations")); +} + // Two queries regardless of page size: one for the page, one for the // membership of every conversation on it. function listConversations( @@ -131,7 +252,7 @@ function queryParticipantsFor( // The cursor carries both halves of the sort key. Paging on a different // expression than the one that orders the page lets a row move across the // boundary between requests and vanish from every later page. -interface ListCursor { +interface ConversationListCursor { readonly updatedAt: string; readonly id: string; } @@ -155,7 +276,7 @@ const CURSOR_ID_RE = function parseListCursor( cursor?: string, -): Effect.Effect { +): Effect.Effect { if (cursor == null) { return Effect.succeed(null); } @@ -178,7 +299,7 @@ function parseListCursor( interface ListRowsInput { readonly agentId: AgentId; readonly limit: number; - readonly cursorParam: ListCursor | null; + readonly cursorParam: ConversationListCursor | null; } // Sort key and cursor key are the same stored pair, so the page boundary lands @@ -202,7 +323,7 @@ function queryConversationListRows( ); } -function cursorListFilter(cursorParam: ListCursor | null) { +function cursorListFilter(cursorParam: ConversationListCursor | null) { if (cursorParam === null) { return sql``; } @@ -370,6 +491,14 @@ export class ConversationService { return listConversations(this.db, { agentId, limit, cursor }); } + search( + agentId: AgentId, + query?: string, + cursor?: string, + ): Effect.Effect { + return searchConversations(this.db, { agentId, query, cursor }); + } + getParticipantAgentIds( conversationId: ConversationId, ): Effect.Effect { diff --git a/packages/server/src/conversation/handlers.ts b/packages/server/src/conversation/handlers.ts index d67b99e30..3b6964a03 100644 --- a/packages/server/src/conversation/handlers.ts +++ b/packages/server/src/conversation/handlers.ts @@ -3,6 +3,7 @@ import { type agentConversationCreate as agentConversationCreateDefinition, conversationCreatedNotificationDefinition, type conversationList as conversationListDefinition, + type conversationSearch as conversationSearchDefinition, type Conversation, type ConversationListItem, } from "@moltzap/protocol/conversation"; @@ -77,6 +78,18 @@ const conversationListBody = Effect.fn("conversation.list")(function* ( return { items, ...(nextCursor !== undefined ? { nextCursor } : {}) }; }); +const conversationSearchBody = Effect.fn("conversation.search")(function* ( + params: ParamsOf, + ctx: AgentContext, +) { + const conversationService = yield* ConversationServiceTag; + return yield* conversationService.search( + ctx.agentId, + params.query, + params.cursor, + ); +}); + /** * Provides the conversation list runtime value. * @param params Request payload to process. @@ -88,6 +101,17 @@ export const conversationList: ServerHandler< return yield* conversationListBody(params, yield* agentArm); }); +/** + * Search the active agent's conversations by exact identifier or member. + * @param params Request payload to process. + * @returns One stable identifier-ordered page. + */ +export const conversationSearch: ServerHandler< + typeof conversationSearchDefinition +> = Effect.fn("conversationSearch")(function* (params) { + return yield* conversationSearchBody(params, yield* agentArm); +}); + /** * Provides the agent conversation create runtime value. * @param params Request payload to process. diff --git a/packages/server/src/db/barrel.ts b/packages/server/src/db/barrel.ts index 2efa0359b..3e143ee99 100644 --- a/packages/server/src/db/barrel.ts +++ b/packages/server/src/db/barrel.ts @@ -26,6 +26,17 @@ export { } from "./list-cursor.js"; /** Re-exports the public API from `./list-cursor.js`. */ export type { ListCursorPosition } from "./list-cursor.js"; +/** Re-exports the public API from `./search-read-cursor.js`. */ +export { + READ_PLANE_PAGE_SIZE, + decodeConversationCheckpoint, + decodeConversationReadCursor, + decodeSearchCursor, + encodeConversationCheckpoint, + encodeConversationReadCursor, + normalizeSearchQuery, + paginateSearchRows, +} from "./search-read-cursor.js"; /** Re-exports the public API from `./kysely-vendor.js`. */ /** Re-exports the public API from `./postgres-dialect.js`. */ export { PostgresDialect } from "./postgres-dialect.js"; diff --git a/packages/server/src/db/search-read-cursor.test.ts b/packages/server/src/db/search-read-cursor.test.ts new file mode 100644 index 000000000..44825f1ad --- /dev/null +++ b/packages/server/src/db/search-read-cursor.test.ts @@ -0,0 +1,198 @@ +/* eslint-disable max-lines-per-function, sonarjs/max-lines-per-function -- Codec scenarios keep each binding and malformed-token matrix beside its roundtrip setup. */ +import { describe, expect, it } from "vitest"; +import { Effect, Either, Schema } from "effect"; +import { conversationId } from "@moltzap/protocol/conversation"; +import { agentId } from "@moltzap/protocol/identity"; +import { InvalidParamsError } from "@moltzap/protocol/rpc"; +import { + READ_PLANE_PAGE_SIZE, + decodeConversationCheckpoint, + decodeConversationReadCursor, + decodeSearchCursor, + encodeConversationCheckpoint, + encodeConversationReadCursor, + encodeSearchCursor, + normalizeSearchQuery, + paginateSearchRows, +} from "./search-read-cursor.js"; + +const CALLER_ID = Schema.decodeSync(agentId)( + "00000000-0000-4000-8000-000000000001", +); +const OTHER_AGENT_ID = Schema.decodeSync(agentId)( + "00000000-0000-4000-8000-000000000002", +); +const CONVERSATION_ID = Schema.decodeSync(conversationId)( + "00000000-0000-4000-8000-000000000010", +); +const OTHER_CONVERSATION_ID = Schema.decodeSync(conversationId)( + "00000000-0000-4000-8000-000000000011", +); +const LAST_ID = "00000000-0000-4000-8000-000000000020"; +const NORMALIZED_QUERY = "exact-name"; + +function expectInvalidParams(effect: Effect.Effect) { + const result = Effect.runSync(Effect.either(effect)); + Either.match(result, { + onLeft: (error) => { + expect(error).toBeInstanceOf(InvalidParamsError); + }, + onRight: () => { + expect.fail("Expected InvalidParamsError"); + }, + }); +} + +function encodeTestPayload(value: unknown): string { + return Buffer.from(JSON.stringify(value), "utf8").toString("base64url"); +} + +// @agent-code-guard/regression-only: fixed cursor bindings and malformed encodings are closed boundary cases rather than a generative input space. +describe("search cursor", () => { + it("normalizes omitted and whitespace-only queries to browse", () => { + expect(normalizeSearchQuery()).toBe(""); + expect(normalizeSearchQuery(" \t\n ")).toBe(""); + expect(normalizeSearchQuery(` ${NORMALIZED_QUERY} `)).toBe( + NORMALIZED_QUERY, + ); + }); + + it("roundtrips the operation, query, caller, and last id binding", () => { + const binding = { + kind: "agents" as const, + query: NORMALIZED_QUERY, + agentId: CALLER_ID, + }; + const cursor = encodeSearchCursor({ ...binding, lastId: LAST_ID }); + + expect(Effect.runSync(decodeSearchCursor(cursor, binding))).toEqual({ + lastId: LAST_ID, + }); + }); + + it("rejects cross-operation, query, and caller reuse", () => { + const binding = { + kind: "agents" as const, + query: NORMALIZED_QUERY, + agentId: CALLER_ID, + }; + const cursor = encodeSearchCursor({ ...binding, lastId: LAST_ID }); + + expectInvalidParams( + decodeSearchCursor(cursor, { ...binding, kind: "conversations" }), + ); + expectInvalidParams( + decodeSearchCursor(cursor, { ...binding, query: "different" }), + ); + expectInvalidParams( + decodeSearchCursor(cursor, { + ...binding, + agentId: OTHER_AGENT_ID, + }), + ); + }); + + it("rejects malformed and non-canonical tokens", () => { + const binding = { + kind: "agents" as const, + query: "", + agentId: CALLER_ID, + }; + expectInvalidParams(decodeSearchCursor("not-base64url!", binding)); + expectInvalidParams( + decodeSearchCursor( + encodeTestPayload({ + version: 1, + query: "", + lastId: LAST_ID, + kind: "agents", + agentId: CALLER_ID, + }), + binding, + ), + ); + }); + + it("emits a continuation only when the fixed-size page overflows", () => { + const binding = { + kind: "agents" as const, + query: "", + agentId: CALLER_ID, + }; + const rows = [...Array(READ_PLANE_PAGE_SIZE + 1).keys()].map((index) => ({ + id: `00000000-0000-4000-8000-${String(index).padStart(12, "0")}`, + })); + const result = paginateSearchRows(rows, binding, (row) => row.id); + + expect(result.page).toHaveLength(READ_PLANE_PAGE_SIZE); + expect(result.nextCursor).toBeDefined(); + expect( + Effect.runSync( + decodeSearchCursor( + /* Safe because this overflow fixture always produces a cursor. */ + result.nextCursor!, + binding, + ), + ), + ).toEqual({ lastId: rows[READ_PLANE_PAGE_SIZE - 1]?.id }); + }); +}); + +// @agent-code-guard/regression-only: checkpoint and frozen-page token cases pin a finite wire boundary. +describe("conversation read positions", () => { + it("roundtrips a conversation-bound durable checkpoint", () => { + const checkpoint = encodeConversationCheckpoint({ + conversationId: CONVERSATION_ID, + throughSeq: "123456789", + }); + + expect( + Effect.runSync(decodeConversationCheckpoint(checkpoint, CONVERSATION_ID)), + ).toEqual({ throughSeq: "123456789" }); + expectInvalidParams( + decodeConversationCheckpoint(checkpoint, OTHER_CONVERSATION_ID), + ); + }); + + it("roundtrips a frozen page cursor and rejects an inverted interval", () => { + const cursor = encodeConversationReadCursor({ + conversationId: CONVERSATION_ID, + throughSeq: "200", + afterSeq: "100", + }); + expect( + Effect.runSync(decodeConversationReadCursor(cursor, CONVERSATION_ID)), + ).toEqual({ throughSeq: "200", afterSeq: "100" }); + + const inverted = encodeConversationReadCursor({ + conversationId: CONVERSATION_ID, + throughSeq: "100", + afterSeq: "101", + }); + expectInvalidParams( + decodeConversationReadCursor(inverted, CONVERSATION_ID), + ); + }); + + it("rejects non-canonical decimal strings and cross-conversation reuse", () => { + const cursor = encodeConversationReadCursor({ + conversationId: CONVERSATION_ID, + throughSeq: "200", + afterSeq: "100", + }); + expectInvalidParams( + decodeConversationReadCursor(cursor, OTHER_CONVERSATION_ID), + ); + + const checkpoint = encodeTestPayload({ + conversationId: CONVERSATION_ID, + kind: "conversation-checkpoint", + throughSeq: "0200", + version: 1, + }); + expectInvalidParams( + decodeConversationCheckpoint(checkpoint, CONVERSATION_ID), + ); + }); +}); +/* eslint-enable max-lines-per-function, sonarjs/max-lines-per-function -- Restore strict defaults after the codec scenarios. */ diff --git a/packages/server/src/db/search-read-cursor.ts b/packages/server/src/db/search-read-cursor.ts new file mode 100644 index 000000000..ee7c06123 --- /dev/null +++ b/packages/server/src/db/search-read-cursor.ts @@ -0,0 +1,418 @@ +/** + * Opaque cursor and checkpoint codecs for the stable read plane. + * + * Search cursors bind the page position to the operation, normalized query, + * and authenticated agent. Conversation reads use a cursor for one frozen + * page chain and a separate checkpoint for the durable high-water mark. + */ +import type { ConversationId } from "@moltzap/protocol/conversation"; +import type { AgentId } from "@moltzap/protocol/identity"; +import { + conversationCheckpoint, + type ConversationCheckpoint, +} from "@moltzap/protocol/message"; +import { + InvalidParamsError, + listCursorSchema, + type ListCursor, +} from "@moltzap/protocol/rpc"; +import { Effect, Schema } from "effect"; + +/** The server-owned page size for directory and conversation reads. */ +export const READ_PLANE_PAGE_SIZE = 50; + +const CODEC_VERSION = 1; +const UUID_RE = + /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/; +const DECIMAL_RE = /^(?:0|[1-9]\d*)$/; + +/** Identifies the search operation a cursor may continue. */ +type SearchCursorKind = "agents" | "conversations"; + +/** Values that bind a search cursor to one caller and request. */ +export interface SearchCursorBinding { + readonly kind: SearchCursorKind; + readonly query: string; + readonly agentId: AgentId; +} + +/** Search cursor position after the last emitted stable identifier. */ +export interface SearchCursorPosition { + readonly lastId: string; +} + +/** Frozen position carried between pages of one conversation read. */ +export interface ConversationReadCursorPosition { + readonly throughSeq: string; + readonly afterSeq: string; +} + +/** Durable high-water mark recovered from a conversation checkpoint. */ +export interface ConversationCheckpointPosition { + readonly throughSeq: string; +} + +interface SearchCursorPayload { + readonly agentId: string; + readonly kind: SearchCursorKind; + readonly lastId: string; + readonly query: string; + readonly version: number; +} + +interface ConversationReadCursorPayload { + readonly afterSeq: string; + readonly conversationId: string; + readonly kind: "conversation-read-page"; + readonly throughSeq: string; + readonly version: number; +} + +interface ConversationCheckpointPayload { + readonly conversationId: string; + readonly kind: "conversation-checkpoint"; + readonly throughSeq: string; + readonly version: number; +} + +/** + * Trim a search query; the empty string is the canonical browse query. + * @param query Untrusted request query. + * @returns The normalized cursor binding value. + */ +export function normalizeSearchQuery(query?: string): string { + return query?.trim() ?? ""; +} + +function invalidParams(message: string): InvalidParamsError { + return new InvalidParamsError({ message }); +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function hasExactKeys( + value: Record, + keys: readonly string[], +): boolean { + const compare = (left: string, right: string) => left.localeCompare(right); + const actual = Object.keys(value).sort(compare); + const expected = [...keys].sort(compare); + return ( + actual.length === expected.length && + actual.every((key, index) => key === expected[index]) + ); +} + +function isCanonicalUuid(value: unknown): value is string { + return typeof value === "string" && UUID_RE.test(value); +} + +function isCanonicalDecimal(value: unknown): value is string { + return typeof value === "string" && DECIMAL_RE.test(value); +} + +function encodePayload(payload: object): string { + return Buffer.from(JSON.stringify(payload), "utf8").toString("base64url"); +} + +function decodePayload( + token: string, +): Effect.Effect, InvalidParamsError> { + return Effect.try({ + try: () => Buffer.from(token, "base64url"), + catch: () => invalidParams("Cursor is not base64url"), + }).pipe( + Effect.flatMap((bytes) => { + if (bytes.toString("base64url") !== token) { + return Effect.fail(invalidParams("Cursor is not canonical base64url")); + } + const json = bytes.toString("utf8"); + return Effect.try({ + try: () => { + const value: unknown = JSON.parse(json); + return { json, value }; + }, + catch: () => invalidParams("Cursor payload is not valid JSON"), + }); + }), + Effect.flatMap(({ json, value }) => { + if (!isRecord(value) || JSON.stringify(value) !== json) { + return Effect.fail( + invalidParams("Cursor payload is not canonical JSON"), + ); + } + return Effect.succeed(value); + }), + ); +} + +function searchPayload( + binding: SearchCursorBinding, + lastId: string, +): SearchCursorPayload { + return { + agentId: binding.agentId, + kind: binding.kind, + lastId, + query: binding.query, + version: CODEC_VERSION, + }; +} + +function isSearchPayloadFor( + value: Record, + binding: SearchCursorBinding, +): value is Record & SearchCursorPayload { + if (!hasExactKeys(value, ["agentId", "kind", "lastId", "query", "version"])) { + return false; + } + if (value.version !== CODEC_VERSION || value.kind !== binding.kind) { + return false; + } + if (value.query !== binding.query || value.agentId !== binding.agentId) { + return false; + } + return isCanonicalUuid(value.lastId); +} + +/** + * Encode the position after one search page. + * @param input Bound request and last emitted identifier. + * @returns An opaque search cursor. + */ +export function encodeSearchCursor( + input: SearchCursorBinding & SearchCursorPosition, +): ListCursor { + return Schema.decodeSync(listCursorSchema())( + encodePayload(searchPayload(input, input.lastId)), + ); +} + +/** + * Decode and validate a search cursor against its active request binding. + * @param cursor Opaque continuation supplied by the caller. + * @param binding Active operation, query, and agent identity. + * @returns The last emitted stable identifier. + */ +export function decodeSearchCursor( + cursor: ListCursor | string, + binding: SearchCursorBinding, +): Effect.Effect { + return decodePayload(cursor).pipe( + Effect.flatMap((value) => { + if (!isSearchPayloadFor(value, binding)) { + return Effect.fail( + invalidParams("Cursor does not match this search request"), + ); + } + const expected = searchPayload(binding, value.lastId); + if (encodePayload(expected) !== cursor) { + return Effect.fail(invalidParams("Cursor payload is not canonical")); + } + return Effect.succeed({ lastId: value.lastId }); + }), + ); +} + +/** + * Split a fixed-size `page + 1` search batch and encode its continuation. + * @param rows Ordered result batch containing at most one overflow row. + * @param binding Active search cursor binding. + * @param idOf Selects a row's stable identifier. + * @returns The visible page and optional continuation cursor. + */ +export function paginateSearchRows( + rows: readonly Row[], + binding: SearchCursorBinding, + idOf: (row: Row) => string, +): { readonly page: readonly Row[]; readonly nextCursor?: ListCursor } { + if (rows.length <= READ_PLANE_PAGE_SIZE) { + return { page: rows }; + } + const page = rows.slice(0, READ_PLANE_PAGE_SIZE); + const last = page[page.length - 1]; + if (last === undefined) { + return { page }; + } + return { + page, + nextCursor: encodeSearchCursor({ ...binding, lastId: idOf(last) }), + }; +} + +function checkpointPayload( + conversationId: ConversationId, + throughSeq: string, +): ConversationCheckpointPayload { + return { + conversationId, + kind: "conversation-checkpoint", + throughSeq, + version: CODEC_VERSION, + }; +} + +function isCheckpointPayloadFor( + value: Record, + conversationId: ConversationId, +): value is Record & ConversationCheckpointPayload { + if ( + !hasExactKeys(value, ["conversationId", "kind", "throughSeq", "version"]) + ) { + return false; + } + if ( + value.version !== CODEC_VERSION || + value.kind !== "conversation-checkpoint" + ) { + return false; + } + if (value.conversationId !== conversationId) { + return false; + } + return isCanonicalDecimal(value.throughSeq); +} + +/** + * Encode a durable, conversation-bound read checkpoint. + * @param input Stable conversation high-water mark. + * @param input.conversationId Conversation owning the checkpoint. + * @param input.throughSeq Canonical decimal high-water sequence. + * @returns An opaque durable checkpoint. + */ +export function encodeConversationCheckpoint(input: { + readonly conversationId: ConversationId; + readonly throughSeq: string; +}): ConversationCheckpoint { + return Schema.decodeSync(conversationCheckpoint)( + encodePayload(checkpointPayload(input.conversationId, input.throughSeq)), + ); +} + +/** + * Decode a checkpoint and prove it belongs to the requested conversation. + * @param checkpoint Opaque durable checkpoint supplied by the caller. + * @param conversationId Requested conversation. + * @returns The stable high-water sequence. + */ +export function decodeConversationCheckpoint( + checkpoint: ConversationCheckpoint | string, + conversationId: ConversationId, +): Effect.Effect { + return decodePayload(checkpoint).pipe( + Effect.flatMap((value) => { + if (!isCheckpointPayloadFor(value, conversationId)) { + return Effect.fail( + invalidParams("Checkpoint does not match this conversation"), + ); + } + const expected = checkpointPayload(conversationId, value.throughSeq); + if (encodePayload(expected) !== checkpoint) { + return Effect.fail( + invalidParams("Checkpoint payload is not canonical"), + ); + } + return Effect.succeed({ throughSeq: value.throughSeq }); + }), + ); +} + +function readCursorPayload(input: { + readonly conversationId: ConversationId; + readonly throughSeq: string; + readonly afterSeq: string; +}): ConversationReadCursorPayload { + return { + afterSeq: input.afterSeq, + conversationId: input.conversationId, + kind: "conversation-read-page", + throughSeq: input.throughSeq, + version: CODEC_VERSION, + }; +} + +function isReadCursorPayloadFor( + value: Record, + conversationId: ConversationId, +): value is Record & ConversationReadCursorPayload { + if ( + !hasExactKeys(value, [ + "afterSeq", + "conversationId", + "kind", + "throughSeq", + "version", + ]) + ) { + return false; + } + if ( + value.version !== CODEC_VERSION || + value.kind !== "conversation-read-page" + ) { + return false; + } + if (value.conversationId !== conversationId) { + return false; + } + if ( + !isCanonicalDecimal(value.throughSeq) || + !isCanonicalDecimal(value.afterSeq) + ) { + return false; + } + return BigInt(value.afterSeq) <= BigInt(value.throughSeq); +} + +/** + * Encode one continuation within a frozen conversation page chain. + * @param input Frozen conversation read interval. + * @param input.conversationId Conversation owning the page chain. + * @param input.throughSeq Frozen canonical decimal high-water sequence. + * @param input.afterSeq Last emitted canonical decimal sequence. + * @returns An opaque page cursor. + */ +export function encodeConversationReadCursor(input: { + readonly conversationId: ConversationId; + readonly throughSeq: string; + readonly afterSeq: string; +}): ListCursor { + return Schema.decodeSync(listCursorSchema())( + encodePayload(readCursorPayload(input)), + ); +} + +/** + * Decode a frozen conversation cursor and validate its sequence interval. + * @param cursor Opaque page continuation supplied by the caller. + * @param conversationId Requested conversation. + * @returns The frozen high-water and last-emitted sequences. + */ +export function decodeConversationReadCursor( + cursor: ListCursor | string, + conversationId: ConversationId, +): Effect.Effect { + return decodePayload(cursor).pipe( + Effect.flatMap((value) => { + if (!isReadCursorPayloadFor(value, conversationId)) { + return Effect.fail( + invalidParams("Cursor does not match this conversation read"), + ); + } + const expected = readCursorPayload({ + conversationId, + throughSeq: value.throughSeq, + afterSeq: value.afterSeq, + }); + if (encodePayload(expected) !== cursor) { + return Effect.fail(invalidParams("Cursor payload is not canonical")); + } + return Effect.succeed({ + throughSeq: value.throughSeq, + afterSeq: value.afterSeq, + }); + }), + ); +} diff --git a/packages/server/src/identity/agents/MODULE.md b/packages/server/src/identity/agents/MODULE.md index 6b6c6c5d3..9a1daff51 100644 --- a/packages/server/src/identity/agents/MODULE.md +++ b/packages/server/src/identity/agents/MODULE.md @@ -8,7 +8,7 @@ Agent identity server internals. ## Public surface -### [`agentsList`](./handlers.ts#L123) +### [`agentsList`](./handlers.ts#L205) _Variable_ @@ -24,6 +24,21 @@ Provides the agents list runtime value. **Returns:** The agents list result. +### [`agentsSearch`](./handlers.ts#L216) + +_Variable_ + +```ts +export const agentsSearch: ServerHandler = + Effect.fn("agentsSearch")(function* (params) { + return yield* agentsSearchBody(params, yield* agentArm); + }) +``` + +Search agent cards by exact identifier or exact name. + +**Returns:** One stable identifier-ordered page. + ### [`AuthService`](./auth.service.ts#L24) _Class_ diff --git a/packages/server/src/identity/agents/handlers.ts b/packages/server/src/identity/agents/handlers.ts index df12d4bb0..59463b0eb 100644 --- a/packages/server/src/identity/agents/handlers.ts +++ b/packages/server/src/identity/agents/handlers.ts @@ -1,10 +1,12 @@ -import { Effect, Schema } from "effect"; +import { Effect, Option, Schema } from "effect"; import { agentName, type agentsList as agentsListDefinition, + type agentsSearch as agentsSearchDefinition, type AgentCard, type AgentId, type UserId, + agentId, } from "@moltzap/protocol/identity"; import { DEFAULT_PAGE_LIMIT, @@ -14,13 +16,19 @@ import { import type { ServerHandler } from "@moltzap/protocol/socket/catalog"; import { DbTag, + READ_PLANE_PAGE_SIZE, catchSqlErrorAsDefect, decodeListCursor, + decodeSearchCursor, keysetWhere, + normalizeSearchQuery, paginate, + paginateSearchRows, sortKeyExpr, type ListCursorPosition, } from "#db"; +import { agentArm } from "#moltzap/runtime"; +import type { AgentContext } from "#socket"; function toAgentCard(row: { id: AgentId; @@ -96,6 +104,57 @@ const agentsListPageEffect = Effect.fn("agents.list")(function* ( const agentsListPage = (input: AgentsListPageInput) => catchSqlErrorAsDefect(agentsListPageEffect(input)); +interface AgentsSearchPageInput { + readonly normalizedQuery: string; + readonly agentId: AgentId; + readonly lastId?: AgentId; +} + +const agentsSearchPageEffect = Effect.fn("agents.search")(function* ( + input: AgentsSearchPageInput, +) { + const db = yield* DbTag; + const searchId = Schema.decodeOption(agentId)(input.normalizedQuery); + let query = db + .selectFrom("agents") + .select([ + "id", + "name", + "display_name", + "description", + "status", + "owner_user_id", + ]); + if (input.normalizedQuery !== "") { + query = Option.isSome(searchId) + ? query.where("id", "=", searchId.value) + : query.where("name", "=", input.normalizedQuery); + } + if (input.lastId !== undefined) { + query = query.where("id", ">", input.lastId); + } + const rows = yield* query + .orderBy("id", "asc") + .limit(READ_PLANE_PAGE_SIZE + 1); + const binding = { + kind: "agents" as const, + query: input.normalizedQuery, + agentId: input.agentId, + }; + const { page, nextCursor } = paginateSearchRows( + rows, + binding, + (row) => row.id, + ); + return { + agents: page.map(toAgentCard), + ...(nextCursor === undefined ? {} : { nextCursor }), + }; +}); + +const agentsSearchPage = (input: AgentsSearchPageInput) => + catchSqlErrorAsDefect(agentsSearchPageEffect(input)); + const agentsListBody = Effect.fn("agents.list.handler")(function* ( params: ParamsOf, ) { @@ -113,6 +172,29 @@ const agentsListBody = Effect.fn("agents.list.handler")(function* ( }); }); +const agentsSearchBody = Effect.fn("agents.search.handler")(function* ( + params: ParamsOf, + ctx: AgentContext, +) { + const normalizedQuery = normalizeSearchQuery(params.query); + const binding = { + kind: "agents" as const, + query: normalizedQuery, + agentId: ctx.agentId, + }; + const position = + params.cursor === undefined + ? undefined + : yield* decodeSearchCursor(params.cursor, binding); + return yield* agentsSearchPage({ + normalizedQuery, + agentId: ctx.agentId, + ...(position === undefined + ? {} + : { lastId: Schema.decodeSync(agentId)(position.lastId) }), + }); +}); + // ── @effect/rpc handler bodies ─────────────────────────────────────── /** @@ -125,3 +207,13 @@ export const agentsList: ServerHandler = Effect.fn( )(function* (params) { return yield* agentsListBody(params); }); + +/** + * Search agent cards by exact identifier or exact name. + * @param params Request payload to process. + * @returns One stable identifier-ordered page. + */ +export const agentsSearch: ServerHandler = + Effect.fn("agentsSearch")(function* (params) { + return yield* agentsSearchBody(params, yield* agentArm); + }); diff --git a/packages/server/src/identity/agents/index.ts b/packages/server/src/identity/agents/index.ts index 474809885..ad837786c 100644 --- a/packages/server/src/identity/agents/index.ts +++ b/packages/server/src/identity/agents/index.ts @@ -1,7 +1,7 @@ /** @file Agent identity server internals. */ /** Re-exports the public API from `./handlers.js`. */ -export { agentsList } from "./handlers.js"; +export { agentsList, agentsSearch } from "./handlers.js"; /** Re-exports the public API from `./auth.service.js`. */ export { AuthService } from "./auth.service.js"; /** Re-exports the public API from `./layer.js`. */ diff --git a/packages/server/src/message/MODULE.md b/packages/server/src/message/MODULE.md index 7c42351f0..3d7fedca7 100644 --- a/packages/server/src/message/MODULE.md +++ b/packages/server/src/message/MODULE.md @@ -8,7 +8,7 @@ Message-domain service barrel. ## Public surface -### [`MessageService`](./message.service.ts#L93) +### [`MessageService`](./message.service.ts#L125) _Class_ @@ -177,7 +177,7 @@ export class MessageServiceTag extends Context.Tag("moltzap/MessageService")< Implements message service tag. -### [`messagesList`](./handlers.ts#L64) +### [`messagesList`](./handlers.ts#L80) _Variable_ @@ -195,7 +195,23 @@ Provides the messages list runtime value. **Returns:** The messages list result. -### [`messagesSend`](./handlers.ts#L50) +### [`messagesRead`](./handlers.ts#L93) + +_Variable_ + +```ts +export const messagesRead: ServerHandler = + Effect.fn("messagesRead")(function* (params) { + const ctx = yield* agentArm; + return yield* handleMessageRead(params, ctx); + }) +``` + +Provides the checkpointed messages read runtime value. + +**Returns:** The messages read result. + +### [`messagesSend`](./handlers.ts#L66) _Variable_ diff --git a/packages/server/src/message/handlers.ts b/packages/server/src/message/handlers.ts index c94fc466d..39b0ccde1 100644 --- a/packages/server/src/message/handlers.ts +++ b/packages/server/src/message/handlers.ts @@ -1,5 +1,6 @@ import type { messagesList as messagesListDefinition, + messagesRead as messagesReadDefinition, messagesSend as messagesSendDefinition, } from "@moltzap/protocol/message"; import type { ParamsOf } from "@moltzap/protocol/rpc"; @@ -36,6 +37,21 @@ const handleMessageList = Effect.fn("messages.list")(function* ( }); }); +const handleMessageRead = Effect.fn("messages.read")(function* ( + params: ParamsOf, + ctx: AgentContext, +) { + const messageService = yield* MessageServiceTag; + return yield* messageService.read({ + conversationId: params.conversationId, + requesterAgentId: ctx.agentId, + ...(params.checkpoint === undefined + ? {} + : { checkpoint: params.checkpoint }), + ...(params.cursor === undefined ? {} : { cursor: params.cursor }), + }); +}); + // ── @effect/rpc handler bodies ─────────────────────────────────────── // // Requirement middleware gates each frame before these bodies run. The bodies @@ -68,3 +84,14 @@ export const messagesList: ServerHandler = const ctx = yield* agentArm; return yield* handleMessageList(params, ctx); }); + +/** + * Provides the checkpointed messages read runtime value. + * @param params Request payload to process. + * @returns The messages read result. + */ +export const messagesRead: ServerHandler = + Effect.fn("messagesRead")(function* (params) { + const ctx = yield* agentArm; + return yield* handleMessageRead(params, ctx); + }); diff --git a/packages/server/src/message/message.service.ts b/packages/server/src/message/message.service.ts index 1ff5088cc..5ae28f083 100644 --- a/packages/server/src/message/message.service.ts +++ b/packages/server/src/message/message.service.ts @@ -1,12 +1,18 @@ import { + READ_PLANE_PAGE_SIZE, type Db, - nextSnowflakeId, type MessageRow, catchSqlErrorAsDefect, + decodeConversationCheckpoint, + decodeConversationReadCursor, + encodeConversationCheckpoint, + encodeConversationReadCursor, + nextSnowflakeId, takeFirstOption, takeFirstOrFail, } from "#db"; import { + type ConversationCheckpoint, type Message, type MessageParts, type Part, @@ -22,6 +28,8 @@ import type { ConnectionId } from "@moltzap/protocol/socket"; import { DEFAULT_PAGE_LIMIT, type ForbiddenError, + InvalidParamsError, + type ListCursor, MAX_PAGE_LIMIT, } from "@moltzap/protocol/rpc"; import { type Cause, Effect, Option, Schema } from "effect"; @@ -50,6 +58,12 @@ function textPartsMetadata(parts: readonly Part[]): { const decodeMessageId = Schema.decodeUnknownSync(MessageIdSchema); +// PostgreSQL adapters may materialize BIGINT as either a decimal string or a +// safe integer. Opaque read positions use one canonical decimal representation. +function storedSequenceString(value: unknown): string { + return typeof value === "string" ? value : String(value); +} + interface SendInsertResult { readonly message: Message; readonly parts: MessageParts; @@ -71,6 +85,24 @@ interface SendCommitInput { readonly senderAgentId: AgentId; } +interface ReadMessagesInput { + readonly conversationId: ConversationId; + readonly requesterAgentId: AgentId; + readonly checkpoint?: ConversationCheckpoint; + readonly cursor?: ListCursor; +} + +interface ReadMessagesResult { + readonly messages: Message[]; + readonly checkpoint: ConversationCheckpoint; + readonly nextCursor?: ListCursor; +} + +interface ReadWindow { + readonly afterSeq: string; + readonly throughSeq: string; +} + /** Existence projection of the conversation a send targets. */ interface SendConversationRow { readonly id: ConversationId; @@ -320,12 +352,121 @@ export class MessageService { limit, }); const messages = yield* this.messageRowsToMessages(rows); + messages.reverse(); return { messages }; }.bind(this), ), ); } + read( + input: ReadMessagesInput, + ): Effect.Effect { + return catchSqlErrorAsDefect( + Effect.gen( + function* (this: MessageService) { + // Participation is checked before parsing either opaque token. An + // inaccessible conversation therefore reveals nothing about token + // validity or the conversation's stored position. + yield* this.conversations.assertConversationParticipant( + input.conversationId, + input.requesterAgentId, + ); + const window = yield* this.resolveReadWindow(input); + const rows = yield* this.readMessageRows({ + conversationId: input.conversationId, + ...window, + }); + const hasMore = rows.length > READ_PLANE_PAGE_SIZE; + const pageRows = hasMore ? rows.slice(0, READ_PLANE_PAGE_SIZE) : rows; + const messages = yield* this.messageRowsToMessages(pageRows); + const checkpoint = encodeConversationCheckpoint({ + conversationId: input.conversationId, + throughSeq: window.throughSeq, + }); + const last = pageRows.at(-1); + const nextCursor = + hasMore && last !== undefined + ? encodeConversationReadCursor({ + conversationId: input.conversationId, + throughSeq: window.throughSeq, + afterSeq: storedSequenceString(last.seq), + }) + : undefined; + return { + messages, + checkpoint, + ...(nextCursor === undefined ? {} : { nextCursor }), + }; + }.bind(this), + ), + ); + } + + private resolveReadWindow( + input: ReadMessagesInput, + ): Effect.Effect { + return Effect.gen( + function* (this: MessageService) { + if (input.checkpoint !== undefined && input.cursor !== undefined) { + return yield* new InvalidParamsError({ + message: "checkpoint and cursor cannot be used together", + }); + } + if (input.cursor !== undefined) { + return yield* decodeConversationReadCursor( + input.cursor, + input.conversationId, + ); + } + const priorThroughSeq = + input.checkpoint === undefined + ? "0" + : (yield* decodeConversationCheckpoint( + input.checkpoint, + input.conversationId, + )).throughSeq; + const currentMaxSeq = yield* this.currentMaxVisibleSeq( + input.conversationId, + ); + return { + afterSeq: priorThroughSeq, + throughSeq: + BigInt(priorThroughSeq) >= BigInt(currentMaxSeq) + ? priorThroughSeq + : currentMaxSeq, + }; + }.bind(this), + ); + } + + private currentMaxVisibleSeq( + conversationId: ConversationId, + ): Effect.Effect { + return this.db + .selectFrom("messages") + .select((eb) => eb.fn.max("seq").as("maxSeq")) + .where("conversation_id", "=", conversationId) + .where("is_deleted", "=", false) + .pipe(Effect.map((rows) => storedSequenceString(rows[0]?.maxSeq ?? 0))); + } + + private readMessageRows(args: { + readonly conversationId: ConversationId; + readonly afterSeq: string; + readonly throughSeq: string; + }): Effect.Effect { + return this.db + .selectFrom("messages") + .selectAll() + .where("conversation_id", "=", args.conversationId) + .where("is_deleted", "=", false) + .where("seq", ">", args.afterSeq) + .where("seq", "<=", args.throughSeq) + .orderBy("seq", "asc") + .limit(READ_PLANE_PAGE_SIZE + 1); + } + private visibleMessageRows(args: { readonly conversationId: ConversationId; readonly limit: number; @@ -359,7 +500,6 @@ export class MessageService { const parts = yield* decodeMessageParts(row.parts); messages.push(this.mapMessage(row, parts)); } - messages.reverse(); return messages; }.bind(this), ); diff --git a/packages/server/src/moltzap/handler-catalog.ts b/packages/server/src/moltzap/handler-catalog.ts index 51c9c37eb..35ad4718e 100644 --- a/packages/server/src/moltzap/handler-catalog.ts +++ b/packages/server/src/moltzap/handler-catalog.ts @@ -16,11 +16,12 @@ * handler body. */ import { connectAgent } from "#network"; -import { agentsList } from "#identity/agents"; -import { messagesSend, messagesList } from "#message/handlers"; +import { agentsList, agentsSearch } from "#identity/agents"; +import { messagesSend, messagesList, messagesRead } from "#message/handlers"; import { agentConversationCreate, conversationList, + conversationSearch, } from "#conversation/handlers"; import type { ServerHandlers } from "@moltzap/protocol/socket/catalog"; @@ -31,8 +32,11 @@ import type { ServerHandlers } from "@moltzap/protocol/socket/catalog"; export const serverHandlers: ServerHandlers = { "agent/network/connect": connectAgent, "agent/identity/agents/list": agentsList, + "agent/identity/agents/search": agentsSearch, "agent/message/send": messagesSend, "agent/message/list": messagesList, + "agent/message/read": messagesRead, "agent/conversation/list": conversationList, + "agent/conversation/search": conversationSearch, "agent/conversation/create": agentConversationCreate, } as const;