From 41e835a2629eca64a46103ef49c0fb5014b26e62 Mon Sep 17 00:00:00 2001 From: Tapan Chugh Date: Tue, 4 Aug 2026 01:48:45 -0700 Subject: [PATCH] feat(harness): project client-owned turns --- docs/modules/client/src.mdx | 33 +- packages/client/src/MODULE.md | 33 +- .../service/core/moltzapd.integration.test.ts | 23 +- .../client/src/channel-core-enrichment.ts | 207 ++++++++-- packages/client/src/channel-core.ts | 2 + packages/client/src/harness-client.test.ts | 367 ++++++++++++++++-- packages/client/src/harness-client.ts | 175 ++++++++- .../src/harness-context-projection.test.ts | 144 ++++++- .../client/src/harness-context-projection.ts | 60 ++- packages/client/src/harness-mcp-wire.ts | 58 ++- packages/client/src/harness/client-runtime.ts | 49 ++- packages/client/src/harness/index.ts | 14 +- packages/client/src/harness/runtime.test.ts | 21 + packages/client/src/harness/runtime.ts | 54 ++- .../src/harness/turn-projection.test.ts | 239 ++++++++++++ packages/client/src/moltzapd.ts | 42 +- 16 files changed, 1338 insertions(+), 183 deletions(-) create mode 100644 packages/client/src/harness/turn-projection.test.ts diff --git a/docs/modules/client/src.mdx b/docs/modules/client/src.mdx index c79314411..7cd803be3 100644 --- a/docs/modules/client/src.mdx +++ b/docs/modules/client/src.mdx @@ -13,18 +13,23 @@ Public barrel for the MoltZap client package. ## Public surface -### [`acquireHarnessClient`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/harness-client.ts#L41) +### [`acquireHarnessClient`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/harness-client.ts#L170) _Function_ ```ts export const acquireHarnessClient = ( options: HarnessClientOptions, -): Effect.Effect +): Effect.Effect< + HarnessClientService, + Error, + Scope.Scope | KeyValueStore.KeyValueStore +> ``` Acquires one turn-ready harness connection and receive stream for the -lifetime of the enclosing scope. The private adapter owns MCP translation. +lifetime of the enclosing scope. The supplied KeyValueStore is local to the +active agent and holds only stable presentation checkpoints. **Returns:** The scoped adapter-facing service value. @@ -71,7 +76,7 @@ export interface ConversationMeta { Describes conversation meta. -### [`HarnessClient`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/harness-client.ts#L23) +### [`HarnessClient`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/harness-client.ts#L43) _Class_ @@ -84,7 +89,7 @@ export class HarnessClient extends Context.Tag("@moltzap/client/HarnessClient")< Effect service tag consumed by runtime adapters. -### [`HarnessClientOptions`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/harness-client.ts#L29) +### [`HarnessClientOptions`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/harness-client.ts#L49) _Interface_ @@ -97,12 +102,14 @@ export interface HarnessClientOptions { Inputs needed to connect one scoped harness client. -### [`HarnessClientService`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/harness-client.ts#L17) +### [`HarnessClientService`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/harness-client.ts#L35) _Interface_ ```ts export interface HarnessClientService { + /** Active identity used by adapters when rendering self-authored context. */ + readonly agentId: AgentId; /** The sole receive stream owned by this scoped client. */ readonly turns: Stream.Stream; } @@ -110,31 +117,27 @@ export interface HarnessClientService { Adapter-facing capability backed only by the daemon's loopback MCP surface. -### [`HarnessTurn`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/harness-client.ts#L7) +### [`HarnessTurn`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/harness-client.ts#L29) _Interface_ ```ts -export interface HarnessTurn { - /** Existing conversation associated with every message in this turn. */ - readonly conversationId: ConversationId; - /** Existing protocol messages in their daemon-provided order. */ - readonly messages: readonly [Message, ...Message[]]; +export interface HarnessTurn extends EnrichedInboundMessage { /** Sends model output through the MCP reply route captured by this turn. */ readonly reply: (payload: string) => Effect.Effect; } ``` -One reply-capable batch emitted by the local harness daemon. +Existing adapter presentation with reply authority bound to its live turn. -### [`makeHarnessClientLayer`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/harness-client.ts#L52) +### [`makeHarnessClientLayer`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/harness-client.ts#L197) _Function_ ```ts export const makeHarnessClientLayer = ( options: HarnessClientOptions, -): Layer.Layer +): Layer.Layer ``` Builds the scoped runtime-adapter layer for one daemon endpoint. diff --git a/packages/client/src/MODULE.md b/packages/client/src/MODULE.md index 0b946da00..9c62c8309 100644 --- a/packages/client/src/MODULE.md +++ b/packages/client/src/MODULE.md @@ -8,18 +8,23 @@ Public barrel for the MoltZap client package. ## Public surface -### [`acquireHarnessClient`](./harness-client.ts#L41) +### [`acquireHarnessClient`](./harness-client.ts#L170) _Function_ ```ts export const acquireHarnessClient = ( options: HarnessClientOptions, -): Effect.Effect +): Effect.Effect< + HarnessClientService, + Error, + Scope.Scope | KeyValueStore.KeyValueStore +> ``` Acquires one turn-ready harness connection and receive stream for the -lifetime of the enclosing scope. The private adapter owns MCP translation. +lifetime of the enclosing scope. The supplied KeyValueStore is local to the +active agent and holds only stable presentation checkpoints. **Returns:** The scoped adapter-facing service value. @@ -66,7 +71,7 @@ export interface ConversationMeta { Describes conversation meta. -### [`HarnessClient`](./harness-client.ts#L23) +### [`HarnessClient`](./harness-client.ts#L43) _Class_ @@ -79,7 +84,7 @@ export class HarnessClient extends Context.Tag("@moltzap/client/HarnessClient")< Effect service tag consumed by runtime adapters. -### [`HarnessClientOptions`](./harness-client.ts#L29) +### [`HarnessClientOptions`](./harness-client.ts#L49) _Interface_ @@ -92,12 +97,14 @@ export interface HarnessClientOptions { Inputs needed to connect one scoped harness client. -### [`HarnessClientService`](./harness-client.ts#L17) +### [`HarnessClientService`](./harness-client.ts#L35) _Interface_ ```ts export interface HarnessClientService { + /** Active identity used by adapters when rendering self-authored context. */ + readonly agentId: AgentId; /** The sole receive stream owned by this scoped client. */ readonly turns: Stream.Stream; } @@ -105,31 +112,27 @@ export interface HarnessClientService { Adapter-facing capability backed only by the daemon's loopback MCP surface. -### [`HarnessTurn`](./harness-client.ts#L7) +### [`HarnessTurn`](./harness-client.ts#L29) _Interface_ ```ts -export interface HarnessTurn { - /** Existing conversation associated with every message in this turn. */ - readonly conversationId: ConversationId; - /** Existing protocol messages in their daemon-provided order. */ - readonly messages: readonly [Message, ...Message[]]; +export interface HarnessTurn extends EnrichedInboundMessage { /** Sends model output through the MCP reply route captured by this turn. */ readonly reply: (payload: string) => Effect.Effect; } ``` -One reply-capable batch emitted by the local harness daemon. +Existing adapter presentation with reply authority bound to its live turn. -### [`makeHarnessClientLayer`](./harness-client.ts#L52) +### [`makeHarnessClientLayer`](./harness-client.ts#L197) _Function_ ```ts export const makeHarnessClientLayer = ( options: HarnessClientOptions, -): Layer.Layer +): Layer.Layer ``` Builds the scoped runtime-adapter layer for one daemon endpoint. 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 3200256f9..b169071b2 100644 --- a/packages/client/src/__tests__/service/core/moltzapd.integration.test.ts +++ b/packages/client/src/__tests__/service/core/moltzapd.integration.test.ts @@ -1,4 +1,5 @@ import { FileSystem, HttpClient } from "@effect/platform"; +import * as KeyValueStore from "@effect/platform/KeyValueStore"; import { NodeContext, NodeHttpClient } from "@effect/platform-node"; import { Client, @@ -253,18 +254,19 @@ const expectNoUnixSocket = (socketPath: string) => const expectHarnessTurn = ( turn: HarnessTurn, + owner: RegisteredAgent, peer: RegisteredAgent, conversationId: ConversationId, ): void => { expect(turn.conversationId).toBe(conversationId); - expect(turn.messages).toHaveLength(1); - const inbound = turn.messages[0]; - if (inbound === undefined) { - throw new Error("expected one inbound harness message"); - } - expect(inbound.conversationId).toBe(conversationId); - expect(inbound.senderId).toBe(peer.agentId); - expect(H.textContent(inbound)).toBe(PEER_MESSAGE); + expect(turn.sender).toEqual({ id: peer.agentId, name: peer.name }); + expect(turn.text).toBe(PEER_MESSAGE); + expect(turn.isFromMe).toBe(false); + expect(turn.conversationMeta?.type).toBe("dm"); + expect(new Set(turn.conversationMeta?.participants)).toEqual( + new Set([`agent:${peer.agentId}`, `agent:${owner.agentId}`]), + ); + expect(turn).not.toHaveProperty("messages"); }; const expectPeerReply = ( @@ -379,7 +381,7 @@ const runMcpMessageRoundTrip = ({ }); const turn = yield* Fiber.join(turnFiber); - expectHarnessTurn(turn, peer, conversationId); + expectHarnessTurn(turn, owner, peer, conversationId); yield* expectNoUnixSocket(socketPath); yield* turn.reply(HARNESS_REPLY); @@ -409,7 +411,8 @@ function runHarnessRoundTrip(owner: RegisteredAgent, peer: RegisteredAgent) { }); const harness = yield* acquireHarnessClient({ url: harnessUrl(server).href, - }); + }).pipe(Effect.provide(KeyValueStore.layerMemory)); + expect(harness.agentId).toBe(owner.agentId); const mcp = yield* acquireMcpClient(harnessUrl(server)); yield* expectNoUnixSocket(socketPath); diff --git a/packages/client/src/channel-core-enrichment.ts b/packages/client/src/channel-core-enrichment.ts index 92ab054cd..6813f49a9 100644 --- a/packages/client/src/channel-core-enrichment.ts +++ b/packages/client/src/channel-core-enrichment.ts @@ -1,4 +1,9 @@ import { Effect } from "effect"; +import type { + Conversation, + ConversationId, +} from "@moltzap/protocol/conversation"; +import type { AgentCard, AgentId } from "@moltzap/protocol/identity"; import type { Message } from "@moltzap/protocol/message"; import type { ChannelService, @@ -6,6 +11,7 @@ import type { EnrichedConversationMeta, EnrichedInboundMessage, } from "./channel-core.js"; +import { renderPart } from "./message-rendering.js"; type CoalescedMessage = NonNullable< EnrichedInboundMessage["coalescedMessages"] @@ -17,12 +23,21 @@ interface EnrichmentContext { readonly commitContext?: () => void; } -interface EnrichedMessageInput { - readonly service: ChannelService; +interface ResolvedInboundMessage { readonly message: Message; readonly senderName: string; - readonly coalesced: readonly CoalescedMessage[]; - readonly context: EnrichmentContext; +} + +type ResolvedInboundMessages = readonly [ + ResolvedInboundMessage, + ...ResolvedInboundMessage[], +]; + +interface EnrichedInboundProjectionInput { + readonly messages: ResolvedInboundMessages; + readonly ownAgentId?: string; + readonly conversationMeta?: EnrichedConversationMeta; + readonly contextBlocks: ContextBlocks; } function isMessageList( @@ -87,9 +102,9 @@ function resolveSenderName( } function coalescedMessageFrom( - message: Message, - senderName: string, + resolved: ResolvedInboundMessage, ): CoalescedMessage { + const { message, senderName } = resolved; return { id: message.id, sender: { @@ -101,29 +116,26 @@ function coalescedMessageFrom( }; } -function buildCoalescedMessages( +function resolveInboundMessages( service: ChannelService, messages: readonly Message[], primarySenderName: string, -): Effect.Effect { +): Effect.Effect { return Effect.gen(function* () { const primaryMessage = /* Safe because the surrounding invariant establishes this asserted shape. */ messages[0]!; - const coalesced = [coalescedMessageFrom(primaryMessage, primarySenderName)]; + const remaining: ResolvedInboundMessage[] = []; for (const message of messages.slice(1)) { const senderName = yield* resolveSenderName(service, message.senderId); - coalesced.push(coalescedMessageFrom(message, senderName)); + remaining.push({ message, senderName }); } - return coalesced; + return [ + { message: primaryMessage, senderName: primarySenderName }, + ...remaining, + ]; }); } -function isFromOwnAgent(service: ChannelService, message: Message): boolean { - return ( - service.ownAgentId !== undefined && message.senderId === service.ownAgentId - ); -} - function collectContextBlocks( service: ChannelService, conversationId: string, @@ -158,13 +170,18 @@ function collectContextBlocks( }; } -function buildEnrichedInboundMessage({ - service, - message, - senderName, - coalesced, - context, -}: EnrichedMessageInput): EnrichedInboundMessage { +/** + * Projects a materialized nonempty message batch into the channel-owned + * enriched shape without reading or advancing presentation state. + * @param input Resolved messages, identity, metadata, and context blocks. + * @returns The enriched inbound message shared by channel and harness turns. + */ +function projectEnrichedInboundMessage( + input: EnrichedInboundProjectionInput, +): EnrichedInboundMessage { + const primary = input.messages[0]; + const { message, senderName } = primary; + const coalesced = input.messages.map(coalescedMessageFrom); return { id: message.id, conversationId: message.conversationId, @@ -173,16 +190,133 @@ function buildEnrichedInboundMessage({ name: senderName, }, text: formatCoalescedText(coalesced), - isFromMe: isFromOwnAgent(service, message), + isFromMe: + input.ownAgentId !== undefined && message.senderId === input.ownAgentId, createdAt: message.createdAt, - contextBlocks: context.contextBlocks, - ...(context.conversationMeta - ? { conversationMeta: context.conversationMeta } + contextBlocks: input.contextBlocks, + ...(input.conversationMeta + ? { conversationMeta: input.conversationMeta } : {}), ...(coalesced.length > 1 ? { coalescedMessages: coalesced } : {}), }; } +type CrossConvMessage = NonNullable< + ContextBlocks["crossConversationMessages"] +>[number]; + +type ConversationWithParticipants = Conversation & { + readonly participants: readonly AgentId[]; +}; + +interface HarnessTurnProjectionInput { + readonly context: { + readonly conversations: readonly ConversationWithParticipants[]; + readonly currentMessages: readonly [Message, ...Message[]]; + readonly crossConversationMessages: readonly Message[]; + }; + readonly agents: readonly AgentCard[]; + readonly ownAgentId: AgentId; +} + +const agentNamesFrom = ( + agents: readonly AgentCard[], +): ReadonlyMap => + new Map(agents.map((agent) => [agent.id, agent.name] as const)); + +const senderNameFrom = ( + names: ReadonlyMap, + senderId: AgentId, +): string => names.get(senderId) ?? senderId; + +const harnessConversationMetaFrom = ( + conversation?: ConversationWithParticipants, +): EnrichedConversationMeta | undefined => + conversation === undefined + ? undefined + : { + type: conversation.participants.length > 2 ? "group" : "dm", + ...(conversation.name === undefined ? {} : { name: conversation.name }), + participants: conversation.participants.map( + (participant) => `agent:${participant}`, + ), + }; + +const renderMessageText = (message: Message): string => + message.parts.map(renderPart).join(" "); + +const crossConversationMessagesFrom = ( + messages: readonly Message[], + conversations: ReadonlyMap, + agentNames: ReadonlyMap, +): readonly CrossConvMessage[] => + messages.map((message) => { + const conversationName = conversations.get(message.conversationId)?.name; + return { + conversationId: message.conversationId, + ...(conversationName === undefined ? {} : { conversationName }), + senderName: senderNameFrom(agentNames, message.senderId), + senderId: message.senderId, + text: renderMessageText(message), + timestamp: message.createdAt, + }; + }); + +/** + * Projects MCP-reconstructed context into the channel-owned enriched shape. + * @param input Reconstructed messages plus resolved identity information. + * @param input.context Current and cross-conversation message context. + * @param input.agents Agent cards used for presentation names. + * @param input.ownAgentId Active identity used to mark self-authored content. + * @returns The enriched inbound message exposed by a harness turn. + */ +export const projectHarnessTurn = ({ + context, + agents, + ownAgentId, +}: HarnessTurnProjectionInput): EnrichedInboundMessage => { + const agentNames = agentNamesFrom(agents); + const conversations = new Map( + context.conversations.map( + (conversation) => [conversation.id, conversation] as const, + ), + ); + const [primary, ...remaining] = context.currentMessages; + const resolvedMessages: ResolvedInboundMessages = [ + { + message: primary, + senderName: senderNameFrom(agentNames, primary.senderId), + }, + ...remaining.map((message) => ({ + message, + senderName: senderNameFrom(agentNames, message.senderId), + })), + ]; + const conversationMeta = harnessConversationMetaFrom( + conversations.get(primary.conversationId), + ); + const crossConversationMessages = crossConversationMessagesFrom( + context.crossConversationMessages, + conversations, + agentNames, + ); + const contextBlocks: ContextBlocks = { + ...(conversationMeta?.type === "group" + ? { groupMetadata: conversationMeta } + : {}), + ...(crossConversationMessages.length === 0 + ? {} + : { crossConversationMessages: [...crossConversationMessages] }), + }; + + return projectEnrichedInboundMessage({ + messages: resolvedMessages, + ownAgentId, + ...(conversationMeta === undefined ? {} : { conversationMeta }), + contextBlocks, + }); +}; + /** * Executes the enrich channel message operation. * @param service Value supplied to the operation. @@ -201,7 +335,7 @@ export function enrichChannelMessage( const message = /* Safe because the surrounding invariant establishes this asserted shape. */ messages[0]!; const senderName = yield* resolveSenderName(service, message.senderId); - const coalesced = yield* buildCoalescedMessages( + const resolvedMessages = yield* resolveInboundMessages( service, messages, senderName, @@ -216,12 +350,15 @@ export function enrichChannelMessage( ); return { - enriched: buildEnrichedInboundMessage({ - service, - message, - senderName, - coalesced, - context, + enriched: projectEnrichedInboundMessage({ + messages: resolvedMessages, + ...(service.ownAgentId === undefined + ? {} + : { ownAgentId: service.ownAgentId }), + ...(context.conversationMeta === undefined + ? {} + : { conversationMeta: context.conversationMeta }), + contextBlocks: context.contextBlocks, }), ...(context.commitContext ? { commitContext: context.commitContext } diff --git a/packages/client/src/channel-core.ts b/packages/client/src/channel-core.ts index 6521edb57..ad641e93d 100644 --- a/packages/client/src/channel-core.ts +++ b/packages/client/src/channel-core.ts @@ -15,6 +15,8 @@ import type { ServiceRpcError, } from "./service.js"; import { enrichChannelMessage } from "./channel-core-enrichment.js"; +/** Projects MCP-reconstructed context through the channel-owned presentation. */ +export { projectHarnessTurn } from "./channel-core-enrichment.js"; /** Describes enriched sender. */ export interface EnrichedSender { diff --git a/packages/client/src/harness-client.test.ts b/packages/client/src/harness-client.test.ts index bfa924766..50914ee2d 100644 --- a/packages/client/src/harness-client.test.ts +++ b/packages/client/src/harness-client.test.ts @@ -1,4 +1,5 @@ /* eslint-disable agent-code-guard/async-keyword -- This loopback contract test hosts the Promise-native official MCP SDK. */ +import * as KeyValueStore from "@effect/platform/KeyValueStore"; import { Client } from "@modelcontextprotocol/client"; import { createMcpHandler, @@ -7,10 +8,37 @@ import { type Implementation, type JsonSchemaType, } from "@modelcontextprotocol/server"; -import { Chunk, Effect, Exit, Fiber, Option, Scope, Stream } from "effect"; +import { + Chunk, + Effect, + Exit, + Fiber, + JSONSchema, + Layer, + Option, + Schema, + Scope, + Stream, +} from "effect"; import { describe, expect, it, vi } from "vitest"; -import type { Message } from "@moltzap/protocol/message"; -import { agentId, conversationId, messageId } from "@moltzap/protocol/testing"; +import { conversationSearch } from "@moltzap/protocol/conversation"; +import { agentsSearch, type AgentCard } from "@moltzap/protocol/identity"; +import { + conversationCheckpoint, + messagesRead, + type Message, +} from "@moltzap/protocol/message"; +import type { + ParamsOf, + ResultOf, + RpcDefinitionAny, +} from "@moltzap/protocol/rpc"; +import { + agentId, + agentName, + conversationId, + messageId, +} from "@moltzap/protocol/testing"; import { acquireHarnessClient, HarnessClient, @@ -21,18 +49,26 @@ import { acquireHarnessMcpHttpServer } from "./harness-mcp-server.js"; import { decodeHarnessReplyRoute, HARNESS_EVENTS_EXTENSION, + HARNESS_READ_CONVERSATION_TOOL, HARNESS_REPLY_TOOL, + HARNESS_SEARCH_AGENTS_TOOL, + HARNESS_SEARCH_CONVERSATIONS_TOOL, + HARNESS_STATUS_TOOL, + harnessSearchConversationsResultJsonSchema, harnessReplyInputJsonSchema, harnessReplyResultJsonSchema, + type ConversationWithParticipants, type HarnessReplyInput, type HarnessReplyResult, type HarnessReplyRoute, + type HarnessSearchConversationsResult, type HarnessTurnEvent, } from "./harness/index.js"; import { makeHarnessMcpSubscriptionHandler, type HarnessMcpSubscriptionHandler, } from "./harness-mcp-subscription.js"; +import { statusCommandRpc } from "./local-daemon-rpc.js"; const SERVER_IMPLEMENTATION = { name: "harness-client-test", @@ -45,17 +81,49 @@ const SECOND_CONVERSATION = conversationId( "00000000-0000-4000-8000-000000000002", ); const SENDER_ID = agentId("00000000-0000-4000-8000-000000000003"); +const SELF_ID = agentId("00000000-0000-4000-8000-000000000006"); +const THIRD_ID = agentId("00000000-0000-4000-8000-000000000007"); +const CHECKPOINT = Schema.decodeSync(conversationCheckpoint)( + "harness-client-checkpoint", +); +const CREATED_AT = "2026-08-03T12:00:00.000Z"; + +const AGENTS = [ + { id: SELF_ID, name: agentName("self-agent"), status: "active" }, + { id: SENDER_ID, name: agentName("peer-agent"), status: "active" }, + { id: THIRD_ID, name: agentName("third-agent"), status: "active" }, +] satisfies readonly AgentCard[]; + +const CONVERSATIONS = [ + { + id: FIRST_CONVERSATION, + name: "first dm", + createdBy: SELF_ID, + createdAt: CREATED_AT, + updatedAt: CREATED_AT, + participants: [SELF_ID, SENDER_ID], + }, + { + id: SECOND_CONVERSATION, + name: "second group", + createdBy: SELF_ID, + createdAt: CREATED_AT, + updatedAt: CREATED_AT, + participants: [SELF_ID, SENDER_ID, THIRD_ID], + }, +] satisfies readonly ConversationWithParticipants[]; const message = ( id: string, conversation: typeof FIRST_CONVERSATION, text: string, + senderId = SENDER_ID, ): Message => ({ id: messageId(id), conversationId: conversation, - senderId: SENDER_ID, + senderId, parts: [{ type: "text", text }], - createdAt: "2026-08-03T12:00:00.000Z", + createdAt: CREATED_AT, }); const firstEvent = { @@ -65,6 +133,11 @@ const firstEvent = { FIRST_CONVERSATION, "first", ), + message( + "00000000-0000-4000-8000-000000000008", + FIRST_CONVERSATION, + "queued", + ), ], } satisfies HarnessTurnEvent; const secondEvent = { @@ -77,6 +150,13 @@ const secondEvent = { ], } satisfies HarnessTurnEvent; +const selfAuthoredHistory = message( + "00000000-0000-4000-8000-000000000009", + FIRST_CONVERSATION, + "self-authored history", + SELF_ID, +); + interface ObservedReply { readonly input: HarnessReplyInput; readonly route: HarnessReplyRoute; @@ -90,6 +170,113 @@ const replyResultSchema = fromJsonSchema( /* Safe because Effect and MCP expose the same JSON Schema wire shape with different array mutability declarations. */ harnessReplyResultJsonSchema as JsonSchemaType, ); +const searchConversationsResultSchema = + fromJsonSchema( + /* Safe because Effect and MCP expose the same JSON Schema wire shape with different array mutability declarations. */ + harnessSearchConversationsResultJsonSchema as JsonSchemaType, + ); + +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, + ); + +type DescriptorHandler = ( + input: ParamsOf, +) => ResultOf; + +const registerDescriptorTool = ( + server: McpServer, + name: string, + definition: D, + handler: DescriptorHandler, +): void => { + server.registerTool( + name, + { + inputSchema: effectSchemaToMcpSchema>( + definition.paramsSchema, + ), + outputSchema: effectSchemaToMcpSchema>( + definition.resultSchema, + ), + }, + (input) => { + const result = handler(input); + return Effect.runPromise( + Effect.succeed({ + content: [{ type: "text" as const, text: JSON.stringify(result) }], + structuredContent: result, + }), + ); + }, + ); +}; + +const registerStatusTool = (server: McpServer): void => { + const status = { + agentId: SELF_ID, + connected: true, + conversations: CONVERSATIONS.length, + }; + server.registerTool( + HARNESS_STATUS_TOOL, + { + inputSchema: effectSchemaToMcpSchema(statusCommandRpc.payloadSchema), + outputSchema: effectSchemaToMcpSchema(statusCommandRpc.successSchema), + }, + () => + Effect.runPromise( + Effect.succeed({ + content: [{ type: "text" as const, text: JSON.stringify(status) }], + structuredContent: status, + }), + ), + ); +}; + +const registerSearchConversationsTool = (server: McpServer): void => { + server.registerTool( + HARNESS_SEARCH_CONVERSATIONS_TOOL, + { + inputSchema: effectSchemaToMcpSchema(conversationSearch.paramsSchema), + outputSchema: searchConversationsResultSchema, + }, + () => { + const result = { conversations: [...CONVERSATIONS] }; + return Effect.runPromise( + Effect.succeed({ + content: [{ type: "text" as const, text: JSON.stringify(result) }], + structuredContent: result, + }), + ); + }, + ); +}; + +const registerReadPlaneTools = (server: McpServer): void => { + registerStatusTool(server); + registerDescriptorTool( + server, + HARNESS_SEARCH_AGENTS_TOOL, + agentsSearch, + () => ({ agents: [...AGENTS] }), + ); + registerSearchConversationsTool(server); + registerDescriptorTool( + server, + HARNESS_READ_CONVERSATION_TOOL, + messagesRead, + ({ conversationId }) => ({ + messages: + conversationId === FIRST_CONVERSATION ? [selfAuthoredHistory] : [], + checkpoint: CHECKPOINT, + }), + ); +}; const makeHarnessHandler = ( observed: ObservedReply[], @@ -102,6 +289,7 @@ const makeHarnessHandler = ( ? { extensions: { [HARNESS_EVENTS_EXTENSION]: {} } } : {}, }); + registerReadPlaneTools(server); server.registerTool( HARNESS_REPLY_TOOL, { @@ -163,6 +351,7 @@ const useHarness = ( ): Effect.Effect => Effect.gen(function* () { const harness = yield* HarnessClient; + expect(harness.agentId).toBe(SELF_ID); const receive = yield* harness.turns.pipe( Stream.take(2), Stream.runCollect, @@ -180,6 +369,82 @@ const useHarness = ( return turns; }); +const expectFirstTurn = (turn: HarnessTurn): void => { + expect(turn).toMatchObject({ + id: firstEvent.messages[0].id, + conversationId: FIRST_CONVERSATION, + sender: { id: SENDER_ID, name: "peer-agent" }, + text: `first\n\n[queued message from peer-agent at ${CREATED_AT}]\nqueued`, + isFromMe: false, + createdAt: CREATED_AT, + conversationMeta: { + type: "dm", + name: "first dm", + participants: [`agent:${SELF_ID}`, `agent:${SENDER_ID}`], + }, + contextBlocks: {}, + coalescedMessages: [ + { + id: firstEvent.messages[0].id, + sender: { id: SENDER_ID, name: "peer-agent" }, + text: "first", + createdAt: CREATED_AT, + }, + { + id: firstEvent.messages[1].id, + sender: { id: SENDER_ID, name: "peer-agent" }, + text: "queued", + createdAt: CREATED_AT, + }, + ], + }); + expect(turn).not.toHaveProperty("messages"); +}; + +const expectSecondTurn = (turn: HarnessTurn): void => { + expect(turn).toMatchObject({ + conversationId: SECOND_CONVERSATION, + conversationMeta: { + type: "group", + name: "second group", + participants: [ + `agent:${SELF_ID}`, + `agent:${SENDER_ID}`, + `agent:${THIRD_ID}`, + ], + }, + contextBlocks: { + groupMetadata: { + type: "group", + name: "second group", + }, + crossConversationMessages: [ + { + conversationId: FIRST_CONVERSATION, + conversationName: "first dm", + senderName: "self-agent", + senderId: SELF_ID, + text: "self-authored history", + timestamp: CREATED_AT, + }, + ], + }, + }); +}; + +const expectBoundReplies = (observed: readonly ObservedReply[]): void => { + expect(observed).toEqual([ + { + input: { payload: "first reply" }, + route: { conversationId: FIRST_CONVERSATION }, + }, + { + input: { payload: "second reply" }, + route: { conversationId: FIRST_CONVERSATION }, + }, + ]); +}; + const preservesBoundConversation = async () => { const observed: ObservedReply[] = []; const handler = makeHarnessHandler(observed); @@ -191,7 +456,7 @@ const preservesBoundConversation = async () => { Effect.provide( makeHarnessClientLayer({ url: running.url.href, - }), + }).pipe(Layer.provide(KeyValueStore.layerMemory)), ), ), ); @@ -199,18 +464,13 @@ const preservesBoundConversation = async () => { FIRST_CONVERSATION, SECOND_CONVERSATION, ]); - expect(turns[0]?.messages).toEqual(firstEvent.messages); - - expect(observed).toEqual([ - { - input: { payload: "first reply" }, - route: { conversationId: FIRST_CONVERSATION }, - }, - { - input: { payload: "second reply" }, - route: { conversationId: FIRST_CONVERSATION }, - }, - ]); + const [firstTurn, secondTurn] = turns; + if (firstTurn === undefined || secondTurn === undefined) { + throw new Error("expected two harness turns"); + } + expectFirstTurn(firstTurn); + expectSecondTurn(secondTurn); + expectBoundReplies(observed); } finally { await Effect.runPromise(Scope.close(running.scope, Exit.void)); } @@ -221,7 +481,9 @@ const rejectsMissingServerExtension = async () => { try { await expect( Effect.runPromise( - Effect.scoped(acquireHarnessClient({ url: running.url.href })), + Effect.scoped(acquireHarnessClient({ url: running.url.href })).pipe( + Effect.provide(KeyValueStore.layerMemory), + ), ), ).rejects.toThrow(HARNESS_EVENTS_EXTENSION); } finally { @@ -243,43 +505,62 @@ const rejectsUnexpectedTurnFields = async () => { expect(handler.publish(eventWithExtraField)).toBe(true); return yield* Fiber.join(next); }), - ); + ).pipe(Effect.provide(KeyValueStore.layerMemory)); await expect(Effect.runPromise(nextTurn)).rejects.toBeDefined(); } finally { await Effect.runPromise(Scope.close(running.scope, Exit.void)); } }; +interface ReplyCallObservation { + count: number; + signal?: AbortSignal; +} + +const originalClientCallTool = Reflect.get(Client.prototype, "callTool"); + +const makeReplyCallImplementation = ( + observation: ReplyCallObservation, +): Client["callTool"] => + function (this: Client, params, options) { + if (params.name !== HARNESS_REPLY_TOOL) { + return originalClientCallTool.call(this, params, options); + } + observation.count += 1; + observation.signal = options?.signal; + return new Promise((resolve, reject) => { + if (observation.signal === undefined) { + resolve({ content: [], isError: true }); + return; + } + observation.signal.addEventListener( + "abort", + () => { + reject(new Error("reply request aborted")); + }, + { once: true }, + ); + }); + }; + const abortsReplyCallWhenInterrupted = async () => { const handler = makeHarnessHandler([]); const running = await startHarnessServer(handler); const clientScope = Effect.runSync(Scope.make()); - let observedSignal: AbortSignal | undefined; - const callTool = vi - .spyOn(Client.prototype, "callTool") - .mockImplementation((params, options) => { - expect(params.name).toBe(HARNESS_REPLY_TOOL); - observedSignal = options?.signal; - return new Promise((resolve, reject) => { - if (observedSignal === undefined) { - resolve({ content: [], isError: true }); - return; - } - observedSignal?.addEventListener( - "abort", - () => { - reject(new Error("reply request aborted")); - }, - { once: true }, - ); - }); - }); + const observation: ReplyCallObservation = { + count: 0, + }; + let callTool: { readonly mockRestore: () => void } | undefined; try { const harness = await Effect.runPromise( acquireHarnessClient({ url: running.url.href }).pipe( Scope.extend(clientScope), + Effect.provide(KeyValueStore.layerMemory), ), ); + callTool = vi + .spyOn(Client.prototype, "callTool") + .mockImplementation(makeReplyCallImplementation(observation)); const received = Effect.runPromise(harness.turns.pipe(Stream.runHead)); expect(handler.publish(firstEvent)).toBe(true); const turn = Option.getOrThrowWith( @@ -288,12 +569,12 @@ const abortsReplyCallWhenInterrupted = async () => { ); const reply = Effect.runFork(turn.reply("cancel me")); await vi.waitFor(() => { - expect(callTool).toHaveBeenCalledOnce(); + expect(observation.count).toBe(1); }); await Effect.runPromise(Fiber.interrupt(reply)); - expect(observedSignal?.aborted).toBe(true); + expect(observation.signal?.aborted).toBe(true); } finally { - callTool.mockRestore(); + callTool?.mockRestore(); await Effect.runPromise(Scope.close(clientScope, Exit.void)); await Effect.runPromise(Scope.close(running.scope, Exit.void)); } diff --git a/packages/client/src/harness-client.ts b/packages/client/src/harness-client.ts index 81e3d068d..737f55857 100644 --- a/packages/client/src/harness-client.ts +++ b/packages/client/src/harness-client.ts @@ -1,20 +1,40 @@ -import { Context, Layer, type Effect, type Scope, type Stream } from "effect"; -import type { ConversationId } from "@moltzap/protocol/conversation"; -import type { Message } from "@moltzap/protocol/message"; -import { acquireHarnessClientInternal } from "./harness/index.js"; - -/** One reply-capable batch emitted by the local harness daemon. */ -export interface HarnessTurn { - /** Existing conversation associated with every message in this turn. */ - readonly conversationId: ConversationId; - /** Existing protocol messages in their daemon-provided order. */ - readonly messages: readonly [Message, ...Message[]]; +import * as KeyValueStore from "@effect/platform/KeyValueStore"; +import { Context, Effect, Layer, Schema, Stream, type Scope } from "effect"; +import type { conversationSearch } from "@moltzap/protocol/conversation"; +import { agentsSearch, type AgentId } from "@moltzap/protocol/identity"; +import { messagesRead } from "@moltzap/protocol/message"; +import type { + ParamsOf, + ResultOf, + RpcDefinitionAny, +} from "@moltzap/protocol/rpc"; +import { + projectHarnessTurn, + type EnrichedInboundMessage, +} from "./channel-core.js"; +import { reconstructHarnessContext } from "./harness-context-projection.js"; +import { + acquireHarnessClientInternal, + HARNESS_READ_CONVERSATION_TOOL, + HARNESS_SEARCH_AGENTS_TOOL, + HARNESS_SEARCH_CONVERSATIONS_TOOL, + HARNESS_STATUS_TOOL, + decodeHarnessSearchConversationsResult, + type HarnessClientInternalService, + type HarnessTurnInternal, +} from "./harness/index.js"; +import { statusCommandRpc } from "./local-daemon-rpc.js"; + +/** Existing adapter presentation with reply authority bound to its live turn. */ +export interface HarnessTurn extends EnrichedInboundMessage { /** Sends model output through the MCP reply route captured by this turn. */ readonly reply: (payload: string) => Effect.Effect; } /** Adapter-facing capability backed only by the daemon's loopback MCP surface. */ export interface HarnessClientService { + /** Active identity used by adapters when rendering self-authored context. */ + readonly agentId: AgentId; /** The sole receive stream owned by this scoped client. */ readonly turns: Stream.Stream; } @@ -31,17 +51,142 @@ export interface HarnessClientOptions { readonly url: string; } +const strictDecodeOptions = { onExcessProperty: "error" } as const; + +const asError = (cause: unknown): Error => + cause instanceof Error ? cause : new Error(String(cause)); + +const callDescriptorTool = ( + session: HarnessClientInternalService, + toolName: string, + definition: D, + params: ParamsOf, +): Effect.Effect, Error> => + session + .callTool( + toolName, + /* Safe because every RPC params Schema used here is a closed Struct and MCP tool arguments are JSON objects. */ params as Readonly< + Record + >, + ) + .pipe( + Effect.flatMap((result) => + Schema.decodeUnknown(definition.resultSchema)( + result, + strictDecodeOptions, + ).pipe( + Effect.map( + (decoded) => + /* Safe because ResultOf derives from this exact descriptor's resultSchema; RpcDefinitionAny erases only the runtime schema property's generic surface. */ decoded as ResultOf, + ), + ), + ), + Effect.mapError(asError), + ); + +const readActiveAgentId = ( + session: HarnessClientInternalService, +): Effect.Effect => + session.callTool(HARNESS_STATUS_TOOL, {}).pipe( + Effect.flatMap((result) => + Schema.decodeUnknown(statusCommandRpc.successSchema)( + result, + strictDecodeOptions, + ), + ), + Effect.mapError(asError), + Effect.flatMap((status) => { + if (status.agentId === undefined) { + // eslint-disable-next-line agent-code-guard/effect-error-erasure -- A daemon with no active identity is rejected at the public client boundary, whose existing error contract is Error. + return Effect.fail( + new Error("Harness MCP status has no active AgentId"), + ); + } + return Effect.succeed(status.agentId); + }), + ); + +const searchConversations = ( + session: HarnessClientInternalService, + params: ParamsOf, +) => + session + .callTool(HARNESS_SEARCH_CONVERSATIONS_TOOL, params) + .pipe( + Effect.flatMap(decodeHarnessSearchConversationsResult), + Effect.mapError(asError), + ); + +const contextReadPlane = (session: HarnessClientInternalService) => ({ + searchAgents: (params: ParamsOf) => + callDescriptorTool( + session, + HARNESS_SEARCH_AGENTS_TOOL, + agentsSearch, + params, + ), + searchConversations: (params: ParamsOf) => + searchConversations(session, params), + readConversation: (params: ParamsOf) => + callDescriptorTool( + session, + HARNESS_READ_CONVERSATION_TOOL, + messagesRead, + params, + ), +}); + +const projectTurn = ( + session: HarnessClientInternalService, + checkpointStore: KeyValueStore.KeyValueStore, + agentId: AgentId, + turn: HarnessTurnInternal, +): Effect.Effect => + reconstructHarnessContext(contextReadPlane(session), turn.event).pipe( + Effect.provideService(KeyValueStore.KeyValueStore, checkpointStore), + Effect.map((context) => ({ + ...projectHarnessTurn({ + agents: context.agents, + context: { + conversations: context.conversations, + crossConversationMessages: context.crossConversationMessages, + currentMessages: context.currentMessages, + }, + ownAgentId: agentId, + }), + reply: turn.reply, + })), + Effect.mapError(asError), + ); + /** * Acquires one turn-ready harness connection and receive stream for the - * lifetime of the enclosing scope. The private adapter owns MCP translation. + * lifetime of the enclosing scope. The supplied KeyValueStore is local to the + * active agent and holds only stable presentation checkpoints. * * @param options Fixed loopback MCP endpoint. * @returns The scoped adapter-facing service value. */ export const acquireHarnessClient = ( options: HarnessClientOptions, -): Effect.Effect => - acquireHarnessClientInternal(options); +): Effect.Effect< + HarnessClientService, + Error, + Scope.Scope | KeyValueStore.KeyValueStore +> => + Effect.gen(function* () { + const checkpointStore = yield* KeyValueStore.KeyValueStore; + const session = yield* acquireHarnessClientInternal(options); + const agentId = yield* readActiveAgentId(session); + return { + agentId, + turns: session.turns.pipe( + Stream.mapEffect((turn) => + projectTurn(session, checkpointStore, agentId, turn), + ), + ), + }; + }).pipe(Effect.withSpan("acquireHarnessClient.presentation")); /** * Builds the scoped runtime-adapter layer for one daemon endpoint. @@ -51,5 +196,5 @@ export const acquireHarnessClient = ( */ export const makeHarnessClientLayer = ( options: HarnessClientOptions, -): Layer.Layer => +): Layer.Layer => Layer.scoped(HarnessClient, acquireHarnessClient(options)); diff --git a/packages/client/src/harness-context-projection.test.ts b/packages/client/src/harness-context-projection.test.ts index 3ca8f7470..c46893b55 100644 --- a/packages/client/src/harness-context-projection.test.ts +++ b/packages/client/src/harness-context-projection.test.ts @@ -1,18 +1,19 @@ import * as KeyValueStore from "@effect/platform/KeyValueStore"; import { Effect, Option, Schema } from "effect"; import { describe, expect, it, vi } from "vitest"; -import { - conversationSearch, - type Conversation, - type ConversationId, -} from "@moltzap/protocol/conversation"; +import type { ConversationId } from "@moltzap/protocol/conversation"; +import { agentsSearch } from "@moltzap/protocol/identity"; import { messagesRead, type Message } from "@moltzap/protocol/message"; import { agentId, conversationId, messageId } from "@moltzap/protocol/testing"; import { reconstructHarnessContext, type ContextProjectionReadPlane, } from "./harness-context-projection.js"; -import type { HarnessTurnEvent } from "./harness/runtime.js"; +import { + decodeHarnessSearchConversationsResult, + type ConversationWithParticipants, + type HarnessTurnEvent, +} from "./harness/runtime.js"; import { NonAdvancingCursorError } from "./pagination.js"; const TARGET = conversationId("00000000-0000-4000-8000-000000000001"); @@ -20,13 +21,15 @@ const SOURCE = conversationId("00000000-0000-4000-8000-000000000002"); const OTHER_TARGET = conversationId("00000000-0000-4000-8000-000000000003"); const SENDER = agentId("00000000-0000-4000-8000-000000000004"); const CREATED_BY = agentId("00000000-0000-4000-8000-000000000005"); +const AGENT_PAGE_CURSOR = "agent-page-2"; const CONVERSATION_PAGE_CURSOR = "conversation-page-2"; const SOURCE_PAGE_CURSOR = "source-page-2"; const SOURCE_CHECKPOINT = "source-checkpoint"; -const conversation = (id: ConversationId): Conversation => ({ +const conversation = (id: ConversationId): ConversationWithParticipants => ({ id, createdBy: CREATED_BY, + participants: [CREATED_BY, SENDER], createdAt: "2026-08-03T12:00:00.000Z", updatedAt: "2026-08-03T12:00:00.000Z", }); @@ -43,8 +46,10 @@ const message = ( createdAt, }); -const decodeSearchPage = Schema.decodeUnknownSync( - conversationSearch.resultSchema, +const decodeSearchPage = (value: unknown) => + Effect.runSync(decodeHarnessSearchConversationsResult(value)); +const decodeAgentSearchPage = Schema.decodeUnknownSync( + agentsSearch.resultSchema, ); const decodeReadPage = Schema.decodeUnknownSync(messagesRead.resultSchema); const decodeStoredCheckpointMap = Schema.decodeUnknown( @@ -82,6 +87,17 @@ const LATE_CROSS_MESSAGE = message( "2026-08-03T12:00:02.000Z", ); +const FIRST_AGENT_PAGE = decodeAgentSearchPage({ + agents: [{ id: SENDER, name: "sender-agent", status: "active" }], + nextCursor: AGENT_PAGE_CURSOR, +}); +const SECOND_AGENT_PAGE = decodeAgentSearchPage({ + agents: [{ id: CREATED_BY, name: "creator-agent", status: "active" }], +}); + +type AgentSearchParams = Parameters< + ContextProjectionReadPlane["searchAgents"] +>[0]; type SearchParams = Parameters< ContextProjectionReadPlane["searchConversations"] >[0]; @@ -99,6 +115,14 @@ const paginatedSearch = (params: SearchParams) => : decodeSearchPage({ conversations: [conversation(SOURCE)] }), ); +const paginatedAgentSearch = (params: AgentSearchParams) => + Effect.succeed( + params.cursor === undefined ? FIRST_AGENT_PAGE : SECOND_AGENT_PAGE, + ); + +const emptyAgentSearch: ContextProjectionReadPlane["searchAgents"] = + () => Effect.succeed(decodeAgentSearchPage({ agents: [] })); + const paginatedRead = (params: ReadParams) => { if (params.conversationId === TARGET) { return Effect.dieMessage("current-conversation history must not be read"); @@ -118,17 +142,19 @@ const paginatedRead = (params: ReadParams) => { }; const makePaginatedReadPlane = () => { + const searchAgents = vi.fn(paginatedAgentSearch); const searchConversations = vi.fn(paginatedSearch); const readConversation = vi.fn(paginatedRead); const readPlane = { + searchAgents, searchConversations, readConversation, } satisfies ContextProjectionReadPlane; - return { readPlane, readConversation, searchConversations }; + return { readPlane, readConversation, searchAgents, searchConversations }; }; const reconstructsPaginatedContext = () => { - const { readPlane, readConversation, searchConversations } = + const { readPlane, readConversation, searchAgents, searchConversations } = makePaginatedReadPlane(); return Effect.gen(function* () { @@ -139,6 +165,10 @@ const reconstructsPaginatedContext = () => { const persisted = yield* storedCheckpoints(TARGET); expect(context.conversationId).toBe(TARGET); + expect(context.agents).toEqual([ + ...FIRST_AGENT_PAGE.agents, + ...SECOND_AGENT_PAGE.agents, + ]); expect(context.currentMessages).toEqual([TARGET_MESSAGE]); expect(context.crossConversationMessages).toEqual([ EARLY_CROSS_MESSAGE, @@ -149,6 +179,10 @@ const reconstructsPaginatedContext = () => { }); expect(Object.values(persisted)).not.toContain(CONVERSATION_PAGE_CURSOR); expect(Object.values(persisted)).not.toContain(SOURCE_PAGE_CURSOR); + expect(searchAgents).toHaveBeenNthCalledWith(1, {}); + expect(searchAgents).toHaveBeenNthCalledWith(2, { + cursor: AGENT_PAGE_CURSOR, + }); expect(searchConversations).toHaveBeenNthCalledWith(1, {}); expect(searchConversations).toHaveBeenNthCalledWith(2, { cursor: CONVERSATION_PAGE_CURSOR, @@ -203,6 +237,7 @@ const makeIndependentReadPlane = () => { ), ); return { + searchAgents: emptyAgentSearch, searchConversations, readConversation, } satisfies ContextProjectionReadPlane; @@ -254,6 +289,7 @@ const makeRestartReadPlane = (firstCrossMessage: Message) => { ), ); const readPlane = { + searchAgents: emptyAgentSearch, searchConversations, readConversation, } satisfies ContextProjectionReadPlane; @@ -303,6 +339,7 @@ const reusesOnlyStableCheckpointsForLaterObservation = () => { }; const cyclicSearchReadPlane = { + searchAgents: () => Effect.dieMessage("conversation search must finish"), searchConversations: () => Effect.succeed( decodeSearchPage({ @@ -343,6 +380,7 @@ const makeFailingReadPlane = () => { : Effect.fail(READ_FAILURE), ); return { + searchAgents: () => Effect.dieMessage("history reads must finish"), searchConversations, readConversation, } satisfies ContextProjectionReadPlane; @@ -371,9 +409,85 @@ const preservesPriorCheckpointWhenReadFails = () => { }).pipe(Effect.provide(KeyValueStore.layerMemory)); }; -// @agent-code-guard/regression-only: these examples pin target/source persistence and keep temporary page cursors out of durable client state. +const cyclicAgentSearch = vi.fn(() => + Effect.succeed( + decodeAgentSearchPage({ + agents: [], + nextCursor: AGENT_PAGE_CURSOR, + }), + ), +); + +const cyclicAgentReadPlane = { + searchAgents: cyclicAgentSearch, + searchConversations: () => + Effect.succeed(decodeSearchPage({ conversations: [conversation(TARGET)] })), + readConversation: () => Effect.dieMessage("target history is not read"), +} satisfies ContextProjectionReadPlane; + +const rejectsCyclicAgentSearchWithoutCheckpointing = () => + Effect.gen(function* () { + const error = yield* reconstructHarnessContext( + cyclicAgentReadPlane, + liveEvent(TARGET_MESSAGE), + ).pipe(Effect.flip); + const store = yield* KeyValueStore.KeyValueStore; + + expect(error).toBeInstanceOf(NonAdvancingCursorError); + expect(error).toMatchObject({ method: agentsSearch.name }); + expect(cyclicAgentSearch).toHaveBeenNthCalledWith(1, {}); + expect(cyclicAgentSearch).toHaveBeenNthCalledWith(2, { + cursor: AGENT_PAGE_CURSOR, + }); + expect(Option.isNone(yield* store.get(TARGET))).toBe(true); + }).pipe(Effect.provide(KeyValueStore.layerMemory)); + +const AGENT_SEARCH_FAILURE = "agent search failed"; + +const makeFailingAgentSearchReadPlane = () => { + const readConversation = vi.fn(() => + Effect.succeed( + decodeReadPage({ + messages: [], + checkpoint: SOURCE_CHECKPOINT, + }), + ), + ); + return { + searchAgents: () => Effect.fail(AGENT_SEARCH_FAILURE), + searchConversations: searchTargetAndSource, + readConversation, + } satisfies ContextProjectionReadPlane; +}; + +const preservesPriorCheckpointWhenAgentSearchFails = () => { + const readPlane = makeFailingAgentSearchReadPlane(); + return Effect.gen(function* () { + const store = yield* KeyValueStore.KeyValueStore; + yield* store.set( + TARGET, + JSON.stringify({ [SOURCE]: PRIOR_SOURCE_CHECKPOINT }), + ); + + const error = yield* reconstructHarnessContext( + readPlane, + liveEvent(TARGET_MESSAGE), + ).pipe(Effect.flip); + + expect(error).toBe(AGENT_SEARCH_FAILURE); + expect(readPlane.readConversation).toHaveBeenCalledWith({ + conversationId: SOURCE, + checkpoint: PRIOR_SOURCE_CHECKPOINT, + }); + expect(yield* storedCheckpoints(TARGET)).toEqual({ + [SOURCE]: PRIOR_SOURCE_CHECKPOINT, + }); + }).pipe(Effect.provide(KeyValueStore.layerMemory)); +}; + +// @agent-code-guard/regression-only: these examples pin target/source persistence and keep temporary directory and history cursors out of durable client state. describe("Harness context reconstruction", () => { - it("drains current and cross-conversation pages before persisting checkpoints", () => + it("drains agent, conversation, and history pages before persisting checkpoints", () => Effect.runPromise(reconstructsPaginatedContext())); it("keeps checkpoint maps independent for each target conversation", () => Effect.runPromise(keepsTargetSourcePositionsIndependent())); @@ -381,6 +495,10 @@ describe("Harness context reconstruction", () => { Effect.runPromise(reusesOnlyStableCheckpointsForLaterObservation())); it("rejects a cyclic search cursor without storing a checkpoint", () => Effect.runPromise(rejectsCyclicSearchWithoutCheckpointing())); + it("rejects a cyclic agent cursor without storing a checkpoint", () => + Effect.runPromise(rejectsCyclicAgentSearchWithoutCheckpointing())); it("leaves a prior checkpoint unchanged when a page read fails", () => Effect.runPromise(preservesPriorCheckpointWhenReadFails())); + it("leaves a prior checkpoint unchanged when agent search fails", () => + Effect.runPromise(preservesPriorCheckpointWhenAgentSearchFails())); }); diff --git a/packages/client/src/harness-context-projection.ts b/packages/client/src/harness-context-projection.ts index a8c71c97c..6bff60c3d 100644 --- a/packages/client/src/harness-context-projection.ts +++ b/packages/client/src/harness-context-projection.ts @@ -4,9 +4,9 @@ import { Effect, Option, Schema, type ParseResult } from "effect"; import { conversationId, conversationSearch, - type Conversation, type ConversationId, } from "@moltzap/protocol/conversation"; +import { agentsSearch, type AgentCard } from "@moltzap/protocol/identity"; import { conversationCheckpoint, messagesRead, @@ -14,7 +14,11 @@ import { type Message, } from "@moltzap/protocol/message"; import type { ParamsOf, ResultOf } from "@moltzap/protocol/rpc"; -import type { HarnessTurnEvent } from "./harness/runtime.js"; +import type { + ConversationWithParticipants, + HarnessSearchConversationsResult, + HarnessTurnEvent, +} from "./harness/runtime.js"; import { NonAdvancingCursorError } from "./pagination.js"; const checkpointMapSchema = Schema.Record({ @@ -23,8 +27,11 @@ const checkpointMapSchema = Schema.Record({ }); type CheckpointMap = Schema.Schema.Type; +type AgentSearchCursor = NonNullable< + ResultOf["nextCursor"] +>; type ConversationSearchCursor = NonNullable< - ResultOf["nextCursor"] + HarnessSearchConversationsResult["nextCursor"] >; type ConversationReadCursor = NonNullable< ResultOf["nextCursor"] @@ -32,9 +39,12 @@ type ConversationReadCursor = NonNullable< /** Package-private MCP read capabilities used for presentation recovery. */ export interface ContextProjectionReadPlane { + readonly searchAgents: ( + params: ParamsOf, + ) => Effect.Effect, E>; readonly searchConversations: ( params: ParamsOf, - ) => Effect.Effect, E>; + ) => Effect.Effect; readonly readConversation: ( params: ParamsOf, ) => Effect.Effect, E>; @@ -43,8 +53,9 @@ export interface ContextProjectionReadPlane { /** Raw production context reconstructed for one live replyable observation. */ export interface ReconstructedHarnessContext { readonly conversationId: ConversationId; - readonly conversations: readonly Conversation[]; - readonly currentMessages: readonly Message[]; + readonly agents: readonly AgentCard[]; + readonly conversations: readonly ConversationWithParticipants[]; + readonly currentMessages: HarnessTurnEvent["messages"]; readonly crossConversationMessages: readonly Message[]; } @@ -72,11 +83,35 @@ const acceptNextCursor = ( return Effect.succeed(nextCursor); }; +const drainAgentSearch = ( + readPlane: ContextProjectionReadPlane, +): Effect.Effect => + Effect.gen(function* () { + const agents: AgentCard[] = []; + const seenCursors = new Set(); + let cursor: AgentSearchCursor | undefined; + do { + const page = yield* readPlane.searchAgents( + cursor === undefined ? {} : { cursor }, + ); + agents.push(...page.agents); + cursor = yield* acceptNextCursor( + seenCursors, + agentsSearch.name, + page.nextCursor, + ); + } while (cursor !== undefined); + return agents; + }).pipe(Effect.withSpan("HarnessContextProjection.searchAgents")); + const drainConversationSearch = ( readPlane: ContextProjectionReadPlane, -): Effect.Effect => +): Effect.Effect< + readonly ConversationWithParticipants[], + E | NonAdvancingCursorError +> => Effect.gen(function* () { - const conversations: Conversation[] = []; + const conversations: ConversationWithParticipants[] = []; const seenCursors = new Set(); let cursor: ConversationSearchCursor | undefined; do { @@ -141,7 +176,7 @@ const drainConversationRead = ( * @returns Conversation identifiers eligible for cross-context recovery. */ const sourceConversationIds = ( - conversations: readonly Conversation[], + conversations: readonly ConversationWithParticipants[], targetConversationId: ConversationId, ): readonly ConversationId[] => conversations @@ -173,12 +208,14 @@ const checkpointMapAfter = ( const contextFrom = ( event: HarnessTurnEvent, - conversations: readonly Conversation[], + agents: readonly AgentCard[], + conversations: readonly ConversationWithParticipants[], deltas: readonly ConversationDelta[], ): ReconstructedHarnessContext => { const conversationId = event.messages[0].conversationId; return { conversationId, + agents, conversations, currentMessages: event.messages, crossConversationMessages: chronologicalCrossMessages( @@ -228,7 +265,8 @@ export const reconstructHarnessContext = ( ), { concurrency: 1 }, ); - const context = contextFrom(event, conversations, deltas); + const agents = yield* drainAgentSearch(readPlane); + const context = contextFrom(event, agents, conversations, deltas); yield* checkpoints.set( targetConversationId, checkpointMapAfter(priorCheckpoints, deltas), diff --git a/packages/client/src/harness-mcp-wire.ts b/packages/client/src/harness-mcp-wire.ts index b947615e1..854bf8455 100644 --- a/packages/client/src/harness-mcp-wire.ts +++ b/packages/client/src/harness-mcp-wire.ts @@ -23,11 +23,17 @@ import type { import { decodeHarnessReplyRoute, HARNESS_EVENTS_EXTENSION, + HARNESS_READ_CONVERSATION_TOOL, HARNESS_REPLY_TOOL, + HARNESS_SEARCH_AGENTS_TOOL, + HARNESS_SEARCH_CONVERSATIONS_TOOL, + HARNESS_STATUS_TOOL, + harnessSearchConversationsResultJsonSchema, harnessReplyInputJsonSchema, harnessReplyResultJsonSchema, type HarnessReplyInput, type HarnessReplyResult, + type HarnessSearchConversationsResult, type HarnessTurnEvent, } from "./harness/index.js"; import { @@ -40,11 +46,6 @@ import { type LocalDaemonHandlers, } 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; type StatusHandler = LocalDaemonHandlers[typeof localDaemonCommands.status]; @@ -55,13 +56,16 @@ type ReplyHandler = ( type DescriptorHandler = ( payload: ParamsOf, ) => Effect.Effect, unknown>; +type SearchConversationsHandler = ( + payload: ParamsOf, +) => Effect.Effect; interface HarnessMcpHandlerOptions { readonly implementation: Implementation; readonly readConversation: DescriptorHandler; readonly reply: ReplyHandler; readonly searchAgents: DescriptorHandler; - readonly searchConversations: DescriptorHandler; + readonly searchConversations: SearchConversationsHandler; readonly status: StatusHandler; } @@ -92,6 +96,10 @@ const replyInputSchema = fromJsonSchema( const replyOutputSchema = fromJsonSchema( /* Safe because Effect and MCP expose the same JSON Schema wire shape with different array mutability declarations. */ harnessReplyResultJsonSchema as JsonSchemaType, ); +const searchConversationsOutputSchema = + fromJsonSchema( + /* Safe because Effect and MCP expose the same JSON Schema wire shape with different array mutability declarations. */ harnessSearchConversationsResultJsonSchema as JsonSchemaType, + ); const registerDescriptorTool = ( server: McpServer, @@ -131,12 +139,37 @@ const registerDescriptorTool = ( ); }; +const registerSearchConversationsTool = ( + server: McpServer, + handler: SearchConversationsHandler, +): void => { + server.registerTool( + HARNESS_SEARCH_CONVERSATIONS_TOOL, + { + inputSchema: effectSchemaToMcpSchema>( + conversationSearch.paramsSchema, + ), + outputSchema: searchConversationsOutputSchema, + }, + (payload, context) => + Effect.runPromise( + handler(payload).pipe( + Effect.map((result) => ({ + content: [{ type: "text" as const, text: JSON.stringify(result) }], + structuredContent: result, + })), + ), + { signal: context.mcpReq.signal }, + ), + ); +}; + const makeRegistrationServer = (implementation: Implementation): McpServer => new McpServer(implementation); const registerStatusTool = (server: McpServer, status: StatusHandler): void => { server.registerTool( - STATUS_TOOL_NAME, + HARNESS_STATUS_TOOL, { inputSchema: statusInputSchema, outputSchema: statusOutputSchema, @@ -201,19 +234,14 @@ const makeActiveServer = ({ registerStatusTool(server, status); registerDescriptorTool( server, - SEARCH_AGENTS_TOOL_NAME, + HARNESS_SEARCH_AGENTS_TOOL, agentsSearch, searchAgents, ); + registerSearchConversationsTool(server, searchConversations); registerDescriptorTool( server, - SEARCH_CONVERSATIONS_TOOL_NAME, - conversationSearch, - searchConversations, - ); - registerDescriptorTool( - server, - READ_CONVERSATION_TOOL_NAME, + HARNESS_READ_CONVERSATION_TOOL, messagesRead, readConversation, ); diff --git a/packages/client/src/harness/client-runtime.ts b/packages/client/src/harness/client-runtime.ts index 4e70d407e..2d96be114 100644 --- a/packages/client/src/harness/client-runtime.ts +++ b/packages/client/src/harness/client-runtime.ts @@ -25,13 +25,24 @@ interface HarnessClientInternalOptions { readonly url: string; } -interface HarnessTurnInternal { - readonly conversationId: ConversationId; - readonly messages: HarnessTurnEvent["messages"]; +/** + * Decoded live observation and its private reply authority. + * @internal + */ +export interface HarnessTurnInternal { + readonly event: HarnessTurnEvent; readonly reply: (payload: string) => Effect.Effect; } -interface HarnessClientInternalService { +/** + * Package-owned MCP session consumed by the public domain projection. + * @internal + */ +export interface HarnessClientInternalService { + readonly callTool: ( + name: string, + input: Readonly>, + ) => Effect.Effect; readonly turns: Stream.Stream; } @@ -55,6 +66,31 @@ const asError = (cause: unknown): Error => const closeQuietly = (close: () => Promise): Effect.Effect => Effect.tryPromise({ try: close, catch: asError }).pipe(Effect.ignore); +const callStructuredTool = ( + client: Client, + name: string, + input: Readonly>, +): Effect.Effect => + Effect.tryPromise({ + try: (signal) => + client.callTool({ name, arguments: { ...input } }, { signal }), + catch: asError, + }).pipe( + Effect.flatMap((result) => { + if (result.isError === true) { + // eslint-disable-next-line agent-code-guard/effect-error-erasure -- The private MCP adapter normalizes untyped tool failures to the public client's existing Error contract. + return Effect.fail(new Error(`Harness MCP tool ${name} failed`)); + } + if (result.structuredContent === undefined) { + // eslint-disable-next-line agent-code-guard/effect-error-erasure -- Missing structured content is an incompatible MCP response at the public client's existing Error boundary. + return Effect.fail( + new Error(`Harness MCP tool ${name} returned no structured content`), + ); + } + return Effect.succeed(result.structuredContent); + }), + ); + const turnPayload = (params: unknown): unknown => { if (typeof params !== "object" || params === null || Array.isArray(params)) { return params; @@ -97,8 +133,7 @@ const makeTurn = ( ): HarnessTurnInternal => { const originatingConversationId = harnessTurnConversationId(event); return { - conversationId: originatingConversationId, - messages: event.messages, + event, reply: (payload) => callReply(client, originatingConversationId, payload), }; }; @@ -208,6 +243,8 @@ export const acquireHarnessClientInternal = ( yield* Effect.forkScoped(observeSubscription(subscription, queue)); return { + callTool: (name: string, input: Readonly>) => + callStructuredTool(client, name, input), turns: Stream.fromQueue(queue).pipe(Stream.flattenTake), }; }).pipe(Effect.withSpan("acquireHarnessClient")); diff --git a/packages/client/src/harness/index.ts b/packages/client/src/harness/index.ts index 379dccee2..d761f66f7 100644 --- a/packages/client/src/harness/index.ts +++ b/packages/client/src/harness/index.ts @@ -1,16 +1,28 @@ /** @internal */ -export { acquireHarnessClientInternal } from "./client-runtime.js"; +export { + acquireHarnessClientInternal, + type HarnessClientInternalService, + type HarnessTurnInternal, +} from "./client-runtime.js"; /** @internal */ export { decodeHarnessReplyRoute, + decodeHarnessSearchConversationsResult, HARNESS_EVENTS_EXTENSION, + HARNESS_READ_CONVERSATION_TOOL, HARNESS_REPLY_TOOL, + HARNESS_SEARCH_AGENTS_TOOL, + HARNESS_SEARCH_CONVERSATIONS_TOOL, + HARNESS_STATUS_TOOL, HARNESS_TURN_READY_FILTER, HARNESS_TURN_READY_NOTIFICATION, + harnessSearchConversationsResultJsonSchema, harnessReplyInputJsonSchema, harnessReplyResultJsonSchema, + type ConversationWithParticipants, type HarnessReplyInput, type HarnessReplyResult, type HarnessReplyRoute, + type HarnessSearchConversationsResult, type HarnessTurnEvent, } from "./runtime.js"; diff --git a/packages/client/src/harness/runtime.test.ts b/packages/client/src/harness/runtime.test.ts index 8b243faa7..2d9755658 100644 --- a/packages/client/src/harness/runtime.test.ts +++ b/packages/client/src/harness/runtime.test.ts @@ -11,11 +11,13 @@ import { } from "@modelcontextprotocol/server"; import { Effect, Exit, Scope } from "effect"; import { describe, expect, it } from "vitest"; +import { conversationSearch } from "@moltzap/protocol/conversation"; import type { Message } from "@moltzap/protocol/message"; import { agentId, conversationId, messageId } from "@moltzap/protocol/testing"; import { acquireHarnessMcpHttpServer } from "../harness-mcp-server.js"; import { decodeHarnessReplyRoute, + decodeHarnessSearchConversationsResult, decodeHarnessTurnEvent, HARNESS_EVENTS_EXTENSION, HARNESS_REPLY_TOOL, @@ -53,6 +55,14 @@ const otherConversationMessage = { conversationId: conversationId("00000000-0000-4000-8000-000000000002"), } satisfies Message; +const conversationWithParticipants = { + id: CONVERSATION_ID, + createdBy: SENDER_ID, + participants: [SENDER_ID], + createdAt: firstMessage.createdAt, + updatedAt: secondMessage.createdAt, +}; + const replyInputJsonSchema = fromJsonSchema( /* Safe because Effect and MCP expose the same JSON Schema wire shape with different array mutability declarations. */ harnessReplyInputJsonSchema as JsonSchemaType, ); @@ -100,6 +110,14 @@ const keepsPrivateRoutingMetadataClosed = async () => { ).rejects.toBeDefined(); }; +const keepsConversationMembershipOnMcpOnly = () => { + const page = { conversations: [conversationWithParticipants] }; + expect(Effect.runSync(decodeHarnessSearchConversationsResult(page))).toEqual( + page, + ); + expect(conversationSearch.validateResult(page)).toBe(false); +}; + interface ObservedReply { arguments?: unknown; route?: HarnessReplyRoute; @@ -206,6 +224,9 @@ describe("Harness MCP runtime contract", () => { decodesProtocolMessageBatch()); it("keeps private routing metadata closed", () => keepsPrivateRoutingMetadataClosed()); + it("adds conversation membership only on the MCP projection", () => { + keepsConversationMembershipOnMcpOnly(); + }); it("preserves the private route through an official MCP client call", () => preservesPrivateRoute()); }); diff --git a/packages/client/src/harness/runtime.ts b/packages/client/src/harness/runtime.ts index 92e1e9c17..c9423586a 100644 --- a/packages/client/src/harness/runtime.ts +++ b/packages/client/src/harness/runtime.ts @@ -1,8 +1,11 @@ import { JSONSchema, Schema } from "effect"; import { conversationId, + conversationSchema, + conversationSearch, type ConversationId, } from "@moltzap/protocol/conversation"; +import { agentId } from "@moltzap/protocol/identity"; import { messageReceivedNotificationDefinition } from "@moltzap/protocol/message"; /** Harness MCP extension carrying the runtime event contract. */ @@ -18,9 +21,32 @@ export const HARNESS_TURN_READY_NOTIFICATION = /** Tool used for model output in the current conversation. */ export const HARNESS_REPLY_TOOL = "reply"; +/** Tool returning the active daemon identity and connection state. */ +export const HARNESS_STATUS_TOOL = "status"; + +/** Tool browsing or matching visible agent cards. */ +export const HARNESS_SEARCH_AGENTS_TOOL = "search_agents"; + +/** Tool browsing or matching visible conversations. */ +export const HARNESS_SEARCH_CONVERSATIONS_TOOL = "search_conversations"; + +/** Tool reading one checkpointed conversation history. */ +export const HARNESS_READ_CONVERSATION_TOOL = "read_conversation"; + const messageSchema = messageReceivedNotificationDefinition.paramsSchema.fields.message; +const conversationWithParticipantsSchema = Schema.Struct({ + ...conversationSchema().fields, + participants: Schema.Array(agentId), +}); + +/** MCP-local search result used to reconstruct endpoint presentation. */ +const harnessSearchConversationsResultSchema = Schema.Struct({ + ...conversationSearch.resultSchema.fields, + conversations: Schema.Array(conversationWithParticipantsSchema), +}); + /** One nonempty batch of protocol messages delivered as a model turn. */ const harnessTurnEventSchema = Schema.Struct({ messages: Schema.NonEmptyArray(messageSchema), @@ -51,6 +77,16 @@ export type HarnessTurnEvent = Schema.Schema.Type< typeof harnessTurnEventSchema >; +/** Conversation projection carried only between the daemon and HarnessClient. */ +export type ConversationWithParticipants = Schema.Schema.Type< + typeof conversationWithParticipantsSchema +>; + +/** Decoded MCP-local conversation search page. */ +export type HarnessSearchConversationsResult = Schema.Schema.Type< + typeof harnessSearchConversationsResultSchema +>; + /** Decoded reply input. */ export type HarnessReplyInput = Schema.Schema.Type< typeof harnessReplyInputSchema @@ -66,10 +102,18 @@ export type HarnessReplyRoute = Schema.Schema.Type< typeof harnessReplyRouteSchema >; +const strictDecodeOptions = { onExcessProperty: "error" } as const; const decodeTurnEvent = Schema.decodeUnknown(harnessTurnEventSchema); +const decodeSearchConversationsResult = Schema.decodeUnknown( + harnessSearchConversationsResultSchema, +); const decodeReplyRoute = Schema.decodeUnknown(harnessReplyRouteSchema); -const strictDecodeOptions = { onExcessProperty: "error" } as const; +/** JSON Schema advertised for the MCP-local conversation search result. */ +export const harnessSearchConversationsResultJsonSchema = JSONSchema.make( + harnessSearchConversationsResultSchema, + { target: "jsonSchema2020-12" }, +); /** JSON Schema advertised for the payload-only reply tool arguments. */ export const harnessReplyInputJsonSchema = JSONSchema.make( @@ -91,6 +135,14 @@ export const harnessReplyResultJsonSchema = JSONSchema.make( export const decodeHarnessTurnEvent = (value: unknown) => decodeTurnEvent(value, strictDecodeOptions); +/** + * Strictly decode the membership-bearing conversation page received over MCP. + * @param value Untrusted structured tool content. + * @returns The decoded MCP-local conversation page. + */ +export const decodeHarnessSearchConversationsResult = (value: unknown) => + decodeSearchConversationsResult(value, strictDecodeOptions); + /** * Build the private request metadata consumed by the production harness client. * @param originatingConversationId Conversation associated with the live turn. diff --git a/packages/client/src/harness/turn-projection.test.ts b/packages/client/src/harness/turn-projection.test.ts new file mode 100644 index 000000000..a9014d401 --- /dev/null +++ b/packages/client/src/harness/turn-projection.test.ts @@ -0,0 +1,239 @@ +import { describe, expect, it } from "vitest"; +import type { + Conversation, + ConversationId, +} from "@moltzap/protocol/conversation"; +import type { AgentCard, AgentId } from "@moltzap/protocol/identity"; +import type { Message } from "@moltzap/protocol/message"; +import { + agentId, + agentName, + conversationId, + messageId, +} from "@moltzap/protocol/testing"; +import { projectHarnessTurn } from "../channel-core.js"; + +type ConversationWithParticipants = Conversation & { + readonly participants: readonly AgentId[]; +}; + +const OWN = agentId("00000000-0000-4000-8000-000000000001"); +const ALICE = agentId("00000000-0000-4000-8000-000000000002"); +const BOB = agentId("00000000-0000-4000-8000-000000000003"); +const UNKNOWN = agentId("00000000-0000-4000-8000-000000000004"); +const TARGET = conversationId("00000000-0000-4000-8000-000000000005"); +const SOURCE = conversationId("00000000-0000-4000-8000-000000000006"); + +const agents: readonly AgentCard[] = [ + { id: ALICE, name: agentName("alice"), status: "active" }, + { id: BOB, name: agentName("bob"), status: "active" }, +]; + +const conversation = ( + id: ConversationId, + participants: readonly AgentId[], + name?: string, +): ConversationWithParticipants => ({ + id, + createdBy: OWN, + participants, + ...(name === undefined ? {} : { name }), + createdAt: "2026-08-04T12:00:00.000Z", + updatedAt: "2026-08-04T12:00:00.000Z", +}); + +interface MessageInput { + readonly id: string; + readonly conversationId: ConversationId; + readonly senderId: AgentId; + readonly parts: Message["parts"]; + readonly createdAt: string; +} + +const message = ({ + id, + conversationId, + senderId, + parts, + createdAt, +}: MessageInput): Message => ({ + id: messageId(id), + conversationId, + senderId, + parts, + createdAt, +}); + +interface MaterializedMessages { + readonly first: Message; + readonly queued: Message; + readonly cross: Message; +} + +const materializedMessages = (): MaterializedMessages => ({ + first: message({ + id: "00000000-0000-4000-8000-000000000007", + conversationId: TARGET, + senderId: ALICE, + parts: [ + { type: "text", text: "first" }, + { type: "image", url: "https://example.com/first.png" }, + { type: "text", text: "continued" }, + ], + createdAt: "2026-08-04T12:00:01.000Z", + }), + queued: message({ + id: "00000000-0000-4000-8000-000000000008", + conversationId: TARGET, + senderId: BOB, + parts: [ + { type: "text", text: "second" }, + { + type: "file", + url: "https://example.com/ignored.txt", + name: "ignored.txt", + }, + ], + createdAt: "2026-08-04T12:00:02.000Z", + }), + cross: message({ + id: "00000000-0000-4000-8000-000000000009", + conversationId: SOURCE, + senderId: UNKNOWN, + parts: [ + { type: "text", text: "context" }, + { + type: "file", + url: "https://example.com/report.pdf", + name: "report.pdf", + }, + { type: "image", url: "https://example.com/chart.png" }, + ], + createdAt: "2026-08-04T11:59:59.000Z", + }), +}); + +const projectMaterializedHarnessContext = ({ + first, + queued, + cross, +}: MaterializedMessages) => + projectHarnessTurn({ + ownAgentId: OWN, + agents, + context: { + currentMessages: [first, queued], + crossConversationMessages: [cross], + conversations: [ + conversation(TARGET, [OWN, ALICE, BOB], "builders"), + conversation(SOURCE, [OWN, UNKNOWN], "research"), + ], + }, + }); + +const groupMetadata = { + type: "group" as const, + name: "builders", + participants: [`agent:${OWN}`, `agent:${ALICE}`, `agent:${BOB}`], +}; + +const expectedCrossContext = (cross: Message) => ({ + groupMetadata, + crossConversationMessages: [ + { + conversationId: SOURCE, + conversationName: "research", + senderName: UNKNOWN, + senderId: UNKNOWN, + text: "context [file: report.pdf] [image]", + timestamp: cross.createdAt, + }, + ], +}); + +const expectedCoalescedMessages = (first: Message, queued: Message) => [ + { + id: first.id, + sender: { id: ALICE, name: "alice" }, + text: "first\ncontinued", + createdAt: first.createdAt, + }, + { + id: queued.id, + sender: { id: BOB, name: "bob" }, + text: "second", + createdAt: queued.createdAt, + }, +]; + +const expectMaterializedProjection = ( + projected: ReturnType, + { first, queued, cross }: MaterializedMessages, +) => { + expect(projected).toEqual({ + id: first.id, + conversationId: TARGET, + sender: { id: ALICE, name: "alice" }, + text: "first\ncontinued\n\n[queued message from bob at 2026-08-04T12:00:02.000Z]\nsecond", + isFromMe: false, + createdAt: first.createdAt, + conversationMeta: groupMetadata, + contextBlocks: expectedCrossContext(cross), + coalescedMessages: expectedCoalescedMessages(first, queued), + }); +}; + +const projectsMaterializedHarnessContext = () => { + const messages = materializedMessages(); + expectMaterializedProjection( + projectMaterializedHarnessContext(messages), + messages, + ); +}; + +const preservesSparseDirectShape = () => { + const ownMessage = message({ + id: "00000000-0000-4000-8000-000000000010", + conversationId: TARGET, + senderId: OWN, + parts: [{ type: "text", text: "self" }], + createdAt: "2026-08-04T12:00:03.000Z", + }); + const projected = projectHarnessTurn({ + ownAgentId: OWN, + agents: [], + context: { + currentMessages: [ownMessage], + crossConversationMessages: [], + conversations: [conversation(TARGET, [OWN, ALICE])], + }, + }); + + expect(projected).toEqual({ + id: ownMessage.id, + conversationId: TARGET, + sender: { id: OWN, name: OWN }, + text: "self", + isFromMe: true, + createdAt: ownMessage.createdAt, + conversationMeta: { + type: "dm", + participants: [`agent:${OWN}`, `agent:${ALICE}`], + }, + contextBlocks: {}, + }); + expect(projected).not.toHaveProperty("coalescedMessages"); + expect(projected.contextBlocks).not.toHaveProperty( + "crossConversationMessages", + ); +}; + +// @agent-code-guard/regression-only: these examples pin the exact channel-owned shape reused by package-private harness projection. +describe("harness turn projection", () => { + it("projects names, membership, cross-history, and coalesced current text", () => { + projectsMaterializedHarnessContext(); + }); + it("preserves sparse direct-message shape and identity fallback", () => { + preservesSparseDirectShape(); + }); +}); diff --git a/packages/client/src/moltzapd.ts b/packages/client/src/moltzapd.ts index 6127c4d33..5a8653df5 100644 --- a/packages/client/src/moltzapd.ts +++ b/packages/client/src/moltzapd.ts @@ -1,11 +1,18 @@ import type { Implementation } from "@modelcontextprotocol/server"; import { Effect, ExecutionStrategy, Exit, Scope } from "effect"; -import { conversationSearch } from "@moltzap/protocol/conversation"; +import { + conversationList, + conversationSearch, +} from "@moltzap/protocol/conversation"; import { agentsSearch } from "@moltzap/protocol/identity"; import { messagesRead } from "@moltzap/protocol/message"; +import type { ParamsOf } from "@moltzap/protocol/rpc"; import packageJson from "../package.json" with { type: "json" }; import { MoltZapChannelCore } from "./channel-core.js"; -import type { HarnessTurnEvent } from "./harness/index.js"; +import type { + HarnessSearchConversationsResult, + HarnessTurnEvent, +} from "./harness/index.js"; import { acquireHarnessMcpHttpServer } from "./harness-mcp-server.js"; import { makeHarnessMcpHttpHandlers } from "./harness-mcp-wire.js"; import type { @@ -14,6 +21,7 @@ import type { } from "./local-daemon-rpc.js"; import { MoltZapService, type ServiceRpcError } from "./service.js"; import type { ServiceConfigError } from "./config.js"; +import { drainPaginatedList } from "./pagination.js"; interface MoltzapdOptions { readonly profileName: string; @@ -65,6 +73,34 @@ const installTurnPublisher = ( ); }; +const searchConversationsForHarness = ( + service: MoltZapService, + params: ParamsOf, +): Effect.Effect => + Effect.gen(function* () { + const page = yield* service.callDefinition(conversationSearch, params); + const listed = yield* drainPaginatedList({ + definition: conversationList, + sendRpc: (definition, listParams) => + service.callDefinition(definition, listParams), + paramsForCursor: (cursor) => (cursor === undefined ? {} : { cursor }), + rowsForPage: (listPage) => listPage.items, + nextCursorForPage: (listPage) => listPage.nextCursor, + }); + const participantsByConversation = new Map( + listed.map((item) => [item.conversation.id, item.participants] as const), + ); + return { + conversations: page.conversations.map((conversation) => ({ + ...conversation, + participants: [ + ...(participantsByConversation.get(conversation.id) ?? []), + ], + })), + ...(page.nextCursor === undefined ? {} : { nextCursor: page.nextCursor }), + }; + }).pipe(Effect.withSpan("moltzapd.searchConversations")); + /** * Owns one registered agent's service, channel core, network connection, and * guarded loopback MCP listener for the lifetime of the caller's scope. @@ -115,7 +151,7 @@ export const acquireMoltzapd = ( searchAgents: (payload) => service.callDefinition(agentsSearch, payload), searchConversations: (payload) => - service.callDefinition(conversationSearch, payload), + searchConversationsForHarness(service, payload), status: makeStatusHandler(service, core), }); installTurnPublisher(core, handlers.active.publish);