diff --git a/CHANGELOG.md b/CHANGELOG.md index 6670090c4..0089bbe1b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added: daemon-backed `HarnessClient` + +`@moltzap/client` exposes an Effect `HarnessClient` for runtime adapters. Its +scoped turn stream comes from the daemon's loopback MCP surface, and each turn +carries a payload-only reply function bound to the conversation that produced +it. The daemon can consume the channel core's existing coalesced `Message` +batch without reading or committing presentation context. + +The old client-side `ReplyGuard` and OpenClaw duplicate-reply callback are +removed. A bound reply is not locally suppressed: every invocation reaches the +daemon. + +The official MCP handler remains responsible for discovery, tools, and all +standard subscription behavior. A package-local adapter owns only the MoltZap +turn-ready extension filter and notification required by the daemon's retained +listen response. + ### Removed: app-minted conversations Only endpoints open conversations. An agent creates one through diff --git a/docs/modules/client/src.mdx b/docs/modules/client/src.mdx index 50f11f9f0..1c88edff5 100644 --- a/docs/modules/client/src.mdx +++ b/docs/modules/client/src.mdx @@ -13,10 +13,33 @@ Public barrel for the MoltZap client package. ## Public surface -### [`AgentClientOptions`](https://github.com/chughtapan/moltzap/blob/main/protocol/dist/socket/agent-client.d.ts#L13) +### [`acquireHarnessClient`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/harness-client.ts#L41) + +_Function_ + +```ts +export const acquireHarnessClient = ( + options: HarnessClientOptions, +): Effect.Effect +``` + +Acquires one turn-ready harness connection and receive stream for the +lifetime of the enclosing scope. The private adapter owns MCP translation. + +**Returns:** The scoped adapter-facing service value. + +### [`AgentClientOptions`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/dist/socket/agent-client.d.ts#L13) _Interface_ +```ts +export interface AgentClientOptions { + readonly serverUrl: string; + readonly agentKey: AgentKey; + readonly onDisconnect?: (close: CloseInfo) => void; +} +``` + Configures agent client. ### [`ContextOptions`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/service.ts#L131) @@ -48,10 +71,87 @@ export interface ConversationMeta { Describes conversation meta. -### [`MoltZapAgentClient`](https://github.com/chughtapan/moltzap/blob/main/protocol/dist/socket/agent-client.d.ts#L19) +### [`HarnessClient`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/harness-client.ts#L23) _Class_ +```ts +export class HarnessClient extends Context.Tag("@moltzap/client/HarnessClient")< + HarnessClient, + HarnessClientService +>() {} +``` + +Effect service tag consumed by runtime adapters. + +### [`HarnessClientOptions`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/harness-client.ts#L29) + +_Interface_ + +```ts +export interface HarnessClientOptions { + /** Loopback `POST /mcp` endpoint owned by one running `moltzapd`. */ + readonly url: string; +} +``` + +Inputs needed to connect one scoped harness client. + +### [`HarnessClientService`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/harness-client.ts#L17) + +_Interface_ + +```ts +export interface HarnessClientService { + /** The sole receive stream owned by this scoped client. */ + readonly turns: Stream.Stream; +} +``` + +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) + +_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[]]; + /** 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. + +### [`makeHarnessClientLayer`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/harness-client.ts#L52) + +_Function_ + +```ts +export const makeHarnessClientLayer = ( + options: HarnessClientOptions, +): Layer.Layer +``` + +Builds the scoped runtime-adapter layer for one daemon endpoint. + +**Returns:** A Layer providing the scoped HarnessClient capability. + +### [`MoltZapAgentClient`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/dist/socket/agent-client.d.ts#L19) + +_Class_ + +```ts +export declare class MoltZapAgentClient extends ProtocolClientLifecycle { + constructor(options: AgentClientOptions); + call(tag: Tag, payload: PayloadForTag, opts?: RpcCallOptions): Effect.Effect, ErrorForTag | NotConnectedError | RpcTimeoutError>; +} +``` + Implements molt zap agent client. ### [`MoltZapService`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/service.ts#L262) @@ -189,10 +289,16 @@ Promise siblings — async/await consumers run the Effect at the edge with `Effect.runPromise`. Keep this class Effect-only so downstream callers compose failures and cancellation explicitly. -### [`RpcCallOptions`](https://github.com/chughtapan/moltzap/blob/main/protocol/dist/socket/lifecycle.d.ts#L12) +### [`RpcCallOptions`](https://github.com/chughtapan/moltzap/blob/main/packages/protocol/dist/socket/lifecycle.d.ts#L12) _Interface_ +```ts +export interface RpcCallOptions { + readonly timeoutMs?: number; +} +``` + Configures rpc call. ### [`ServiceRpcError`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/service.ts#L111) @@ -212,4 +318,5 @@ to that method's errors at the `call` site. ## Files +- `harness-client.ts` - `service.ts` diff --git a/docs/modules/openclaw-channel/src.mdx b/docs/modules/openclaw-channel/src.mdx index 8a364ca2c..59259f4b6 100644 --- a/docs/modules/openclaw-channel/src.mdx +++ b/docs/modules/openclaw-channel/src.mdx @@ -15,7 +15,7 @@ runtime entries from `index.*` at the extension root only, so the built ## Public surface -### [`createMoltzapChannelPlugin`](https://github.com/chughtapan/moltzap/blob/main/packages/openclaw-channel/src/openclaw-entry.ts#L1295) +### [`createMoltzapChannelPlugin`](https://github.com/chughtapan/moltzap/blob/main/packages/openclaw-channel/src/openclaw-entry.ts#L1226) _Function_ @@ -47,21 +47,15 @@ sequenceDiagram Core->>Plugin: enriched message arrives Plugin->>OC: dispatchReplyWithBufferedBlockDispatcher note over OC: agent pipeline → LLM - OC->>Plugin: deliver(payload, opts) — createReplyGuardedDeliver + OC->>Plugin: deliver(payload, opts) — createReplyDeliver Plugin->>Server: core.sendReply(conversationId, text) - alt second final reply for the same turn - Plugin->>Plugin: ReplyGuard already stamped
onDuplicateReply callback, return false - end OC->>Plugin: stopAccount(ctx) Plugin->>Core: core.disconnect() Plugin->>Plugin: activeClients.delete(account) ``` `deliver` returns `PromiseLike<boolean>` per openclaw contract; -false signals "not delivered" without throwing. The reply guard is -single-shot per inbound turn: a second final reply is suppressed locally -and reported through `MoltzapChannelPluginDeps.onDuplicateReply` rather -than a throw. +false signals a failed send without throwing. `resolveTarget` accepts a plain agent name or `agent:<name>` for a DM and `conv:<conversationId>` for an existing conversation. Plain names normalize @@ -69,7 +63,7 @@ to `agent:<name>`. Other colon-prefixed shapes are rejected. **Returns:** The created moltzap channel plugin. -### [`default`](https://github.com/chughtapan/moltzap/blob/main/packages/openclaw-channel/src/openclaw-entry.ts#L1325) +### [`default`](https://github.com/chughtapan/moltzap/blob/main/packages/openclaw-channel/src/openclaw-entry.ts#L1256) _Variable_ @@ -77,7 +71,7 @@ _Variable_ const plugin = ``` -### [`moltzapChannelPlugin`](https://github.com/chughtapan/moltzap/blob/main/packages/openclaw-channel/src/openclaw-entry.ts#L1322) +### [`moltzapChannelPlugin`](https://github.com/chughtapan/moltzap/blob/main/packages/openclaw-channel/src/openclaw-entry.ts#L1253) _Variable_ @@ -90,7 +84,7 @@ Shared singleton so a single registration reuses the same `activeClients` closure across `startAccount` and `sendText`. Tests import this directly to assert against that shared state. -### [`MoltzapChannelPlugin`](https://github.com/chughtapan/moltzap/blob/main/packages/openclaw-channel/src/openclaw-entry.ts#L1313) +### [`MoltzapChannelPlugin`](https://github.com/chughtapan/moltzap/blob/main/packages/openclaw-channel/src/openclaw-entry.ts#L1244) _TypeAlias_ diff --git a/packages/client/AGENTS.md b/packages/client/AGENTS.md index 2d7467e79..ff936e21f 100644 --- a/packages/client/AGENTS.md +++ b/packages/client/AGENTS.md @@ -6,10 +6,11 @@ lowest surface that meets the need: | Surface | Use when | |---|---| +| `HarnessClient` (via `@moltzap/client/harness-client`) | Runtime-adapter turns and conversation-bound reply over daemon MCP | | `MoltZapAgentClient` | Raw outbound RPC + inbound notifications | | `MoltZapChannelCore` (via `@moltzap/client/channel-base`) | Inbound turn-taking, coalescing, and enrichment | | `MoltZapService` | Managed conversation/context state on top of RPC | -| `@moltzap/client/channel-base` | Building a channel adapter; shared reply-guard + formatter primitives | +| `@moltzap/client/channel-base` | Building a channel adapter; shared turn and formatter primitives | ## Structure @@ -19,6 +20,11 @@ lowest surface that meets the need: - `src/moltzapd.ts` — the daemon: agent ownership + single-flight teardown; `src/harness-mcp-server.ts` / `harness-mcp-wire.ts` are its MCP HTTP boundary. +- `src/harness-client.ts` — public adapter-facing Effect capability; + `src/harness/` owns its private MCP client and shared wire contract. +- `src/harness-mcp-subscription.ts` — package-owned adapter for the exact + turn-ready extension to `subscriptions/listen`; every other MCP request and + lifecycle remains delegated to the official SDK handler. - `src/agent-client.ts` — re-exports `MoltZapAgentClient` from `@moltzap/protocol/socket`. - `src/auth.ts` — `registerAgent` HTTP bootstrap (mints agentId + @@ -29,7 +35,7 @@ lowest surface that meets the need: - `src/cli/` — `moltzap` CLI binary, per-command files under `commands/`. -Subpath exports: `./channel-base`, `./test-utils`, `./auth`, +Subpath exports: `./channel-base`, `./harness-client`, `./test-utils`, `./auth`, `./pagination`, `./notification`. ## Concepts @@ -51,8 +57,9 @@ Subpath exports: `./channel-base`, `./test-utils`, `./auth`, continues; unset means unbounded, so a hung handler stalls the drain. - **Inbound interceptor** — optional `ChannelCoreOptions.inboundInterceptor`, the endpoint-side gate - between enrichment and the handler: deliver or drop, judged on the - batch's newest message and binding on the whole turn. Pacing is + before the selected handler: deliver or drop, judged on the batch's + newest message and binding on the whole turn. Enriched adapter delivery + enriches before this gate; raw daemon delivery does not enrich. Pacing is suspension inside the gate, not a verdict; a broken gate delivers. - **Cross-conversation context** — snippets from the agent's other conversations, attached to the enriched inbound message and @@ -62,12 +69,13 @@ Subpath exports: `./channel-base`, `./test-utils`, `./auth`, ## Code -- `@moltzap/client/channel-base` is the single definition site for - `ReplyGuard` (per-turn single-shot guard; the server accepts every - well-formed send, so nothing else stops a runtime that replies - twice) and the markup-parameterized formatters `formatCrossConv` / - `formatGroupBlock` / `getGroupFields`. Detail JSDoc: the - `src/channel-base/*.ts` file headers. +- `@moltzap/client/channel-base` owns the markup-parameterized formatters + `formatCrossConv` / `formatGroupBlock` / `getGroupFields`. Detail JSDoc: + the `src/channel-base/*.ts` file headers. +- Keep `harness-mcp-subscription.ts` limited to extension capability checking, + one retained turn-ready response, and its acknowledgement/event/completion + frames. Discovery, tools, standard subscriptions, and unrelated MCP + lifecycle behavior stay SDK-owned. ## Tests diff --git a/packages/client/package.json b/packages/client/package.json index 8773786bb..08a486319 100644 --- a/packages/client/package.json +++ b/packages/client/package.json @@ -33,6 +33,10 @@ "types": "./dist/auth.d.ts", "import": "./dist/auth.js" }, + "./harness-client": { + "types": "./dist/harness-client.d.ts", + "import": "./dist/harness-client.js" + }, "./pagination": { "types": "./dist/pagination.d.ts", "import": "./dist/pagination.js" @@ -84,6 +88,7 @@ "@effect/printer-ansi": "^0.50.0", "@effect/rpc": "^0.76.0", "@effect/typeclass": "^0.41.0", + "@modelcontextprotocol/client": "2.0.0-beta.5", "@modelcontextprotocol/node": "2.0.0-beta.5", "@modelcontextprotocol/server": "2.0.0-beta.5", "@moltzap/protocol": "workspace:*", @@ -91,7 +96,6 @@ }, "devDependencies": { "@effect/vitest": "^0.30.0", - "@modelcontextprotocol/client": "2.0.0-beta.5", "@moltzap/server-core": "workspace:*", "@typescript/native": "npm:typescript@^7.0.2", "eslint": "^9", diff --git a/packages/client/safer-architecture.config.json b/packages/client/safer-architecture.config.json index e323de711..e097c80e0 100644 --- a/packages/client/safer-architecture.config.json +++ b/packages/client/safer-architecture.config.json @@ -1,5 +1,6 @@ { "minExportedSiblingModules": 6, + "maxSubpathExports": 6, "maxPublicExports": 29, "minPublicFacadeModules": 9, "folderChildCountOverrides": [ diff --git a/packages/client/src/MODULE.md b/packages/client/src/MODULE.md index c88b01181..27b252081 100644 --- a/packages/client/src/MODULE.md +++ b/packages/client/src/MODULE.md @@ -8,10 +8,33 @@ Public barrel for the MoltZap client package. ## Public surface -### [`AgentClientOptions`](./../../../protocol/dist/socket/agent-client.d.ts#L13) +### [`acquireHarnessClient`](./harness-client.ts#L41) + +_Function_ + +```ts +export const acquireHarnessClient = ( + options: HarnessClientOptions, +): Effect.Effect +``` + +Acquires one turn-ready harness connection and receive stream for the +lifetime of the enclosing scope. The private adapter owns MCP translation. + +**Returns:** The scoped adapter-facing service value. + +### [`AgentClientOptions`](./../../protocol/dist/socket/agent-client.d.ts#L13) _Interface_ +```ts +export interface AgentClientOptions { + readonly serverUrl: string; + readonly agentKey: AgentKey; + readonly onDisconnect?: (close: CloseInfo) => void; +} +``` + Configures agent client. ### [`ContextOptions`](./service.ts#L131) @@ -43,10 +66,87 @@ export interface ConversationMeta { Describes conversation meta. -### [`MoltZapAgentClient`](./../../../protocol/dist/socket/agent-client.d.ts#L19) +### [`HarnessClient`](./harness-client.ts#L23) _Class_ +```ts +export class HarnessClient extends Context.Tag("@moltzap/client/HarnessClient")< + HarnessClient, + HarnessClientService +>() {} +``` + +Effect service tag consumed by runtime adapters. + +### [`HarnessClientOptions`](./harness-client.ts#L29) + +_Interface_ + +```ts +export interface HarnessClientOptions { + /** Loopback `POST /mcp` endpoint owned by one running `moltzapd`. */ + readonly url: string; +} +``` + +Inputs needed to connect one scoped harness client. + +### [`HarnessClientService`](./harness-client.ts#L17) + +_Interface_ + +```ts +export interface HarnessClientService { + /** The sole receive stream owned by this scoped client. */ + readonly turns: Stream.Stream; +} +``` + +Adapter-facing capability backed only by the daemon's loopback MCP surface. + +### [`HarnessTurn`](./harness-client.ts#L7) + +_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[]]; + /** 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. + +### [`makeHarnessClientLayer`](./harness-client.ts#L52) + +_Function_ + +```ts +export const makeHarnessClientLayer = ( + options: HarnessClientOptions, +): Layer.Layer +``` + +Builds the scoped runtime-adapter layer for one daemon endpoint. + +**Returns:** A Layer providing the scoped HarnessClient capability. + +### [`MoltZapAgentClient`](./../../protocol/dist/socket/agent-client.d.ts#L19) + +_Class_ + +```ts +export declare class MoltZapAgentClient extends ProtocolClientLifecycle { + constructor(options: AgentClientOptions); + call(tag: Tag, payload: PayloadForTag, opts?: RpcCallOptions): Effect.Effect, ErrorForTag | NotConnectedError | RpcTimeoutError>; +} +``` + Implements molt zap agent client. ### [`MoltZapService`](./service.ts#L262) @@ -184,10 +284,16 @@ Promise siblings — async/await consumers run the Effect at the edge with `Effect.runPromise`. Keep this class Effect-only so downstream callers compose failures and cancellation explicitly. -### [`RpcCallOptions`](./../../../protocol/dist/socket/lifecycle.d.ts#L12) +### [`RpcCallOptions`](./../../protocol/dist/socket/lifecycle.d.ts#L12) _Interface_ +```ts +export interface RpcCallOptions { + readonly timeoutMs?: number; +} +``` + Configures rpc call. ### [`ServiceRpcError`](./service.ts#L111) @@ -207,4 +313,5 @@ to that method's errors at the `call` site. ## Files +- `harness-client.ts` - `service.ts` diff --git a/packages/client/src/__tests__/channel-base/reply-guard.test.ts b/packages/client/src/__tests__/channel-base/reply-guard.test.ts deleted file mode 100644 index 95b048d87..000000000 --- a/packages/client/src/__tests__/channel-base/reply-guard.test.ts +++ /dev/null @@ -1,146 +0,0 @@ -/** - * Unit tests for `ReplyGuard` initial state, first consume, and repeated - * consume behavior. - * - * Uses Effect's TestClock so the timestamp on first consume is deterministic. - */ - -import * as fc from "fast-check"; -import { describe, expect, it } from "vitest"; -import { Effect, Option, TestClock, TestContext } from "effect"; -import { ReplyGuard } from "../../channel-base/reply-guard.js"; - -const FIXED_TS = 1_700_000_000_500; -const LATER_TS = FIXED_TS + 999; -const PROPERTY_ATTEMPT_MIN = 1; -const PROPERTY_ATTEMPT_MAX = 8; -// Keep the property run count low enough for parallel-suite load while still -// exercising the attempt-range boundaries with multiple shuffles. -const PROPERTY_NUM_RUNS = 20; - -describe("ReplyGuard", () => { - it( - "property: only the first consume returns true; consumedAt is stamped exactly once", - propertySingleShot, - ); - it("initial consumedAt is Option.none", initialConsumedAtIsNone); - it( - "first consume returns true and stamps consumedAt with Clock.currentTimeMillis", - firstConsumeStamps, - ); - it( - "second consume returns false; consumedAt unchanged", - secondConsumeIsFalse, - ); - it("consumedAt is idempotent on repeated reads", consumedAtIdempotent); - it( - "concurrent begin claims admit exactly one sender", - concurrentBeginAdmitsOne, - ); - it("abort reopens an unclaimed-but-unconsumed guard", abortReopensGuard); -}); - -function runSingleShotAttempts(attempts: number) { - return Effect.runPromise( - Effect.gen(function* () { - yield* TestClock.setTime(FIXED_TS); - const guard = new ReplyGuard(); - const results: boolean[] = []; - for (let i = 0; i < attempts; i += 1) { - results.push(yield* guard.consume()); - } - const stamped = yield* guard.consumedAt; - expect(results[0]).toBe(true); - expect(results.slice(1)).toEqual( - Array.from({ length: attempts - 1 }, () => false), - ); - expect(Option.getOrNull(stamped)).toBe(FIXED_TS); - }).pipe(Effect.provide(TestContext.TestContext)), - ); -} - -function propertySingleShot() { - return fc.assert( - fc.asyncProperty( - fc.integer({ min: PROPERTY_ATTEMPT_MIN, max: PROPERTY_ATTEMPT_MAX }), - runSingleShotAttempts, - ), - { numRuns: PROPERTY_NUM_RUNS }, - ); -} - -function initialConsumedAtIsNone(): void { - const guard = new ReplyGuard(); - expect(Option.isNone(Effect.runSync(guard.consumedAt))).toBe(true); -} - -function firstConsumeStamps() { - return Effect.runPromise( - Effect.gen(function* () { - yield* TestClock.setTime(FIXED_TS); - const guard = new ReplyGuard(); - const consumed = yield* guard.consume(); - const stamped = yield* guard.consumedAt; - expect(consumed).toBe(true); - expect(Option.getOrNull(stamped)).toBe(FIXED_TS); - }).pipe(Effect.provide(TestContext.TestContext)), - ); -} - -function secondConsumeIsFalse() { - return Effect.runPromise( - Effect.gen(function* () { - yield* TestClock.setTime(FIXED_TS); - const guard = new ReplyGuard(); - const first = yield* guard.consume(); - yield* TestClock.setTime(LATER_TS); - const second = yield* guard.consume(); - const stamped = yield* guard.consumedAt; - expect(first).toBe(true); - expect(second).toBe(false); - // Unchanged from first-consume moment, NOT the wall clock at second-consume. - expect(Option.getOrNull(stamped)).toBe(FIXED_TS); - }).pipe(Effect.provide(TestContext.TestContext)), - ); -} - -function consumedAtIdempotent(): void { - const guard = new ReplyGuard(); - expect(Option.isNone(Effect.runSync(guard.consumedAt))).toBe(true); - expect(Option.isNone(Effect.runSync(guard.consumedAt))).toBe(true); -} - -// The send between claim and stamp is asynchronous, so exclusivity must be -// decided at begin() time: N racing delivers admit exactly one sender even -// though none has consumed yet. -function concurrentBeginAdmitsOne() { - return Effect.runPromise( - Effect.gen(function* () { - const guard = new ReplyGuard(); - const concurrentAttempts = 8; - const claims = yield* Effect.all( - Array.from({ length: concurrentAttempts }, () => guard.begin()), - { concurrency: concurrentAttempts }, - ); - expect(claims.filter(Boolean)).toHaveLength(1); - }).pipe(Effect.provide(TestContext.TestContext)), - ); -} - -function abortReopensGuard() { - return Effect.runPromise( - Effect.gen(function* () { - yield* TestClock.setTime(FIXED_TS); - const guard = new ReplyGuard(); - expect(yield* guard.begin()).toBe(true); - // A failed send aborts the claim without stamping: the next deliver - // may claim again, and consuming then stamps normally. - yield* guard.abort(); - expect(yield* guard.begin()).toBe(true); - expect(yield* guard.consume()).toBe(true); - // Consumed is terminal: neither begin nor abort reopens it. - yield* guard.abort(); - expect(yield* guard.begin()).toBe(false); - }).pipe(Effect.provide(TestContext.TestContext)), - ); -} 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 08770422f..7752f729d 100644 --- a/packages/client/src/__tests__/service/core/moltzapd.integration.test.ts +++ b/packages/client/src/__tests__/service/core/moltzapd.integration.test.ts @@ -7,9 +7,27 @@ import { import { live as it } from "@effect/vitest"; // eslint-disable-next-line agent-code-guard/prefer-effect-platform -- This integration test needs a passive TCP port blocker to force the package-private Node listener's real bind-failure path. import { createServer, type Server as NodeHttpServer } from "node:http"; -import { Cause, Data, Effect, Exit, Schema, Scope } from "effect"; +import { + Cause, + Data, + Duration, + Effect, + Exit, + Fiber, + Option, + Schema, + Scope, + Stream, +} from "effect"; import { expect } from "vitest"; +import type { ConversationId } from "@moltzap/protocol/conversation"; +import type { Message } from "@moltzap/protocol/message"; import { withTestServiceConfig } from "../../../config.test-utils.js"; +import { + acquireHarnessClient, + type HarnessClientService, + type HarnessTurn, +} from "../../../harness-client.js"; import { getMoltZapAgentServiceSocketPath } from "../../../local-paths.js"; import { acquireMoltzapd } from "../../../moltzapd.js"; import * as H from "../../support/index.js"; @@ -18,6 +36,8 @@ const PROFILE_NAME = "moltzapd-integration"; const MCP_PATH = "/mcp"; const LOOPBACK_HOST = "127.0.0.1"; const MODERN_PROTOCOL_VERSION = "2026-07-28"; +const PEER_MESSAGE = "hello through the harness"; +const HARNESS_REPLY = "reply through the harness"; const healthSchema = Schema.Struct({ connections: Schema.Number }); type RegisteredAgent = Effect.Effect.Success< @@ -25,6 +45,14 @@ type RegisteredAgent = Effect.Effect.Success< >; type MoltzapdServer = Effect.Effect.Success>; +interface RoundTripFixture { + readonly harness: HarnessClientService; + readonly owner: RegisteredAgent; + readonly peer: RegisteredAgent; + readonly conversationId: ConversationId; + readonly socketPath: string; +} + interface PortBlocker { readonly port: number; readonly server: NodeHttpServer; @@ -45,6 +73,27 @@ const healthConnections = (): Effect.Effect => Effect.provide(NodeHttpClient.layer), ); +const waitForConnectionCount = ( + expected: number, +): Effect.Effect => { + const poll: Effect.Effect = Effect.suspend(() => + healthConnections().pipe( + Effect.flatMap((actual) => + actual === expected + ? Effect.void + : Effect.sleep("10 millis").pipe(Effect.zipRight(poll)), + ), + ), + ); + return poll.pipe( + Effect.timeoutFail({ + duration: Duration.millis(H.NOTIFICATION_WAIT_MS), + onTimeout: () => + new Error(`timeout waiting for ${expected} server connections`), + }), + ); +}; + const listenPortBlocker: Effect.Effect = Effect.async((resume) => { const server = createServer(); @@ -164,6 +213,139 @@ function runRegisteredAgent(registered: RegisteredAgent) { }).pipe(Effect.provide(NodeContext.layer)); } +const takeHead = ( + stream: Stream.Stream, + label: string, +): Effect.Effect => + stream.pipe( + Stream.runHead, + Effect.timeoutFail({ + duration: Duration.millis(H.NOTIFICATION_WAIT_MS), + onTimeout: () => new Error(`timeout waiting for ${label}`), + }), + Effect.map((head) => + Option.getOrThrowWith( + head, + () => new Error(`${label} stream closed before delivery`), + ), + ), + ); + +const expectNoUnixSocket = (socketPath: string) => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + expect(yield* fileSystem.exists(socketPath)).toBe(false); + }); + +const expectHarnessTurn = ( + turn: HarnessTurn, + 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); +}; + +const expectPeerReply = ( + reply: Message, + owner: RegisteredAgent, + conversationId: ConversationId, +): void => { + expect(reply.conversationId).toBe(conversationId); + expect(reply.senderId).toBe(owner.agentId); + expect(H.textContent(reply)).toBe(HARNESS_REPLY); +}; + +const waitForPeerReply = ( + peer: RegisteredAgent, + owner: RegisteredAgent, + conversationId: ConversationId, +) => + takeHead( + peer.client + .subscribe(H.messageReceivedNotificationDefinition) + .pipe( + Stream.filter( + ({ message }) => + message.senderId === owner.agentId && + message.conversationId === conversationId, + ), + ), + "harness reply", + ).pipe(Effect.map(({ message }) => message)); + +const runMcpMessageRoundTrip = ({ + harness, + owner, + peer, + conversationId, + socketPath, +}: RoundTripFixture) => + Effect.gen(function* () { + const turnFiber = yield* Effect.fork( + takeHead(harness.turns, "harness turn"), + ); + const peerReplyFiber = yield* Effect.fork( + waitForPeerReply(peer, owner, conversationId), + ); + + yield* peer.client.call(H.messagesSend.name, { + conversationId, + parts: [{ type: "text", text: PEER_MESSAGE }], + }); + + const turn = yield* Fiber.join(turnFiber); + expectHarnessTurn(turn, peer, conversationId); + yield* expectNoUnixSocket(socketPath); + + yield* turn.reply(HARNESS_REPLY); + expectPeerReply(yield* Fiber.join(peerReplyFiber), owner, conversationId); + yield* expectNoUnixSocket(socketPath); + }); + +function runHarnessRoundTrip(owner: RegisteredAgent, peer: RegisteredAgent) { + return Effect.gen(function* () { + const socketPath = getMoltZapAgentServiceSocketPath(owner.agentId); + yield* expectNoUnixSocket(socketPath); + + yield* peer.client.connect(); + yield* Effect.scoped( + Effect.gen(function* () { + const server = yield* acquireMoltzapd({ + profileName: PROFILE_NAME, + port: 0, + }); + const harness = yield* acquireHarnessClient({ + url: harnessUrl(server).href, + }); + yield* expectNoUnixSocket(socketPath); + + const created = yield* peer.client.call( + H.agentConversationCreate.name, + { participants: [owner.agentId] }, + ); + yield* runMcpMessageRoundTrip({ + harness, + owner, + peer, + conversationId: created.conversation.id, + socketPath, + }); + }), + ); + + yield* expectNoUnixSocket(socketPath); + expect(yield* healthConnections()).toBe(1); + }).pipe(Effect.provide(NodeContext.layer)); +} + const runFailedAcquisition = Effect.gen(function* () { const blocker = yield* acquirePortBlocker; const ambientScope = yield* Effect.acquireRelease(Scope.make(), (scope) => @@ -213,6 +395,25 @@ function runFailedAcquisitionWithProfile(registered: RegisteredAgent) { ); } +function runHarnessRoundTripWithProfile({ + owner, + peer, +}: { + readonly owner: RegisteredAgent; + readonly peer: RegisteredAgent; +}) { + return withTestServiceConfig( + { + profileName: PROFILE_NAME, + agentName: PROFILE_NAME, + agentId: owner.agentId, + agentKey: owner.apiKey, + serverUrl: H.coreBaseUrl(), + }, + runHarnessRoundTrip(owner, peer), + ); +} + H.setupServiceIntegration(); it("owns one agent connection and MCP listener without a Unix socket", () => { @@ -224,6 +425,22 @@ it("owns one agent connection and MCP listener without a Unix socket", () => { ); }); +it("round-trips a peer message and bound reply through MCP only", () => { + expect.hasAssertions(); + return Effect.acquireUseRelease( + Effect.all({ + owner: H.registerAgent("moltzapd-round-trip-owner"), + peer: H.registerAgent("moltzapd-round-trip-peer"), + }), + runHarnessRoundTripWithProfile, + ({ owner, peer }) => + H.closeAll([], [owner.client, peer.client]).pipe( + Effect.zipRight(waitForConnectionCount(0)), + Effect.orDieWith(toError), + ), + ); +}); + it("rolls back the agent connection when MCP listener acquisition fails", () => { expect.hasAssertions(); return Effect.acquireUseRelease( diff --git a/packages/client/src/channel-base/README.md b/packages/client/src/channel-base/README.md index 9e41b0d0d..ac001f523 100644 --- a/packages/client/src/channel-base/README.md +++ b/packages/client/src/channel-base/README.md @@ -3,7 +3,6 @@ This folder is the runtime-neutral base layer shared by MoltZap channel adapters. -- `reply-guard.ts` enforces one final reply per inbound turn. - `format-cross-conv.ts` and `format-group-block.ts` render enriched context using caller-selected markup. - `index.ts` is the curated `@moltzap/client/channel-base` surface. diff --git a/packages/client/src/channel-base/index.ts b/packages/client/src/channel-base/index.ts index f8c228608..eb210aab6 100644 --- a/packages/client/src/channel-base/index.ts +++ b/packages/client/src/channel-base/index.ts @@ -4,9 +4,6 @@ /** Re-exports the public API from `../bounded-map.js`. */ export { BoundedMap } from "../bounded-map.js"; -/** Re-exports the public API from `./reply-guard.js`. */ -export { ReplyGuard } from "./reply-guard.js"; - /** Re-exports the public API from `./format-cross-conv.js`. */ export { formatCrossConv, diff --git a/packages/client/src/channel-base/reply-guard.ts b/packages/client/src/channel-base/reply-guard.ts deleted file mode 100644 index b8ea1054c..000000000 --- a/packages/client/src/channel-base/reply-guard.ts +++ /dev/null @@ -1,87 +0,0 @@ -/** - * Channel-base `ReplyGuard` — per-turn single-shot duplicate-reply detection. - * - * One `ReplyGuard` instance per inbound turn (created inside the deliver - * wrapper). This is the only enforcement of "one final reply per inbound - * turn": the server accepts every well-formed send, so a runtime that - * delivers twice would otherwise double-post. - * - * The send a guard protects is asynchronous, so a bare consumed check is a - * race: two concurrent delivers can both observe "not consumed" before - * either send completes. `begin()` therefore claims the guard in one - * synchronous step BEFORE the send; `consume()` stamps after a successful - * send; `abort()` reopens the guard after a failed send so a retried - * deliver can still go through. A deliver arriving while another is - * mid-send is a duplicate, not a queued retry. - */ - -import { Clock, Effect, Option } from "effect"; - -/** Implements reply guard. */ -export class ReplyGuard { - private consumedAtMillis: number | null = null; - private inFlight = false; - - /** - * Claim the guard for one send attempt. `true` exactly once per open - * window: `false` when the guard is already consumed OR another send is - * mid-flight. The claim is taken synchronously, so concurrent callers - * cannot both win. - * @returns Whether this caller owns the send attempt. - */ - begin(): Effect.Effect { - return Effect.sync(() => { - if (this.consumedAtMillis !== null || this.inFlight) { - return false; - } - this.inFlight = true; - return true; - }); - } - - /** - * Reopen the guard after a failed send so a retried deliver can claim it - * again. No-op when the guard is already consumed. - * @returns Completion of the reopen. - */ - abort(): Effect.Effect { - return Effect.sync(() => { - this.inFlight = false; - }); - } - - /** - * Returns `true` on first call (transitions internal state from - * "not-consumed" to "consumed-at-now") and releases any in-flight claim; - * returns `false` on every later call. Reads `Clock.currentTimeMillis` - * inside the Effect on first call. - * @returns The ts result. - */ - consume(): Effect.Effect { - return Effect.gen( - function* (this: ReplyGuard) { - if (this.consumedAtMillis !== null) { - return false; - } - const ts = yield* Clock.currentTimeMillis; - this.consumedAtMillis = ts; - this.inFlight = false; - return true; - }.bind(this), - ); - } - - /** - * `Option.none` before the first `consume`; `Option.some(epochMs)` after, - * where `epochMs` is the value Clock returned at the first-consume moment. - * Idempotent on second-and-later reads. - * @returns The consumed at result. - */ - get consumedAt(): Effect.Effect> { - return Effect.sync(() => - this.consumedAtMillis === null - ? Option.none() - : Option.some(this.consumedAtMillis), - ); - } -} diff --git a/packages/client/src/channel-core-raw.test.ts b/packages/client/src/channel-core-raw.test.ts new file mode 100644 index 000000000..f52b1fe5e --- /dev/null +++ b/packages/client/src/channel-core-raw.test.ts @@ -0,0 +1,293 @@ +/** + * Raw scheduled-turn delivery preserves the channel core's serialized queue + * semantics while leaving presentation enrichment and markers to its client. + */ + +import { live as it } from "@effect/vitest"; +import { Data, Deferred, Effect } from "effect"; +import { expect, vi } from "vitest"; + +import { + FIRST_TEXT, + SECOND_TEXT, + buildMessage, + conversation, + createFakeChannelService, + flushDispatchChainEffect, + message, + MoltZapChannelCore, + type EnrichedInboundMessage, + type FakeChannelService, + type Message, +} from "./channel-core-test-support.js"; + +const CONV_1 = "conv-1"; +const CONV_2 = "conv-2"; +const THIRD_TEXT = "third"; +const HANDLER_TIMEOUT_MS = 20; + +class RawHandlerTestError extends Data.TaggedError("RawHandlerTestError")<{ + readonly message: string; +}> {} + +function emitText( + fake: FakeChannelService, + spec: { readonly id: string; readonly conversationId: string }, + text: string, +): Message { + const raw = buildMessage({ + id: spec.id, + conversationId: spec.conversationId, + parts: [{ type: "text", text }], + }); + fake.emit.message(raw); + return raw; +} + +function awaitSignal( + deferred: Deferred.Deferred, +): Effect.Effect { + return Deferred.await(deferred).pipe(Effect.asVoid); +} + +function expectCoalescedRawBatches( + batches: ReadonlyArray, + originals: readonly [Message, Message, Message], + intercepted: readonly string[], +): void { + expect(batches).toHaveLength(2); + expect(batches.map((batch) => batch.map((raw) => raw.id))).toEqual([ + [message("msg-1")], + [message("msg-2"), message("msg-3")], + ]); + expect(batches[0]?.[0]).toBe(originals[0]); + expect(batches[1]?.[0]).toBe(originals[1]); + expect(batches[1]?.[1]).toBe(originals[2]); + expect(intercepted).toEqual([message("msg-1"), message("msg-3")]); +} + +function rawBatchesCoalesceExistingMessages() { + return Effect.gen(function* () { + const fake = createFakeChannelService({ ownAgentId: "agent-self" }); + const intercepted: string[] = []; + const core = new MoltZapChannelCore({ + service: fake.service, + inboundInterceptor: (raw) => + Effect.sync(() => { + intercepted.push(raw.id); + return { _tag: "deliver" } as const; + }), + }); + const firstStarted = yield* Deferred.make(); + const releaseFirst = yield* Deferred.make(); + const secondHandled = yield* Deferred.make(); + const batches: Array = []; + core.onRawInbound((messages) => + Effect.gen(function* () { + batches.push(messages); + if (batches.length === 1) { + yield* Deferred.succeed(firstStarted, undefined); + yield* awaitSignal(releaseFirst); + return; + } + yield* Deferred.succeed(secondHandled, undefined); + }), + ); + + const first = emitText( + fake, + { id: "msg-1", conversationId: CONV_1 }, + FIRST_TEXT, + ); + yield* awaitSignal(firstStarted); + const second = emitText( + fake, + { id: "msg-2", conversationId: CONV_1 }, + SECOND_TEXT, + ); + const third = emitText( + fake, + { id: "msg-3", conversationId: CONV_1 }, + THIRD_TEXT, + ); + yield* Deferred.succeed(releaseFirst, undefined); + yield* awaitSignal(secondHandled); + + expectCoalescedRawBatches(batches, [first, second, third], intercepted); + }); +} + +it( + "onRawInbound receives the existing same-conversation batch after interception", + rawBatchesCoalesceExistingMessages, +); + +function rawBatchesKeepOtherConversationsInOrder() { + return Effect.gen(function* () { + const fake = createFakeChannelService({ ownAgentId: "agent-self" }); + const core = new MoltZapChannelCore({ service: fake.service }); + const firstStarted = yield* Deferred.make(); + const releaseFirst = yield* Deferred.make(); + const allHandled = yield* Deferred.make(); + const batches: Array = []; + core.onRawInbound((messages) => + Effect.gen(function* () { + batches.push(messages); + if (batches.length === 1) { + yield* Deferred.succeed(firstStarted, undefined); + yield* awaitSignal(releaseFirst); + return; + } + if (batches.length === 3) { + yield* Deferred.succeed(allHandled, undefined); + } + }), + ); + + emitText(fake, { id: "msg-1", conversationId: CONV_1 }, FIRST_TEXT); + yield* awaitSignal(firstStarted); + emitText(fake, { id: "msg-other", conversationId: CONV_2 }, "elsewhere"); + emitText(fake, { id: "msg-2", conversationId: CONV_1 }, SECOND_TEXT); + emitText(fake, { id: "msg-3", conversationId: CONV_1 }, THIRD_TEXT); + yield* Deferred.succeed(releaseFirst, undefined); + yield* awaitSignal(allHandled); + + expect(batches.map((batch) => batch.map((raw) => raw.id))).toEqual([ + [message("msg-1")], + [message("msg-other")], + [message("msg-2"), message("msg-3")], + ]); + expect(batches.map((batch) => batch[0]?.conversationId)).toEqual([ + conversation(CONV_1), + conversation(CONV_2), + conversation(CONV_1), + ]); + }); +} + +it( + "onRawInbound keeps other conversations in arrival order", + rawBatchesKeepOtherConversationsInOrder, +); + +function observeEnrichmentReads(fake: FakeChannelService) { + return [ + vi.spyOn(fake.service, "getConversation"), + vi.spyOn(fake.service, "getAgentName"), + vi.spyOn(fake.service, "resolveAgentName"), + vi.spyOn(fake.service, "peekContextEntries"), + vi.spyOn(fake.service, "peekFullMessages"), + ]; +} + +function seedPresentationContext(fake: FakeChannelService): void { + fake.state.setConversation(CONV_1, { type: "dm", participants: [] }); + fake.state.setAgentName("agent-alice", "Alice"); + fake.state.setContextEntries(CONV_1, [ + { + conversationId: CONV_2, + senderName: "Bob", + text: "context summary", + minutesAgo: 1, + count: 1, + }, + ]); + fake.state.setFullMessages(CONV_1, [ + { + conversationId: conversation(CONV_2), + senderName: "Bob", + senderId: "agent-bob", + text: "full context", + timestamp: "2026-08-03T00:00:00.000Z", + }, + ]); +} + +function rawDeliveryDoesNotAdvancePresentationMarkers() { + return Effect.gen(function* () { + const fake = createFakeChannelService({ ownAgentId: "agent-self" }); + const enrichmentCalls = observeEnrichmentReads(fake); + seedPresentationContext(fake); + const core = new MoltZapChannelCore({ service: fake.service }); + const rawHandled = yield* Deferred.make(); + core.onRawInbound(() => Deferred.succeed(rawHandled, undefined)); + + emitText(fake, { id: "msg-raw", conversationId: CONV_1 }, FIRST_TEXT); + yield* awaitSignal(rawHandled); + + for (const call of enrichmentCalls) { + expect(call).not.toHaveBeenCalled(); + } + + const enrichedHandled = yield* Deferred.make(); + const enriched: EnrichedInboundMessage[] = []; + core.onInbound((incoming) => + Effect.sync(() => { + enriched.push(incoming); + }).pipe(Effect.zipRight(Deferred.succeed(enrichedHandled, undefined))), + ); + emitText(fake, { id: "msg-enriched", conversationId: CONV_1 }, SECOND_TEXT); + yield* awaitSignal(enrichedHandled); + + expect(enriched[0]?.contextBlocks.crossConversation).toHaveLength(1); + expect(enriched[0]?.contextBlocks.crossConversationMessages).toHaveLength( + 1, + ); + }); +} + +it( + "raw delivery leaves presentation markers for a later enriched turn", + rawDeliveryDoesNotAdvancePresentationMarkers, +); + +function rawHandlerFailureKeepsTheConsumerAlive() { + return Effect.gen(function* () { + const fake = createFakeChannelService({ ownAgentId: "agent-self" }); + const core = new MoltZapChannelCore({ service: fake.service }); + const nextHandled = yield* Deferred.make(); + core.onRawInbound((messages) => + messages[0]?.id === message("msg-failed") + ? Effect.fail( + new RawHandlerTestError({ message: "raw handler failed" }), + ) + : Deferred.succeed(nextHandled, undefined), + ); + + emitText(fake, { id: "msg-failed", conversationId: CONV_1 }, FIRST_TEXT); + yield* flushDispatchChainEffect; + emitText(fake, { id: "msg-next", conversationId: CONV_2 }, SECOND_TEXT); + yield* awaitSignal(nextHandled); + }); +} + +it( + "a failed raw handler does not stop the next turn", + rawHandlerFailureKeepsTheConsumerAlive, +); + +function rawHandlerTimeoutKeepsTheConsumerAlive() { + return Effect.gen(function* () { + const fake = createFakeChannelService({ ownAgentId: "agent-self" }); + const core = new MoltZapChannelCore({ + service: fake.service, + turnTimeoutMs: HANDLER_TIMEOUT_MS, + }); + const nextHandled = yield* Deferred.make(); + core.onRawInbound((messages) => + messages[0]?.id === message("msg-stuck") + ? Effect.never + : Deferred.succeed(nextHandled, undefined), + ); + + emitText(fake, { id: "msg-stuck", conversationId: CONV_1 }, FIRST_TEXT); + yield* flushDispatchChainEffect; + emitText(fake, { id: "msg-next", conversationId: CONV_2 }, SECOND_TEXT); + yield* awaitSignal(nextHandled); + }); +} + +it( + "a timed-out raw handler releases the serialized drain", + rawHandlerTimeoutKeepsTheConsumerAlive, +); diff --git a/packages/client/src/channel-core.ts b/packages/client/src/channel-core.ts index 743451d3f..6521edb57 100644 --- a/packages/client/src/channel-core.ts +++ b/packages/client/src/channel-core.ts @@ -1,8 +1,9 @@ /** * Channel core: serialized inbound turn delivery over a MoltZapService. * Owns the inbound queue, same-conversation coalescing, the interceptor - * gate, the per-turn timeout, and channel teardown; enrichment lives in - * channel-core-enrichment.ts. + * gate, the per-turn timeout, and channel teardown. Enriched adapter delivery + * uses channel-core-enrichment.ts; raw daemon delivery receives the coalesced + * Message batch without reading or committing presentation context. */ import { Cause, Chunk, Duration, Effect, Fiber, Option, Queue } from "effect"; @@ -112,9 +113,9 @@ export interface ChannelCoreOptions { turnTimeoutMs?: number; /** - * Endpoint-side gate consulted once per coalesced turn, after enrichment - * and before the handler. Unset is passthrough: every turn is delivered - * and the inbound path does no extra work. + * Endpoint-side gate consulted once per coalesced turn before the selected + * handler. The enriched path performs enrichment before the gate; the raw + * path deliberately does not enrich. Unset is passthrough. */ inboundInterceptor?: InboundInterceptor; } @@ -129,6 +130,20 @@ export type InboundHandler = ( msg: EnrichedInboundMessage, ) => Effect.Effect; +/** + * Handler for one coalesced raw inbound turn. The existing Message values are + * delivered in arrival order after the inbound interceptor admits the batch. + * Raw delivery does not enrich messages or read and commit presentation + * context. + */ +export type RawInboundHandler = ( + messages: readonly Message[], +) => Effect.Effect; + +type RegisteredInboundHandler = + | { readonly _tag: "enriched"; readonly handler: InboundHandler } + | { readonly _tag: "raw"; readonly handler: RawInboundHandler }; + /** * Verdict an {@link InboundInterceptor} returns for one coalesced turn. * @@ -145,10 +160,11 @@ export type InboundInterceptDecision = | { readonly _tag: "drop"; readonly reason?: string }; /** - * Gate the embedder installs between enrichment and the handler. It receives - * the newest message of the coalesced batch — a gate deciding whether a turn - * is still worth running cares about the latest thing said — and its verdict - * governs the whole batch. + * Gate the embedder installs between coalescing and the selected handler. The + * enriched handler path performs enrichment before consulting the gate; the + * raw handler path does not enrich. The gate receives the newest message of + * the coalesced batch — a gate deciding whether a turn is still worth running + * cares about the latest thing said — and its verdict governs the whole batch. * * It runs on the single consumer fiber, so suspending inside it delays this * turn and every message queued behind it, and `turnTimeoutMs` does not bound @@ -203,10 +219,12 @@ function runBackgroundLog(effect: Effect.Effect): void { } /** - * Wraps a `MoltZapService` with message enrichment, one-turn-at-a-time - * inbound delivery, and a send helper. One core per service — - * `getContextEntries()` is side-effectful (advances per-conversation - * markers), so a second core would consume entries the first expected. + * Wraps a `MoltZapService` with one-turn-at-a-time inbound delivery and a send + * helper. Adapter handlers receive enriched messages. A daemon handler may + * instead receive the admitted coalesced raw Message batch. One core per + * service — `getContextEntries()` is side-effectful (advances + * per-conversation markers), so a second core would consume entries the first + * expected. * * Turn-taking is entirely endpoint-local: the server delivers every message * it accepts, and this core decides when the runtime sees them. @@ -219,7 +237,8 @@ function runBackgroundLog(effect: Effect.Effect): void { * participant ws as MoltZapAgentClient * participant svc as MoltZapService * participant core as MoltZapChannelCore - * participant handler as InboundHandler + * participant raw as RawInboundHandler + * participant enriched as InboundHandler * * server->>ws: agent/message/received notification * ws->>svc: subscribers.dispatch — fanout(message) @@ -227,10 +246,17 @@ function runBackgroundLog(effect: Effect.Effect): void { * Note over core: dedup via recordMessageIdIfNew, drop when stopped or no handler is installed, then Queue.unsafeOffer(inboundQueue, message) * Note over core: consumer fiber — Queue.take * Note over core: takeCoalescedConversationMessages drains same-conv backlog into one turn - * Note over core: enrichMessage — sender name, conversation, context entries - * Note over core: inboundInterceptor — deliver or drop this turn - * core->>handler: inboundHandler(enriched) - * handler-->>core: Effect.void + * alt raw daemon handler + * Note over core: inboundInterceptor — deliver or drop this batch + * core->>raw: rawInboundHandler(messages) + * raw-->>core: Effect.void + * else enriched adapter handler + * Note over core: enrichMessage — sender name, conversation, context entries + * Note over core: inboundInterceptor — deliver or drop this turn + * core->>enriched: inboundHandler(enriched) + * enriched-->>core: Effect.void + * Note over core: completed handler commits presentation context + * end * Note over core: handler exceeds turnTimeoutMs — turn abandoned, drain continues * ``` * @@ -249,7 +275,7 @@ export class MoltZapChannelCore { // nothing drains — unbounded growth and silent non-delivery. Stopped means // stopped: inbound observed after disconnect() is dropped at the listener. private stopped = false; - private inboundHandler: InboundHandler | null = null; + private inboundHandler: RegisteredInboundHandler | null = null; /** * Inbound messages with an installed handler enqueue synchronously; a single @@ -329,7 +355,17 @@ export class MoltZapChannelCore { * @param handler Handler invoked for matching requests. */ onInbound(handler: InboundHandler): void { - this.inboundHandler = handler; + this.inboundHandler = { _tag: "enriched", handler }; + } + + /** + * Replaces any previous handler with a raw coalesced-turn handler. The + * admitted batch passes the interceptor and reaches the handler without + * enrichment or presentation-context commits. + * @param handler Handler invoked for each admitted raw batch. + */ + onRawInbound(handler: RawInboundHandler): void { + this.inboundHandler = { _tag: "raw", handler }; } onDisconnect(handler: () => void): void { @@ -425,8 +461,15 @@ export class MoltZapChannelCore { return Effect.gen( function* (this: MoltZapChannelCore) { const messages = yield* this.takeCoalescedConversationMessages(primary); - const handler = this.inboundHandler; - if (!handler) { + const registered = this.inboundHandler; + if (registered === null) { + return; + } + if (registered._tag === "raw") { + if (!(yield* this.interceptTurn(messages))) { + return; + } + yield* this.awaitHandlerTurn(registered.handler, messages, primary); return; } const { enriched, commitContext } = @@ -435,7 +478,7 @@ export class MoltZapChannelCore { return; } const completed = yield* this.awaitHandlerTurn( - handler, + registered.handler, enriched, primary, ); @@ -523,19 +566,19 @@ export class MoltZapChannelCore { /** * Await the user handler, bounded by `turnTimeoutMs` when it is set. * @param handler Installed inbound handler. - * @param enriched Enriched form of the coalesced batch. + * @param input Value delivered to the selected handler. * @param primary Message that opened this turn. * @returns Whether the handler ran to completion. */ - private awaitHandlerTurn( - handler: InboundHandler, - enriched: EnrichedInboundMessage, + private awaitHandlerTurn( + handler: (input: A) => Effect.Effect, + input: A, primary: Message, ): Effect.Effect { // The handler is user code returning an Effect — yield it directly so its // typed error channel propagates to the consumer fiber, which logs and // continues. Awaiting it inline preserves arrival-order delivery. - const turn = handler(enriched); + const turn = handler(input); const timeoutMs = this.turnTimeoutMs; if (timeoutMs === undefined) { return turn.pipe(Effect.as(true)); diff --git a/packages/client/src/harness-client.test.ts b/packages/client/src/harness-client.test.ts new file mode 100644 index 000000000..bfa924766 --- /dev/null +++ b/packages/client/src/harness-client.test.ts @@ -0,0 +1,314 @@ +/* eslint-disable agent-code-guard/async-keyword -- This loopback contract test hosts the Promise-native official MCP SDK. */ +import { Client } from "@modelcontextprotocol/client"; +import { + createMcpHandler, + fromJsonSchema, + McpServer, + type Implementation, + type JsonSchemaType, +} from "@modelcontextprotocol/server"; +import { Chunk, Effect, Exit, Fiber, Option, 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 { + acquireHarnessClient, + HarnessClient, + makeHarnessClientLayer, + type HarnessTurn, +} from "./harness-client.js"; +import { acquireHarnessMcpHttpServer } from "./harness-mcp-server.js"; +import { + decodeHarnessReplyRoute, + HARNESS_EVENTS_EXTENSION, + HARNESS_REPLY_TOOL, + harnessReplyInputJsonSchema, + harnessReplyResultJsonSchema, + type HarnessReplyInput, + type HarnessReplyResult, + type HarnessReplyRoute, + type HarnessTurnEvent, +} from "./harness/index.js"; +import { + makeHarnessMcpSubscriptionHandler, + type HarnessMcpSubscriptionHandler, +} from "./harness-mcp-subscription.js"; + +const SERVER_IMPLEMENTATION = { + name: "harness-client-test", + version: "1.0.0", +} satisfies Implementation; +const FIRST_CONVERSATION = conversationId( + "00000000-0000-4000-8000-000000000001", +); +const SECOND_CONVERSATION = conversationId( + "00000000-0000-4000-8000-000000000002", +); +const SENDER_ID = agentId("00000000-0000-4000-8000-000000000003"); + +const message = ( + id: string, + conversation: typeof FIRST_CONVERSATION, + text: string, +): Message => ({ + id: messageId(id), + conversationId: conversation, + senderId: SENDER_ID, + parts: [{ type: "text", text }], + createdAt: "2026-08-03T12:00:00.000Z", +}); + +const firstEvent = { + messages: [ + message( + "00000000-0000-4000-8000-000000000004", + FIRST_CONVERSATION, + "first", + ), + ], +} satisfies HarnessTurnEvent; +const secondEvent = { + messages: [ + message( + "00000000-0000-4000-8000-000000000005", + SECOND_CONVERSATION, + "second", + ), + ], +} satisfies HarnessTurnEvent; + +interface ObservedReply { + readonly input: HarnessReplyInput; + readonly route: HarnessReplyRoute; +} + +const replyInputSchema = fromJsonSchema( + /* Safe because Effect and MCP expose the same JSON Schema wire shape with different array mutability declarations. */ + harnessReplyInputJsonSchema as JsonSchemaType, +); +const replyResultSchema = fromJsonSchema( + /* Safe because Effect and MCP expose the same JSON Schema wire shape with different array mutability declarations. */ + harnessReplyResultJsonSchema as JsonSchemaType, +); + +const makeHarnessHandler = ( + observed: ObservedReply[], + advertiseExtension = true, +): HarnessMcpSubscriptionHandler => { + const delegate = createMcpHandler( + () => { + const server = new McpServer(SERVER_IMPLEMENTATION, { + capabilities: advertiseExtension + ? { extensions: { [HARNESS_EVENTS_EXTENSION]: {} } } + : {}, + }); + server.registerTool( + HARNESS_REPLY_TOOL, + { + inputSchema: replyInputSchema, + outputSchema: replyResultSchema, + }, + (input, context) => + Effect.runPromise( + decodeHarnessReplyRoute(context.mcpReq._meta).pipe( + Effect.map((route) => { + observed.push({ input, route }); + return { + content: [{ type: "text" as const, text: "{}" }], + structuredContent: {}, + }; + }), + ), + ), + ); + return server; + }, + { legacy: "reject" }, + ); + return makeHarnessMcpSubscriptionHandler({ + delegate, + implementation: SERVER_IMPLEMENTATION, + }); +}; + +const startHarnessServer = async ( + handler: HarnessMcpSubscriptionHandler, +) => { + const registration = createMcpHandler( + () => new McpServer(SERVER_IMPLEMENTATION), + { legacy: "reject" }, + ); + const scope = Effect.runSync(Scope.make()); + const server = await Effect.runPromise( + acquireHarnessMcpHttpServer({ + port: 0, + registrationHandler: registration, + harnessHandler: handler, + }).pipe(Scope.extend(scope)), + ); + const address = server.address(); + if (address === null || typeof address === "string") { + await Effect.runPromise(Scope.close(scope, Exit.void)); + throw new Error("expected a TCP test server address"); + } + return { + scope, + server, + url: new URL(`http://127.0.0.1:${address.port}/mcp`), + }; +}; + +const useHarness = ( + handler: HarnessMcpSubscriptionHandler, +): Effect.Effect => + Effect.gen(function* () { + const harness = yield* HarnessClient; + const receive = yield* harness.turns.pipe( + Stream.take(2), + Stream.runCollect, + Effect.fork, + ); + expect(handler.publish(firstEvent)).toBe(true); + expect(handler.publish(secondEvent)).toBe(true); + const turns = Chunk.toReadonlyArray(yield* Fiber.join(receive)); + const originatingTurn = turns[0]; + if (originatingTurn === undefined) { + throw new Error("expected the originating turn"); + } + yield* originatingTurn.reply("first reply"); + yield* originatingTurn.reply("second reply"); + return turns; + }); + +const preservesBoundConversation = async () => { + const observed: ObservedReply[] = []; + const handler = makeHarnessHandler(observed); + const running = await startHarnessServer(handler); + + try { + const turns = await Effect.runPromise( + useHarness(handler).pipe( + Effect.provide( + makeHarnessClientLayer({ + url: running.url.href, + }), + ), + ), + ); + expect(turns.map((turn) => turn.conversationId)).toEqual([ + 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 }, + }, + ]); + } finally { + await Effect.runPromise(Scope.close(running.scope, Exit.void)); + } +}; + +const rejectsMissingServerExtension = async () => { + const running = await startHarnessServer(makeHarnessHandler([], false)); + try { + await expect( + Effect.runPromise( + Effect.scoped(acquireHarnessClient({ url: running.url.href })), + ), + ).rejects.toThrow(HARNESS_EVENTS_EXTENSION); + } finally { + await Effect.runPromise(Scope.close(running.scope, Exit.void)); + } +}; + +const rejectsUnexpectedTurnFields = async () => { + const handler = makeHarnessHandler([]); + const running = await startHarnessServer(handler); + try { + const nextTurn = Effect.scoped( + Effect.gen(function* () { + const harness = yield* acquireHarnessClient({ + url: running.url.href, + }); + const next = yield* harness.turns.pipe(Stream.runHead, Effect.fork); + const eventWithExtraField = { ...firstEvent, invented: true }; + expect(handler.publish(eventWithExtraField)).toBe(true); + return yield* Fiber.join(next); + }), + ); + await expect(Effect.runPromise(nextTurn)).rejects.toBeDefined(); + } finally { + await Effect.runPromise(Scope.close(running.scope, Exit.void)); + } +}; + +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 }, + ); + }); + }); + try { + const harness = await Effect.runPromise( + acquireHarnessClient({ url: running.url.href }).pipe( + Scope.extend(clientScope), + ), + ); + const received = Effect.runPromise(harness.turns.pipe(Stream.runHead)); + expect(handler.publish(firstEvent)).toBe(true); + const turn = Option.getOrThrowWith( + await received, + () => new Error("expected a harness turn"), + ); + const reply = Effect.runFork(turn.reply("cancel me")); + await vi.waitFor(() => { + expect(callTool).toHaveBeenCalledOnce(); + }); + await Effect.runPromise(Fiber.interrupt(reply)); + expect(observedSignal?.aborted).toBe(true); + } finally { + callTool.mockRestore(); + await Effect.runPromise(Scope.close(clientScope, Exit.void)); + await Effect.runPromise(Scope.close(running.scope, Exit.void)); + } +}; + +// @agent-code-guard/regression-only: the scoped loopback boundary pins every reply closure to its originating turn without suppression. +describe("HarnessClient", () => { + it("sends every reply through the originating conversation after later turns", () => + preservesBoundConversation()); + it("rejects a server without the harness events extension", () => + rejectsMissingServerExtension()); + it("rejects unexpected fields after removing MCP notification metadata", () => + rejectsUnexpectedTurnFields()); + it("aborts an in-flight reply call when its Effect is interrupted", () => + abortsReplyCallWhenInterrupted()); +}); + +/* eslint-enable agent-code-guard/async-keyword -- Restore strict defaults after the Promise-native interoperability fixture. */ diff --git a/packages/client/src/harness-client.ts b/packages/client/src/harness-client.ts new file mode 100644 index 000000000..81e3d068d --- /dev/null +++ b/packages/client/src/harness-client.ts @@ -0,0 +1,55 @@ +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[]]; + /** 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 { + /** The sole receive stream owned by this scoped client. */ + readonly turns: Stream.Stream; +} + +/** Effect service tag consumed by runtime adapters. */ +export class HarnessClient extends Context.Tag("@moltzap/client/HarnessClient")< + HarnessClient, + HarnessClientService +>() {} + +/** Inputs needed to connect one scoped harness client. */ +export interface HarnessClientOptions { + /** Loopback `POST /mcp` endpoint owned by one running `moltzapd`. */ + readonly url: string; +} + +/** + * Acquires one turn-ready harness connection and receive stream for the + * lifetime of the enclosing scope. The private adapter owns MCP translation. + * + * @param options Fixed loopback MCP endpoint. + * @returns The scoped adapter-facing service value. + */ +export const acquireHarnessClient = ( + options: HarnessClientOptions, +): Effect.Effect => + acquireHarnessClientInternal(options); + +/** + * Builds the scoped runtime-adapter layer for one daemon endpoint. + * + * @param options Fixed loopback MCP endpoint. + * @returns A Layer providing the scoped HarnessClient capability. + */ +export const makeHarnessClientLayer = ( + options: HarnessClientOptions, +): Layer.Layer => + Layer.scoped(HarnessClient, acquireHarnessClient(options)); diff --git a/packages/client/src/harness-mcp-server.test.ts b/packages/client/src/harness-mcp-server.test.ts index 1dfbbdeae..6f4532193 100644 --- a/packages/client/src/harness-mcp-server.test.ts +++ b/packages/client/src/harness-mcp-server.test.ts @@ -4,20 +4,31 @@ import { StreamableHTTPClientTransport, } from "@modelcontextprotocol/client"; import { + CLIENT_CAPABILITIES_META_KEY, + CLIENT_INFO_META_KEY, + PROTOCOL_VERSION_META_KEY, + SERVER_INFO_META_KEY, + SUBSCRIPTION_ID_META_KEY, createMcpHandler, McpServer, type Implementation, type McpHttpHandler, } from "@modelcontextprotocol/server"; // eslint-disable-next-line agent-code-guard/prefer-effect-platform -- These loopback contract tests require raw Host headers and connection-refusal assertions that the Effect client does not expose. -import { request as nodeRequest } from "node:http"; -import { Cause, Effect, Exit, Fiber, Scope } from "effect"; +import { + Agent as NodeHttpAgent, + request as nodeRequest, + type IncomingMessage, +} from "node:http"; +import { Cause, Duration, Effect, Exit, Fiber, Option, Scope } from "effect"; import { afterEach, describe, expect, it, vi } from "vitest"; import { agentId } 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"; import { makeLocalDaemonHandlers } from "./service-local-daemon.js"; import { acquireHarnessMcpHttpServer } from "./harness-mcp-server.js"; +import { makeHarnessMcpSubscriptionHandler } from "./harness-mcp-subscription.js"; const LOCALHOST = "127.0.0.1"; const MODERN_PROTOCOL_VERSION = "2026-07-28"; @@ -29,6 +40,10 @@ const FORBIDDEN_STATUS = 403; const NOT_FOUND_STATUS = 404; const METHOD_NOT_ALLOWED_STATUS = 405; const GRACEFUL_SUBSCRIPTION_CLOSE = "graceful"; +const SUBSCRIPTIONS_LISTEN_METHOD = "subscriptions/listen"; +const BACKPRESSURE_PAYLOAD_BYTES = 8 * 1024 * 1024; +const BACKPRESSURE_WRITE_DELAY = Duration.millis(100); +const BACKPRESSURE_CLOSE_DEADLINE = Duration.seconds(3); const SERVER_IMPLEMENTATION = { name: "harness-boundary-test", version: "1.0.0", @@ -322,7 +337,181 @@ const closesActiveSubscriptionWhenScopeReleases = async () => { expect(running.server.listening).toBe(false); }; -const exposesOnlyExistingStatusBehavior = async () => { +const makeSubscriptionHarnessHandlers = () => { + const ownAgentId = agentId("550e8400-e29b-41d4-a716-446655440041"); + const localHandlers = makeLocalDaemonHandlers({ + ownAgentId, + connected: () => true, + conversationCount: () => 0, + call: () => { + throw new Error("subscription must not call an agent RPC"); + }, + handleHistoryRequest: () => { + throw new Error("subscription must not read local history"); + }, + }); + return makeHarnessMcpHttpHandlers({ + implementation: SERVER_IMPLEMENTATION, + reply: () => Effect.void, + status: localHandlers[localDaemonCommands.status], + }); +}; + +const makeListenPayload = (listenId: string): string => + JSON.stringify({ + jsonrpc: "2.0", + id: listenId, + method: SUBSCRIPTIONS_LISTEN_METHOD, + params: { + notifications: { "xyz.moltzap/turnReady": true }, + _meta: { + [PROTOCOL_VERSION_META_KEY]: MODERN_PROTOCOL_VERSION, + [CLIENT_INFO_META_KEY]: { + name: "slow-reader-test", + version: "1.0.0", + }, + [CLIENT_CAPABILITIES_META_KEY]: { + extensions: { [HARNESS_EVENTS_EXTENSION]: {} }, + }, + }, + }, + }); + +interface PausedSubscriptionResponse { + readonly agent: NodeHttpAgent; + readonly body: () => string; + readonly ended: Promise; + readonly response: IncomingMessage; +} + +const openPausedSubscription = async (baseUrl: string, payload: string) => { + const agent = new NodeHttpAgent({ keepAlive: true }); + let responseBody = ""; + const response = await new Promise((resolve, reject) => { + const request = nodeRequest( + new URL(HARNESS_MCP_PATH, baseUrl), + { + agent, + headers: { + "content-length": Buffer.byteLength(payload), + "content-type": "application/json", + "mcp-method": SUBSCRIPTIONS_LISTEN_METHOD, + "mcp-protocol-version": MODERN_PROTOCOL_VERSION, + }, + method: POST_METHOD, + }, + (incoming) => { + incoming.pause(); + incoming.on("data", (chunk: Buffer) => { + responseBody += chunk.toString("utf8"); + }); + resolve(incoming); + }, + ); + request.once("error", reject); + request.end(payload); + }); + const ended = new Promise((resolve, reject) => { + response.once("end", () => { + resolve(undefined); + }); + response.once("aborted", () => { + reject(new Error("response aborted")); + }); + response.once("error", reject); + }); + return { + agent, + body: () => responseBody, + ended, + response, + } satisfies PausedSubscriptionResponse; +}; + +const parseDataFrames = (responseBody: string): readonly unknown[] => + responseBody + .split("\n\n") + .filter((frame) => frame.startsWith("data: ")) + .map((frame) => { + const parsed: unknown = JSON.parse(frame.slice("data: ".length)); + return parsed; + }); + +const closesAfterSlowReaderObservesTerminalCompletion = async () => { + const handlers = makeSubscriptionHarnessHandlers(); + const activeClose = vi.spyOn(handlers.active, "close"); + const running = await acquireServerWithHandlers( + handlers.registration, + handlers.active, + ); + const listenId = "slow-reader"; + const subscription = await openPausedSubscription( + running.baseUrl, + makeListenPayload(listenId), + ); + + const released = releaseServerScope(running.scope); + await vi.waitFor(() => { + expect(activeClose).toHaveBeenCalledOnce(); + }); + subscription.response.resume(); + await subscription.ended; + await released; + subscription.agent.destroy(); + + const frames = parseDataFrames(subscription.body()); + expect(frames).toContainEqual({ + jsonrpc: "2.0", + id: listenId, + result: { + resultType: "complete", + _meta: { + [SUBSCRIPTION_ID_META_KEY]: listenId, + [SERVER_INFO_META_KEY]: SERVER_IMPLEMENTATION, + }, + }, + }); + expect(running.server.listening).toBe(false); +}; + +const closesDespiteBackpressuredReader = async () => { + const active = makeHarnessMcpSubscriptionHandler<{ readonly opaque: string }>( + { + delegate: makeHandler("harness-backpressure-test"), + implementation: SERVER_IMPLEMENTATION, + }, + ); + const running = await acquireServerWithHandlers( + makeHandler("registration-backpressure-test"), + active, + ); + const subscription = await openPausedSubscription( + running.baseUrl, + makeListenPayload("backpressured-reader"), + ); + const ended = subscription.ended.catch(() => undefined); + + expect( + active.publish({ opaque: "x".repeat(BACKPRESSURE_PAYLOAD_BYTES) }), + ).toBe(true); + await Effect.runPromise(Effect.sleep(BACKPRESSURE_WRITE_DELAY)); + + openServerScopes.delete(running.scope); + const closing = Effect.runFork(Scope.close(running.scope, Exit.void)); + const closedWithinDeadline = await Effect.runPromise( + Fiber.join(closing).pipe(Effect.timeoutOption(BACKPRESSURE_CLOSE_DEADLINE)), + ); + + subscription.response.destroy(); + subscription.agent.destroy(); + await Effect.runPromise(Fiber.join(closing)); + await ended; + + expect(Option.isSome(closedWithinDeadline)).toBe(true); + expect(running.server.listening).toBe(false); +}; + +const exposesStatusAndReplyTools = async () => { const ownAgentId = agentId("550e8400-e29b-41d4-a716-446655440040"); const localHandlers = makeLocalDaemonHandlers({ ownAgentId, @@ -337,6 +526,7 @@ const exposesOnlyExistingStatusBehavior = async () => { }); const handlers = makeHarnessMcpHttpHandlers({ implementation: SERVER_IMPLEMENTATION, + reply: () => Effect.void, status: localHandlers[localDaemonCommands.status], }); const baseUrl = await makeServerWithHandlers( @@ -351,9 +541,12 @@ const exposesOnlyExistingStatusBehavior = 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"]); + ).toEqual(["status", "reply"]); const result = await harnessClient.callTool({ name: "status", @@ -386,9 +579,13 @@ describe("scoped Harness MCP HTTP server", () => { closesHandlersWhenListenerBindFails()); it("closes an active MCP subscription when its scope releases", () => closesActiveSubscriptionWhenScopeReleases()); + it("lets a slow reader observe terminal completion before prompt shutdown", () => + closesAfterSlowReaderObservesTerminalCompletion()); + it("bounds shutdown when an MCP reader stops draining its response", () => + closesDespiteBackpressuredReader()); it( - "serves only the existing status behavior through the active catalog", - exposesOnlyExistingStatusBehavior, + "serves status and reply through the active catalog", + exposesStatusAndReplyTools, ); }); diff --git a/packages/client/src/harness-mcp-server.ts b/packages/client/src/harness-mcp-server.ts index 176279958..61ae88252 100644 --- a/packages/client/src/harness-mcp-server.ts +++ b/packages/client/src/harness-mcp-server.ts @@ -12,12 +12,24 @@ import { type RequestListener, type Server as NodeHttpServer, } from "node:http"; -import { Effect, Exit, type Scope } from "effect"; +import type { Socket } from "node:net"; +import { + Deferred, + Duration, + Effect, + Exit, + Fiber, + Option, + type Scope, +} from "effect"; const REGISTER_MCP_PATH = "/register/mcp"; const HARNESS_MCP_PATH = "/mcp"; const POST_METHOD = "POST"; const LOOPBACK_HOST = "127.0.0.1"; +// Loopback readers normally drain immediately; one second tolerates scheduler +// delay without allowing a stopped reader to retain the daemon lifetime. +const RESPONSE_DRAIN_GRACE_PERIOD = Duration.seconds(1); interface HarnessMcpHttpServerOptions { readonly port: number; @@ -25,6 +37,25 @@ interface HarnessMcpHttpServerOptions { readonly harnessHandler: McpHttpHandler; } +interface HarnessMcpRequestListener { + readonly listener: RequestListener; + readonly waitForResponses: Effect.Effect; +} + +interface RunningHarnessMcpHttpServer { + readonly server: NodeHttpServer; + readonly destroyConnections: () => void; + readonly waitForResponses: Effect.Effect; +} + +interface ResponseTracker { + readonly track: ( + response: ReturnType, + nodeResponse: Parameters[1], + ) => void; + readonly waitForResponses: Effect.Effect; +} + const respond = ( status: number, body: string, @@ -38,6 +69,56 @@ const respond = ( response.end(body); }; +const makeResponseTracker = (): ResponseTracker => { + const inFlight = new Set>(); + const responseWaiters = new Set<() => void>(); + const finish = (response: ReturnType): void => { + inFlight.delete(response); + if (inFlight.size === 0) { + for (const resume of responseWaiters) { + resume(); + } + responseWaiters.clear(); + } + }; + + return { + track: (response, nodeResponse) => { + inFlight.add(response); + Effect.runFork( + Effect.tryPromise({ + try: () => response, + catch: (error) => error, + }).pipe( + Effect.catchAll((error) => + Effect.sync(() => { + nodeResponse.destroy(error instanceof Error ? error : undefined); + }), + ), + Effect.ensuring( + Effect.sync(() => { + finish(response); + }), + ), + ), + ); + }, + waitForResponses: Effect.async((resume) => { + if (inFlight.size === 0) { + resume(Effect.succeed(undefined)); + return; + } + const done = (): void => { + resume(Effect.succeed(undefined)); + }; + responseWaiters.add(done); + return Effect.sync(() => { + responseWaiters.delete(done); + }); + }), + }; +}; + /** * Routes the daemon's two loopback MCP surfaces through one Node listener. * @@ -48,17 +129,18 @@ const respond = ( const makeHarnessMcpRequestListener = ( registrationHandler: FetchLikeMcpHandler, harnessHandler: FetchLikeMcpHandler, -): RequestListener => { +): HarnessMcpRequestListener => { const validateHost = localhostHostValidation(); const validateOrigin = localhostOriginValidation(); const registrationNodeHandler = toNodeHandler(registrationHandler); const harnessNodeHandler = toNodeHandler(harnessHandler); + const responses = makeResponseTracker(); const handlers: ReadonlyMap = new Map([ [REGISTER_MCP_PATH, registrationNodeHandler], [HARNESS_MCP_PATH, harnessNodeHandler], ]); - return (request, response): void => { + const listener: RequestListener = (request, response): void => { if ( !validateHost(request, response) || !validateOrigin(request, response) @@ -79,29 +161,54 @@ const makeHarnessMcpRequestListener = ( return; } - handler(request, response).catch((error: unknown) => { - response.destroy(error instanceof Error ? error : undefined); + responses.track(handler(request, response), response); + }; + + return { + listener, + waitForResponses: responses.waitForResponses, + }; +}; + +const trackConnections = (server: NodeHttpServer): (() => void) => { + const sockets = new Set(); + server.on("connection", (socket) => { + sockets.add(socket); + socket.once("close", () => { + sockets.delete(socket); }); + }); + return () => { + for (const socket of sockets) { + socket.destroy(); + } + server.closeAllConnections(); }; }; const listen = ( options: HarnessMcpHttpServerOptions, -): Effect.Effect => - Effect.async((resume) => { - const server = createServer( - makeHarnessMcpRequestListener( - options.registrationHandler, - options.harnessHandler, - ), +): Effect.Effect => + Effect.async((resume) => { + const requests = makeHarnessMcpRequestListener( + options.registrationHandler, + options.harnessHandler, ); + const server = createServer(requests.listener); + const destroyConnections = trackConnections(server); const onError = (error: Error): void => { resume(Effect.fail(error)); }; server.once("error", onError); server.listen(options.port, LOOPBACK_HOST, () => { server.off("error", onError); - resume(Effect.succeed(server)); + resume( + Effect.succeed({ + destroyConnections, + server, + waitForResponses: requests.waitForResponses, + }), + ); }); return Effect.async((cancelResume) => { server.off("error", onError); @@ -123,22 +230,26 @@ const listen = ( }); }); -const close = (server: NodeHttpServer): Effect.Effect => - Effect.async((resume) => { - if (!server.listening) { - resume(Effect.succeed(undefined)); - return; - } - server.close((error) => { - resume( +const beginClose = ( + server: NodeHttpServer, +): Effect.Effect> => + Effect.gen(function* () { + const completion = yield* Deferred.make(); + yield* Effect.sync(() => { + server.close((error) => { + Effect.runSync(Deferred.succeed(completion, error)); + }); + }); + return Deferred.await(completion).pipe( + Effect.flatMap((error) => error === undefined ? Effect.succeed(undefined) : Effect.logWarning( "Harness MCP HTTP server close failed", error, ).pipe(Effect.as(undefined)), - ); - }); + ), + ); }); const closeHandler = (handler: McpHttpHandler): Effect.Effect => @@ -164,10 +275,40 @@ const closeHandlers = ( ); const release = ( - server: NodeHttpServer, + running: RunningHarnessMcpHttpServer, options: HarnessMcpHttpServerOptions, -): Effect.Effect => - closeHandlers(options).pipe(Effect.zipRight(close(server))); +): Effect.Effect => + Effect.gen(function* () { + const closeCompletion = yield* beginClose(running.server); + // Handler closure finishes the Web stream before the Node adapter drains + // it. A local client that stops reading must not retain the daemon scope, + // so graceful draining is bounded before active connections are closed. + const gracefulClose = yield* Effect.fork( + Effect.all([closeHandlers(options), running.waitForResponses], { + concurrency: 2, + discard: true, + }), + ); + // acquireRelease finalizers are uninterruptible. Only this observation is + // interruptible so the grace timer can win; the close fiber remains owned + // here and is joined after any forced connection shutdown. + const drained = yield* Fiber.join(gracefulClose).pipe( + Effect.timeoutOption(RESPONSE_DRAIN_GRACE_PERIOD), + Effect.interruptible, + ); + if (Option.isNone(drained)) { + yield* Effect.logWarning( + "Harness MCP response drain timed out; closing active connections", + ); + yield* Effect.sync(running.destroyConnections); + } else { + running.server.closeAllConnections(); + } + if (Option.isNone(drained)) { + yield* Fiber.join(gracefulClose); + } + yield* closeCompletion; + }); /** * Acquires the guarded loopback HTTP server for one explicitly supplied port. @@ -187,5 +328,5 @@ export const acquireHarnessMcpHttpServer = ( Exit.isSuccess(exit) ? Effect.void : closeHandlers(options), ), ), - (server) => release(server, options), - ); + (running) => release(running, options), + ).pipe(Effect.map((running) => running.server)); diff --git a/packages/client/src/harness-mcp-subscription.test.ts b/packages/client/src/harness-mcp-subscription.test.ts new file mode 100644 index 000000000..07a01a9f1 --- /dev/null +++ b/packages/client/src/harness-mcp-subscription.test.ts @@ -0,0 +1,453 @@ +/* eslint-disable agent-code-guard/async-keyword -- These contract tests exercise the official Promise-native MCP client, handler, and retained response stream. */ +import { + Client, + fromJsonSchema, + StreamableHTTPClientTransport, + type JsonSchemaType, + type SubscriptionFilter, +} from "@modelcontextprotocol/client"; +import { + CLIENT_CAPABILITIES_META_KEY, + CLIENT_INFO_META_KEY, + PROTOCOL_VERSION_META_KEY, + SERVER_INFO_META_KEY, + SUBSCRIPTION_ID_META_KEY, + createMcpHandler, + McpServer, + type Implementation, +} from "@modelcontextprotocol/server"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + makeHarnessMcpSubscriptionHandler, + type HarnessMcpSubscriptionHandler, +} from "./harness-mcp-subscription.js"; +import { + HARNESS_EVENTS_EXTENSION, + HARNESS_TURN_READY_FILTER, + HARNESS_TURN_READY_NOTIFICATION, +} from "./harness/index.js"; + +const SERVER_IMPLEMENTATION = { + name: "harness-subscription-test", + version: "1.0.0", +} satisfies Implementation; +const SUBSCRIPTIONS_LISTEN_METHOD = "subscriptions/listen"; +const SUBSCRIPTIONS_ACKNOWLEDGED_NOTIFICATION = + "notifications/subscriptions/acknowledged"; +const MODERN_PROTOCOL_VERSION = "2026-07-28"; +const DATA_FIELD = "data: "; +const EVENT_FIELD = "event:"; +const ID_FIELD = "id:"; +const RETRY_FIELD = "retry:"; +const FRAME_END = "\n\n"; +const SSE_CONTENT_TYPE = "text/event-stream"; +const OK_STATUS = 200; +const BAD_REQUEST_STATUS = 400; +const CONFLICT_STATUS = 409; +const METHOD_NOT_ALLOWED_STATUS = 405; +const LOCAL_SUBSCRIPTION_CLOSE = "local"; +const GRACEFUL_SUBSCRIPTION_CLOSE = "graceful"; +const TURN_READY_FILTER: SubscriptionFilter & { + readonly [HARNESS_TURN_READY_FILTER]: true; +} = { + [HARNESS_TURN_READY_FILTER]: true, +}; +const opaquePayloadSchema = fromJsonSchema({ + type: "object", + properties: { + ordinal: { type: "number" }, + snapshot: { + type: "object", + properties: { opaque: { type: "string" } }, + required: ["opaque"], + }, + }, + required: ["ordinal", "snapshot"], + additionalProperties: true, +} satisfies JsonSchemaType); + +interface OpaquePayload { + readonly ordinal: number; + readonly snapshot: { + readonly opaque: string; + }; +} + +const openHandlers = new Set>(); + +const makeHandler = () => { + const delegate = createMcpHandler( + () => new McpServer(SERVER_IMPLEMENTATION), + { legacy: "reject" }, + ); + const handler = makeHarnessMcpSubscriptionHandler({ + delegate, + implementation: SERVER_IMPLEMENTATION, + }); + openHandlers.add(handler); + return { delegate, handler }; +}; + +const getReader = ( + response: Response, +): ReadableStreamDefaultReader => { + const reader = response.body?.getReader(); + if (reader === undefined) { + throw new Error("expected retained SSE response body"); + } + return reader; +}; + +interface ListenRequestOptions { + readonly capability?: boolean; + readonly signal?: AbortSignal; + readonly notifications?: Readonly>; +} + +const makeListenRequest = ( + id: string | number, + options: ListenRequestOptions = {}, +): Request => { + const capabilities = + options.capability === false + ? {} + : { extensions: { [HARNESS_EVENTS_EXTENSION]: {} } }; + return new Request("http://127.0.0.1/mcp", { + method: "POST", + headers: { + "content-type": "application/json", + "mcp-method": SUBSCRIPTIONS_LISTEN_METHOD, + }, + body: JSON.stringify({ + jsonrpc: "2.0", + id, + method: SUBSCRIPTIONS_LISTEN_METHOD, + params: { + notifications: options.notifications ?? { + [HARNESS_TURN_READY_FILTER]: true, + }, + _meta: { + [PROTOCOL_VERSION_META_KEY]: MODERN_PROTOCOL_VERSION, + [CLIENT_INFO_META_KEY]: { + name: "harness-subscription-client", + version: "1.0.0", + }, + [CLIENT_CAPABILITIES_META_KEY]: capabilities, + }, + }, + }), + signal: options.signal, + }); +}; + +const readFrame = async (reader: ReadableStreamDefaultReader) => { + const result = await reader.read(); + expect(result.done).toBe(false); + const frame = new TextDecoder().decode(result.value); + expect(frame.startsWith(DATA_FIELD)).toBe(true); + expect(frame.endsWith(FRAME_END)).toBe(true); + expect(frame).not.toContain(EVENT_FIELD); + expect(frame).not.toContain(ID_FIELD); + expect(frame).not.toContain(RETRY_FIELD); + const parsed: unknown = JSON.parse( + frame.slice(DATA_FIELD.length, -FRAME_END.length), + ); + if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) { + throw new Error("expected JSON-RPC object frame"); + } + return parsed; +}; + +const makeOfficialClient = async ( + handler: HarnessMcpSubscriptionHandler, + received: OpaquePayload[], +) => { + const client = new Client( + { name: "harness-subscription-client", version: "1.0.0" }, + { + capabilities: { extensions: { [HARNESS_EVENTS_EXTENSION]: {} } }, + versionNegotiation: { mode: { pin: MODERN_PROTOCOL_VERSION } }, + }, + ); + client.setNotificationHandler( + HARNESS_TURN_READY_NOTIFICATION, + { params: opaquePayloadSchema }, + (payload) => { + received.push(payload); + }, + ); + const transport = new StreamableHTTPClientTransport( + new URL("http://127.0.0.1/mcp"), + { fetch: (url, init) => handler.fetch(new Request(url, init)) }, + ); + await client.connect(transport); + return client; +}; + +afterEach(async () => { + for (const handler of openHandlers) { + await handler.close(); + } + openHandlers.clear(); +}); + +// @agent-code-guard/regression-only: this finite matrix pins the retained SSE wire shape and official beta.5 client interoperability. +describe("Harness MCP subscription delegation", () => { + it("delegates ordinary MCP requests to the official handler", async () => { + const { delegate, handler } = makeHandler(); + const delegateFetch = vi.spyOn(delegate, "fetch"); + const response = await handler.fetch( + new Request("http://127.0.0.1/mcp", { method: "GET" }), + ); + + expect(delegateFetch).toHaveBeenCalledOnce(); + expect(response.status).toBe(METHOD_NOT_ALLOWED_STATUS); + expect(handler.notify).toBe(delegate.notify); + expect(handler.bus).toBe(delegate.bus); + }); + + it("leaves non-custom subscription filters with the official handler", async () => { + const { delegate, handler } = makeHandler(); + const delegateFetch = vi.spyOn(delegate, "fetch"); + const response = await handler.fetch( + makeListenRequest("standard", { + notifications: { toolsListChanged: true }, + }), + ); + + expect(delegateFetch).toHaveBeenCalledOnce(); + await response.body?.cancel(); + }); +}); + +describe("Harness MCP subscription filter isolation", () => { + it("leaves mixed extension and standard filters with the official handler", async () => { + const { delegate, handler } = makeHandler(); + const delegateFetch = vi.spyOn(delegate, "fetch"); + const response = await handler.fetch( + makeListenRequest("mixed", { + notifications: { + [HARNESS_TURN_READY_FILTER]: true, + toolsListChanged: true, + }, + }), + ); + + expect(delegateFetch).toHaveBeenCalledOnce(); + await response.body?.cancel(); + }); + + it("leaves malformed additional filters with the official handler", async () => { + const { delegate, handler } = makeHandler(); + const delegateFetch = vi.spyOn(delegate, "fetch"); + const response = await handler.fetch( + makeListenRequest("malformed-additional", { + notifications: { + [HARNESS_TURN_READY_FILTER]: true, + toolsListChanged: "invalid", + }, + }), + ); + + expect(delegateFetch).toHaveBeenCalledOnce(); + await response.body?.cancel(); + }); +}); + +describe("Harness MCP subscription parsed-body isolation", () => { + it("delegates when the extension filter is inherited", async () => { + const { delegate, handler } = makeHandler(); + const delegateFetch = vi.spyOn(delegate, "fetch"); + class NotificationsWithInheritedExtension { + readonly toolsListChanged = true; + + get [HARNESS_TURN_READY_FILTER](): true { + return true; + } + } + const notifications = new NotificationsWithInheritedExtension(); + const response = await handler.fetch(makeListenRequest("prototype"), { + parsedBody: { + jsonrpc: "2.0", + id: "prototype", + method: SUBSCRIPTIONS_LISTEN_METHOD, + params: { + notifications, + _meta: { + [PROTOCOL_VERSION_META_KEY]: MODERN_PROTOCOL_VERSION, + [CLIENT_INFO_META_KEY]: { + name: "harness-subscription-client", + version: "1.0.0", + }, + [CLIENT_CAPABILITIES_META_KEY]: { + extensions: { [HARNESS_EVENTS_EXTENSION]: {} }, + }, + }, + }, + }, + }); + + expect(delegateFetch).toHaveBeenCalledOnce(); + await response.body?.cancel(); + }); +}); + +describe("Harness MCP subscription framing", () => { + it("acknowledges first and publishes complete typed notification frames", async () => { + const { handler } = makeHandler(); + const response = await handler.fetch(makeListenRequest("listen-7")); + const reader = getReader(response); + + expect(response.status).toBe(OK_STATUS); + expect(response.headers.get("content-type")).toBe(SSE_CONTENT_TYPE); + expect(await readFrame(reader)).toEqual({ + jsonrpc: "2.0", + method: SUBSCRIPTIONS_ACKNOWLEDGED_NOTIFICATION, + params: { + notifications: { [HARNESS_TURN_READY_FILTER]: true }, + _meta: { [SUBSCRIPTION_ID_META_KEY]: "listen-7" }, + }, + }); + + expect( + handler.publish({ ordinal: 3, snapshot: { opaque: "caller-owned" } }), + ).toBe(true); + expect(await readFrame(reader)).toEqual({ + jsonrpc: "2.0", + method: HARNESS_TURN_READY_NOTIFICATION, + params: { + ordinal: 3, + snapshot: { opaque: "caller-owned" }, + _meta: { [SUBSCRIPTION_ID_META_KEY]: "listen-7" }, + }, + }); + await reader.cancel(); + }); +}); + +describe("Harness MCP subscription admission", () => { + it("atomically refuses a racing turn-ready listener", async () => { + const { handler } = makeHandler(); + const first = await handler.fetch(makeListenRequest(1)); + const raced = await handler.fetch(makeListenRequest(2)); + + expect(raced.status).toBe(CONFLICT_STATUS); + expect(await raced.json()).toEqual({ + jsonrpc: "2.0", + error: { + code: -32000, + message: "Turn-ready subscription already in use", + data: { kind: "subscription_in_use" }, + }, + id: 2, + }); + + await first.body?.cancel(); + const replacement = await handler.fetch(makeListenRequest(3)); + expect(replacement.status).toBe(OK_STATUS); + await replacement.body?.cancel(); + }); + + it("uses the core missing-capability error before acquiring SSE", async () => { + const { handler } = makeHandler(); + const response = await handler.fetch( + makeListenRequest("missing-capability", { capability: false }), + ); + + expect(response.status).toBe(BAD_REQUEST_STATUS); + expect(await response.json()).toEqual({ + jsonrpc: "2.0", + error: { + code: -32021, + message: "Missing required client capabilities: extensions", + data: { + requiredCapabilities: { + extensions: { [HARNESS_EVENTS_EXTENSION]: {} }, + }, + }, + }, + id: "missing-capability", + }); + }); +}); + +describe("Harness MCP subscription disconnect", () => { + it("cleans up a disconnected listener without a terminal frame", async () => { + const { handler } = makeHandler(); + const abort = new AbortController(); + const response = await handler.fetch( + makeListenRequest("disconnect", { signal: abort.signal }), + ); + const reader = getReader(response); + await readFrame(reader); + + abort.abort(); + expect(await reader.read()).toEqual({ done: true, value: undefined }); + expect(handler.publish({ ordinal: 4, snapshot: { opaque: "late" } })).toBe( + false, + ); + + const replacement = await handler.fetch(makeListenRequest("replacement")); + expect(replacement.status).toBe(OK_STATUS); + await replacement.body?.cancel(); + }); +}); + +describe("Harness MCP subscription graceful close", () => { + it("gracefully completes the retained response when the handler closes", async () => { + const { handler } = makeHandler(); + const response = await handler.fetch(makeListenRequest("close-me")); + const reader = getReader(response); + await readFrame(reader); + + await handler.close(); + expect(await readFrame(reader)).toEqual({ + jsonrpc: "2.0", + id: "close-me", + result: { + resultType: "complete", + _meta: { + [SUBSCRIPTION_ID_META_KEY]: "close-me", + [SERVER_INFO_META_KEY]: SERVER_IMPLEMENTATION, + }, + }, + }); + expect(await reader.read()).toEqual({ done: true, value: undefined }); + expect(handler.publish({ ordinal: 5, snapshot: { opaque: "late" } })).toBe( + false, + ); + }); +}); + +describe("Harness MCP subscription official client interop", () => { + it("interoperates with the official client listen lifecycle", async () => { + const { handler } = makeHandler(); + const received: OpaquePayload[] = []; + const client = await makeOfficialClient(handler, received); + + try { + const first = await client.listen(TURN_READY_FILTER); + expect(first).toBeDefined(); + + const payload = { + ordinal: 8, + snapshot: { opaque: "official-client" }, + } satisfies OpaquePayload; + expect(handler.publish(payload)).toBe(true); + await vi.waitFor(() => { + expect(received).toHaveLength(1); + expect(received[0]).toMatchObject(payload); + }); + + await first.close(); + expect(await first.closed).toBe(LOCAL_SUBSCRIPTION_CLOSE); + const replacement = await client.listen(TURN_READY_FILTER); + await handler.close(); + expect(await replacement.closed).toBe(GRACEFUL_SUBSCRIPTION_CLOSE); + } finally { + await client.close(); + } + }); +}); + +/* eslint-enable agent-code-guard/async-keyword -- Restore repository defaults after the Promise-native MCP contract tests. */ diff --git a/packages/client/src/harness-mcp-subscription.ts b/packages/client/src/harness-mcp-subscription.ts new file mode 100644 index 000000000..8a4d42b56 --- /dev/null +++ b/packages/client/src/harness-mcp-subscription.ts @@ -0,0 +1,441 @@ +/* eslint-disable agent-code-guard/async-keyword, agent-code-guard/promise-type, @typescript-eslint/no-invalid-void-type -- The official MCP handler and retained POST response stream expose Promise-native lifecycle contracts. */ +import { + CLIENT_CAPABILITIES_META_KEY, + MissingRequiredClientCapabilityError, + SERVER_INFO_META_KEY, + SUBSCRIPTION_ID_META_KEY, + classifyInboundRequest, + isJsonContentType, + type Implementation, + type McpHandlerRequestOptions, + type McpHttpHandler, + type RequestId, +} from "@modelcontextprotocol/server"; +import { + HARNESS_EVENTS_EXTENSION, + HARNESS_TURN_READY_FILTER, + HARNESS_TURN_READY_NOTIFICATION, +} from "./harness/index.js"; + +// The SDK remains authoritative for the MCP server. Its public event publisher +// has a closed event union, so this adapter owns only MoltZap's exact +// turn-ready listen filter and notification. Every other request is delegated +// unchanged. +const SUBSCRIPTIONS_LISTEN_METHOD = "subscriptions/listen"; +const SUBSCRIPTIONS_ACKNOWLEDGED_NOTIFICATION = + "notifications/subscriptions/acknowledged"; +const JSON_RPC_VERSION = "2.0"; +const MODERN_PROTOCOL_VERSION = "2026-07-28"; +const INTERNAL_ERROR = -32603; +const SUBSCRIPTION_IN_USE = -32000; +const BAD_REQUEST_STATUS = 400; +const CONFLICT_STATUS = 409; +const INTERNAL_ERROR_STATUS = 500; +const OK_STATUS = 200; + +type JsonObject = Readonly>; + +interface HarnessMcpSubscriptionOptions { + readonly delegate: McpHttpHandler; + readonly implementation: Implementation; + readonly onerror?: (error: Error) => void; +} + +/** An official MCP handler augmented with one caller-typed event publisher. */ +export interface HarnessMcpSubscriptionHandler + extends McpHttpHandler { + /** Publishes one complete custom notification to the retained POST response. */ + readonly publish: (payload: Payload) => boolean; +} + +interface CustomListenRequest { + readonly id: RequestId; + readonly hasRequiredCapability: boolean; +} + +interface ActiveSubscription { + readonly id: RequestId; + controller?: ReadableStreamDefaultController; + abortCleanup?: () => void; + closed: boolean; +} + +interface JsonRpcErrorOptions { + readonly status: number; + readonly id: RequestId; + readonly code: number; + readonly message: string; + readonly data?: unknown; +} + +const isJsonObject = (value: unknown): value is JsonObject => + typeof value === "object" && value !== null && !Array.isArray(value); + +const jsonRpcError = ({ + status, + id, + code, + message, + data, +}: JsonRpcErrorOptions): Response => + Response.json( + { + jsonrpc: JSON_RPC_VERSION, + error: { + code, + message, + ...(data === undefined ? {} : { data }), + }, + id, + }, + { status }, + ); + +// #ignore-sloppy-code-next-line[async-keyword]: The official handler consumes a Promise-returning Fetch boundary. +const readRequestBody = async ( + request: Request, + options?: McpHandlerRequestOptions, + // #ignore-sloppy-code-next-line[promise-type]: The official handler consumes a Promise-returning Fetch boundary. +): Promise => { + if (options?.parsedBody !== undefined) { + return options.parsedBody; + } + return await request + .clone() + .json() + .catch(() => undefined); +}; + +const modernListenMessage = (request: Request, body: unknown) => { + const outcome = classifyInboundRequest({ + httpMethod: request.method.toUpperCase(), + protocolVersionHeader: + request.headers.get("mcp-protocol-version") ?? undefined, + mcpMethodHeader: request.headers.get("mcp-method") ?? undefined, + mcpNameHeader: request.headers.get("mcp-name") ?? undefined, + body, + }); + if (outcome.kind !== "modern" || outcome.messageKind !== "request") { + return undefined; + } + if ( + outcome.classification.revision !== MODERN_PROTOCOL_VERSION || + outcome.message.method !== SUBSCRIPTIONS_LISTEN_METHOD + ) { + return undefined; + } + return outcome.message; +}; + +const requiredCapabilityDeclared = (params: JsonObject): boolean => { + const meta = params._meta; + if (!isJsonObject(meta)) { + return false; + } + const capabilities = meta[CLIENT_CAPABILITIES_META_KEY]; + if (!isJsonObject(capabilities) || !isJsonObject(capabilities.extensions)) { + return false; + } + return Object.hasOwn(capabilities.extensions, HARNESS_EVENTS_EXTENSION); +}; + +const customListenRequest = ( + request: Request, + body: unknown, +): CustomListenRequest | undefined => { + const message = modernListenMessage(request, body); + if (message === undefined || !isJsonObject(message.params)) { + return undefined; + } + const notifications = message.params.notifications; + const notificationKeys = isJsonObject(notifications) + ? Object.keys(notifications) + : []; + if ( + !isJsonObject(notifications) || + notificationKeys.length !== 1 || + notificationKeys[0] !== HARNESS_TURN_READY_FILTER || + notifications[HARNESS_TURN_READY_FILTER] !== true + ) { + return undefined; + } + return { + id: message.id, + hasRequiredCapability: requiredCapabilityDeclared(message.params), + }; +}; + +const shouldInspect = (request: Request): boolean => + request.method.toUpperCase() === "POST" && + isJsonContentType(request.headers.get("content-type")) && + request.headers.get("mcp-method") === SUBSCRIPTIONS_LISTEN_METHOD; + +class HarnessMcpSubscriptionState { + private readonly delegate: McpHttpHandler; + private readonly implementation: Implementation; + private readonly onerror?: (error: Error) => void; + private readonly encoder = new TextEncoder(); + private active?: ActiveSubscription; + private closed = false; + private closePromise?: Promise; // #ignore-sloppy-code[promise-type]: McpHttpHandler.close is Promise-native. + + constructor(options: HarnessMcpSubscriptionOptions) { + this.delegate = options.delegate; + this.implementation = options.implementation; + this.onerror = options.onerror; + } + + // #ignore-sloppy-code-next-line[async-keyword]: McpHttpHandler.fetch is Promise-native. + readonly fetch: McpHttpHandler["fetch"] = async (request, options) => { + if (this.closed || !shouldInspect(request)) { + return options === undefined + ? await this.delegate.fetch(request) + : await this.delegate.fetch(request, options); + } + const body = + options === undefined + ? await readRequestBody(request) + : await readRequestBody(request, options); + const listenRequest = customListenRequest(request, body); + if (this.closed || listenRequest === undefined) { + return options === undefined + ? await this.delegate.fetch(request) + : await this.delegate.fetch(request, options); + } + return this.serve(request, listenRequest); + }; + + readonly publish = (payload: Payload): boolean => { + const subscription = this.active; + if (subscription === undefined || subscription.closed) { + return false; + } + const published = this.enqueueMessage(subscription, { + jsonrpc: JSON_RPC_VERSION, + method: HARNESS_TURN_READY_NOTIFICATION, + params: { + ...payload, + _meta: { [SUBSCRIPTION_ID_META_KEY]: subscription.id }, + }, + }); + if (!published) { + this.teardown(subscription, false); + } + return published; + }; + + // #ignore-sloppy-code-next-line[promise-type]: McpHttpHandler.close is Promise-native. + readonly close = (): Promise => { + if (this.closePromise !== undefined) { + return this.closePromise; + } + this.closed = true; + if (this.active !== undefined) { + this.teardown(this.active, true); + } + this.closePromise = this.delegate.close(); + return this.closePromise; + }; + + asHandler(): HarnessMcpSubscriptionHandler { + return { + fetch: this.fetch, + close: this.close, + publish: this.publish, + notify: this.delegate.notify, + bus: this.delegate.bus, + }; + } + + private reportError(error: unknown): void { + if (this.onerror === undefined) { + return; + } + try { + this.onerror(error instanceof Error ? error : new Error(String(error))); + } catch (reportingError) { + console.error( + "Harness MCP subscription error reporter failed", + reportingError, + ); + } + } + + private enqueueMessage( + subscription: ActiveSubscription, + message: JsonObject, + ): boolean { + if (subscription.closed || subscription.controller === undefined) { + return false; + } + try { + const frame = `data: ${JSON.stringify(message)}\n\n`; + subscription.controller.enqueue(this.encoder.encode(frame)); + return true; + } catch (error) { + this.reportError(error); + return false; + } + } + + private teardown(subscription: ActiveSubscription, graceful: boolean): void { + if (subscription.closed) { + return; + } + if (graceful) { + this.enqueueMessage(subscription, this.completeMessage(subscription.id)); + } + subscription.closed = true; + subscription.abortCleanup?.(); + if (this.active === subscription) { + this.active = undefined; + } + try { + subscription.controller?.close(); + } catch (error) { + this.reportError(error); + } + } + + private completeMessage(id: RequestId): JsonObject { + return { + jsonrpc: JSON_RPC_VERSION, + id, + result: { + resultType: "complete", + _meta: { + [SUBSCRIPTION_ID_META_KEY]: id, + [SERVER_INFO_META_KEY]: this.implementation, + }, + }, + }; + } + + private serve( + request: Request, + listenRequest: CustomListenRequest, + ): Response { + const refusal = this.admissionRefusal(listenRequest); + if (refusal !== undefined) { + return refusal; + } + const subscription: ActiveSubscription = { + id: listenRequest.id, + closed: false, + }; + this.active = subscription; + const readable = this.makeReadable(request, subscription); + if (readable === undefined) { + return jsonRpcError({ + status: INTERNAL_ERROR_STATUS, + id: listenRequest.id, + code: INTERNAL_ERROR, + message: "Internal server error", + }); + } + return new Response(readable, { + status: OK_STATUS, + headers: { + "cache-control": "no-cache", + connection: "keep-alive", + "content-type": "text/event-stream", + "x-accel-buffering": "no", + }, + }); + } + + private admissionRefusal( + listenRequest: CustomListenRequest, + ): Response | undefined { + if (!listenRequest.hasRequiredCapability) { + const error = new MissingRequiredClientCapabilityError({ + requiredCapabilities: { + extensions: { [HARNESS_EVENTS_EXTENSION]: {} }, + }, + }); + return jsonRpcError({ + status: BAD_REQUEST_STATUS, + id: listenRequest.id, + code: error.code, + message: error.message, + data: error.data, + }); + } + return this.active === undefined + ? undefined + : jsonRpcError({ + status: CONFLICT_STATUS, + id: listenRequest.id, + code: SUBSCRIPTION_IN_USE, + message: "Turn-ready subscription already in use", + data: { kind: "subscription_in_use" }, + }); + } + + private makeReadable( + request: Request, + subscription: ActiveSubscription, + ): ReadableStream | undefined { + try { + return new ReadableStream({ + start: (controller) => { + this.start(request, subscription, controller); + }, + cancel: () => { + this.teardown(subscription, false); + }, + }); + } catch (error) { + this.teardown(subscription, false); + this.reportError(error); + return undefined; + } + } + + private start( + request: Request, + subscription: ActiveSubscription, + controller: ReadableStreamDefaultController, + ): void { + subscription.controller = controller; + if (request.signal.aborted) { + this.teardown(subscription, false); + return; + } + const onAbort = (): void => { + this.teardown(subscription, false); + }; + request.signal.addEventListener("abort", onAbort, { once: true }); + subscription.abortCleanup = () => { + request.signal.removeEventListener("abort", onAbort); + }; + if (!this.enqueueMessage(subscription, this.ackMessage(subscription.id))) { + this.teardown(subscription, false); + } + } + + private ackMessage(id: RequestId): JsonObject { + return { + jsonrpc: JSON_RPC_VERSION, + method: SUBSCRIPTIONS_ACKNOWLEDGED_NOTIFICATION, + params: { + notifications: { [HARNESS_TURN_READY_FILTER]: true }, + _meta: { [SUBSCRIPTION_ID_META_KEY]: id }, + }, + }; + } +} + +/** + * Adds the MoltZap turn-ready subscription extension to an official MCP HTTP + * handler. Every non-extension request remains owned by the SDK delegate. + * + * @param options Official handler, server identity, and optional error sink. + * @returns The official handler surface plus a typed custom publisher. + */ +export const makeHarnessMcpSubscriptionHandler = ( + options: HarnessMcpSubscriptionOptions, +): HarnessMcpSubscriptionHandler => + new HarnessMcpSubscriptionState(options).asHandler(); + +/* eslint-enable agent-code-guard/async-keyword, agent-code-guard/promise-type, @typescript-eslint/no-invalid-void-type -- Restore repository defaults after the Promise-native MCP boundary. */ diff --git a/packages/client/src/harness-mcp-wire.ts b/packages/client/src/harness-mcp-wire.ts index 15b123e2a..3cfbb9b57 100644 --- a/packages/client/src/harness-mcp-wire.ts +++ b/packages/client/src/harness-mcp-wire.ts @@ -9,6 +9,21 @@ 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 { + decodeHarnessReplyRoute, + HARNESS_EVENTS_EXTENSION, + HARNESS_REPLY_TOOL, + harnessReplyInputJsonSchema, + harnessReplyResultJsonSchema, + type HarnessReplyInput, + type HarnessReplyResult, + type HarnessTurnEvent, +} from "./harness/index.js"; +import { + makeHarnessMcpSubscriptionHandler, + type HarnessMcpSubscriptionHandler, +} from "./harness-mcp-subscription.js"; import { statusCommandRpc, type localDaemonCommands, @@ -20,9 +35,14 @@ const STATUS_TOOL_NAME = "status"; type StatusPayload = Schema.Schema.Type; type StatusResult = Schema.Schema.Type; type StatusHandler = LocalDaemonHandlers[typeof localDaemonCommands.status]; +type ReplyHandler = ( + conversationId: ConversationId, + payload: string, +) => Effect.Effect; interface HarnessMcpHandlerOptions { readonly implementation: Implementation; + readonly reply: ReplyHandler; readonly status: StatusHandler; } @@ -39,14 +59,17 @@ const statusOutputSchema = fromJsonSchema( ) as JsonSchemaType, ); +const replyInputSchema = fromJsonSchema( + /* Safe because Effect and MCP expose the same JSON Schema wire shape with different array mutability declarations. */ harnessReplyInputJsonSchema as JsonSchemaType, +); +const replyOutputSchema = fromJsonSchema( + /* Safe because Effect and MCP expose the same JSON Schema wire shape with different array mutability declarations. */ harnessReplyResultJsonSchema as JsonSchemaType, +); + const makeRegistrationServer = (implementation: Implementation): McpServer => new McpServer(implementation); -const makeActiveServer = ( - implementation: Implementation, - status: StatusHandler, -): McpServer => { - const server = new McpServer(implementation); +const registerStatusTool = (server: McpServer, status: StatusHandler): void => { server.registerTool( STATUS_TOOL_NAME, { @@ -72,6 +95,43 @@ const makeActiveServer = ( ); }, ); +}; + +const registerReplyTool = (server: McpServer, reply: ReplyHandler): void => { + server.registerTool( + HARNESS_REPLY_TOOL, + { + inputSchema: replyInputSchema, + outputSchema: replyOutputSchema, + }, + (input, context) => + Effect.runPromise( + Effect.gen(function* () { + const route = yield* decodeHarnessReplyRoute(context.mcpReq._meta); + yield* reply(route.conversationId, input.payload); + const result: HarnessReplyResult = {}; + return { + content: [{ type: "text" as const, text: JSON.stringify(result) }], + structuredContent: result, + }; + }), + { signal: context.mcpReq.signal }, + ), + ); +}; + +const makeActiveServer = ( + implementation: Implementation, + status: StatusHandler, + reply: ReplyHandler, +): McpServer => { + const server = new McpServer(implementation, { + capabilities: { + extensions: { [HARNESS_EVENTS_EXTENSION]: {} }, + }, + }); + registerStatusTool(server, status); + registerReplyTool(server, reply); return server; }; @@ -80,20 +140,30 @@ const makeActiveServer = ( * * @param options Existing daemon capabilities exposed through MCP. * @param options.implementation Existing MCP server identity. + * @param options.reply Conversation-bound raw reply handler. * @param options.status Existing local daemon status handler. * @returns The registration and active-agent HTTP handlers. */ export const makeHarnessMcpHttpHandlers = ({ implementation, + reply, status, }: HarnessMcpHandlerOptions): { readonly registration: McpHttpHandler; - readonly active: McpHttpHandler; -} => ({ - registration: createMcpHandler(() => makeRegistrationServer(implementation), { - legacy: "reject", - }), - active: createMcpHandler(() => makeActiveServer(implementation, status), { - legacy: "reject", - }), -}); + readonly active: HarnessMcpSubscriptionHandler; +} => { + const activeDelegate = createMcpHandler( + () => makeActiveServer(implementation, status, reply), + { legacy: "reject" }, + ); + return { + registration: createMcpHandler( + () => makeRegistrationServer(implementation), + { legacy: "reject" }, + ), + active: makeHarnessMcpSubscriptionHandler({ + delegate: activeDelegate, + implementation, + }), + }; +}; diff --git a/packages/client/src/harness/client-runtime.ts b/packages/client/src/harness/client-runtime.ts new file mode 100644 index 000000000..4e70d407e --- /dev/null +++ b/packages/client/src/harness/client-runtime.ts @@ -0,0 +1,215 @@ +/* eslint-disable agent-code-guard/promise-type, @typescript-eslint/no-invalid-void-type -- The official MCP client lifecycle is Promise-native and is converted to Effect at this private adapter edge. */ +import { + Client, + fromJsonSchema, + StreamableHTTPClientTransport, + type JsonSchemaType, + type McpSubscription, + type SubscriptionFilter, +} from "@modelcontextprotocol/client"; +import { Effect, Queue, Stream, Take, type Scope } from "effect"; +import packageJson from "../../package.json" with { type: "json" }; +import type { ConversationId } from "@moltzap/protocol/conversation"; +import { + decodeHarnessTurnEvent, + HARNESS_EVENTS_EXTENSION, + HARNESS_REPLY_TOOL, + HARNESS_TURN_READY_FILTER, + HARNESS_TURN_READY_NOTIFICATION, + harnessReplyRequestMeta, + harnessTurnConversationId, + type HarnessTurnEvent, +} from "./runtime.js"; + +interface HarnessClientInternalOptions { + readonly url: string; +} + +interface HarnessTurnInternal { + readonly conversationId: ConversationId; + readonly messages: HarnessTurnEvent["messages"]; + readonly reply: (payload: string) => Effect.Effect; +} + +interface HarnessClientInternalService { + readonly turns: Stream.Stream; +} + +const MODERN_PROTOCOL_VERSION = "2026-07-28"; +const CLIENT_IMPLEMENTATION = { + name: "moltzap-harness-client", + version: packageJson.version, +} as const; + +const TURN_READY_FILTER: SubscriptionFilter & { + readonly [HARNESS_TURN_READY_FILTER]: true; +} = { + [HARNESS_TURN_READY_FILTER]: true, +}; + +const unknownNotificationParams = fromJsonSchema({} satisfies JsonSchemaType); + +const asError = (cause: unknown): Error => + cause instanceof Error ? cause : new Error(String(cause)); + +const closeQuietly = (close: () => Promise): Effect.Effect => + Effect.tryPromise({ try: close, catch: asError }).pipe(Effect.ignore); + +const turnPayload = (params: unknown): unknown => { + if (typeof params !== "object" || params === null || Array.isArray(params)) { + return params; + } + const payload: Record = { ...params }; + Reflect.deleteProperty(payload, "_meta"); + return payload; +}; + +const callReply = ( + client: Client, + originatingConversationId: ConversationId, + payload: string, +): Effect.Effect => + Effect.tryPromise({ + try: (signal) => + client.callTool( + { + name: HARNESS_REPLY_TOOL, + arguments: { payload }, + _meta: harnessReplyRequestMeta(originatingConversationId), + }, + { signal }, + ), + catch: asError, + }).pipe( + Effect.flatMap((result) => { + if (result.isError === true) { + // eslint-disable-next-line agent-code-guard/effect-error-erasure -- The raw MCP failure has no portable domain-error contract, so this boundary exposes an ordinary Error. + return Effect.fail(new Error("Harness reply tool failed")); + } + return Effect.void; + }), + Effect.asVoid, + ); + +const makeTurn = ( + client: Client, + event: HarnessTurnEvent, +): HarnessTurnInternal => { + const originatingConversationId = harnessTurnConversationId(event); + return { + conversationId: originatingConversationId, + messages: event.messages, + reply: (payload) => callReply(client, originatingConversationId, payload), + }; +}; + +const connect = (client: Client, url: string): Effect.Effect => + Effect.tryPromise({ + try: (signal) => + client.connect(new StreamableHTTPClientTransport(new URL(url)), { + signal, + }), + catch: asError, + }).pipe( + Effect.onError(() => closeQuietly(() => client.close())), + Effect.as(client), + ); + +const listen = (client: Client): Effect.Effect => + Effect.tryPromise({ + try: (signal) => client.listen(TURN_READY_FILTER, { signal }), + catch: asError, + }); + +const verifyServerExtension = (client: Client): Effect.Effect => { + const extensions = client.getServerCapabilities()?.extensions; + if ( + extensions !== undefined && + Object.hasOwn(extensions, HARNESS_EVENTS_EXTENSION) + ) { + return Effect.void; + } + // eslint-disable-next-line agent-code-guard/effect-error-erasure -- An incompatible MCP peer is rejected at the public client boundary, whose existing error contract is Error. + return Effect.fail( + new Error( + `Harness MCP server does not advertise ${HARNESS_EVENTS_EXTENSION}`, + ), + ); +}; + +const observeSubscription = ( + subscription: McpSubscription, + queue: Queue.Queue>, +): Effect.Effect => + Effect.tryPromise({ + try: () => subscription.closed, + catch: asError, + }).pipe( + // The official MCP subscription contract states that `closed` never rejects. + Effect.orDie, + Effect.flatMap(() => Queue.offer(queue, Take.end)), + Effect.asVoid, + ); + +/** + * Acquires the private official-SDK adapter behind the public HarnessClient. + * + * @internal + * @param options Package-owned loopback endpoint options. + * @returns The scoped structural service consumed by the public facade. + */ +export const acquireHarnessClientInternal = ( + options: HarnessClientInternalOptions, +): Effect.Effect => + Effect.gen(function* () { + const queue = + yield* Queue.unbounded>(); + yield* Effect.addFinalizer(() => Queue.shutdown(queue)); + + const client = new Client(CLIENT_IMPLEMENTATION, { + capabilities: { + extensions: { [HARNESS_EVENTS_EXTENSION]: {} }, + }, + versionNegotiation: { + mode: { pin: MODERN_PROTOCOL_VERSION }, + }, + }); + + client.setNotificationHandler( + HARNESS_TURN_READY_NOTIFICATION, + { params: unknownNotificationParams }, + (params) => + Effect.runPromise( + decodeHarnessTurnEvent(turnPayload(params)).pipe( + Effect.matchEffect({ + onFailure: (cause) => + Queue.offer(queue, Take.fail(asError(cause))), + onSuccess: (event) => + Queue.offer(queue, Take.of(makeTurn(client, event))), + }), + Effect.asVoid, + ), + ), + ); + + yield* ( + // eslint-disable-next-line agent-code-guard/acquire-release-requires-scope -- The internal acquisition retains Scope in its return type and the public facade supplies that enclosing scope. + Effect.acquireRelease(connect(client, options.url), () => + closeQuietly(() => client.close()), + ) + ); + yield* verifyServerExtension(client); + const subscription = yield* ( + // eslint-disable-next-line agent-code-guard/acquire-release-requires-scope -- The internal acquisition retains Scope in its return type and the public facade supplies that enclosing scope. + Effect.acquireRelease(listen(client), (subscription) => + closeQuietly(() => subscription.close()), + ) + ); + yield* Effect.forkScoped(observeSubscription(subscription, queue)); + + return { + turns: Stream.fromQueue(queue).pipe(Stream.flattenTake), + }; + }).pipe(Effect.withSpan("acquireHarnessClient")); + +/* eslint-enable agent-code-guard/promise-type, @typescript-eslint/no-invalid-void-type -- Restore strict defaults after the Promise-native MCP adapter edge. */ diff --git a/packages/client/src/harness/index.ts b/packages/client/src/harness/index.ts new file mode 100644 index 000000000..379dccee2 --- /dev/null +++ b/packages/client/src/harness/index.ts @@ -0,0 +1,16 @@ +/** @internal */ +export { acquireHarnessClientInternal } from "./client-runtime.js"; +/** @internal */ +export { + decodeHarnessReplyRoute, + HARNESS_EVENTS_EXTENSION, + HARNESS_REPLY_TOOL, + HARNESS_TURN_READY_FILTER, + HARNESS_TURN_READY_NOTIFICATION, + harnessReplyInputJsonSchema, + harnessReplyResultJsonSchema, + type HarnessReplyInput, + type HarnessReplyResult, + type HarnessReplyRoute, + type HarnessTurnEvent, +} from "./runtime.js"; diff --git a/packages/client/src/harness/runtime.test.ts b/packages/client/src/harness/runtime.test.ts new file mode 100644 index 000000000..8b243faa7 --- /dev/null +++ b/packages/client/src/harness/runtime.test.ts @@ -0,0 +1,213 @@ +/* eslint-disable agent-code-guard/async-keyword -- The official MCP SDK exposes Promise-native client and handler lifecycle APIs at this interoperability boundary. */ +import { + Client, + StreamableHTTPClientTransport, +} from "@modelcontextprotocol/client"; +import { + createMcpHandler, + fromJsonSchema, + McpServer, + type JsonSchemaType, +} from "@modelcontextprotocol/server"; +import { Effect, Exit, Scope } from "effect"; +import { describe, expect, it } from "vitest"; +import type { Message } from "@moltzap/protocol/message"; +import { agentId, conversationId, messageId } from "@moltzap/protocol/testing"; +import { acquireHarnessMcpHttpServer } from "../harness-mcp-server.js"; +import { + decodeHarnessReplyRoute, + decodeHarnessTurnEvent, + HARNESS_EVENTS_EXTENSION, + HARNESS_REPLY_TOOL, + harnessReplyInputJsonSchema, + harnessReplyRequestMeta, + harnessReplyResultJsonSchema, + harnessTurnConversationId, + type HarnessReplyInput, + type HarnessReplyResult, + type HarnessReplyRoute, +} from "./runtime.js"; + +const MODERN_PROTOCOL_VERSION = "2026-07-28"; +const CONVERSATION_ID = conversationId("00000000-0000-4000-8000-000000000001"); +const SENDER_ID = agentId("00000000-0000-4000-8000-000000000003"); + +const firstMessage = { + id: messageId("00000000-0000-4000-8000-000000000004"), + conversationId: CONVERSATION_ID, + senderId: SENDER_ID, + parts: [{ type: "text", text: "first" }], + createdAt: "2026-08-03T12:00:00.000Z", +} satisfies Message; + +const secondMessage = { + id: messageId("00000000-0000-4000-8000-000000000005"), + conversationId: CONVERSATION_ID, + senderId: SENDER_ID, + parts: [{ type: "text", text: "second" }], + createdAt: "2026-08-03T12:00:01.000Z", +} satisfies Message; + +const otherConversationMessage = { + ...secondMessage, + conversationId: conversationId("00000000-0000-4000-8000-000000000002"), +} satisfies Message; + +const replyInputJsonSchema = fromJsonSchema( + /* Safe because Effect and MCP expose the same JSON Schema wire shape with different array mutability declarations. */ harnessReplyInputJsonSchema as JsonSchemaType, +); + +const replyResultJsonSchema = fromJsonSchema( + /* Safe because Effect and MCP expose the same JSON Schema wire shape with different array mutability declarations. */ harnessReplyResultJsonSchema as JsonSchemaType, +); + +const decodesProtocolMessageBatch = async () => { + const turn = await Effect.runPromise( + decodeHarnessTurnEvent({ messages: [firstMessage, secondMessage] }), + ); + + expect(turn.messages).toEqual([firstMessage, secondMessage]); + expect(harnessTurnConversationId(turn)).toBe(CONVERSATION_ID); + await expect( + Effect.runPromise(decodeHarnessTurnEvent({ messages: [] })), + ).rejects.toBeDefined(); + await expect( + Effect.runPromise( + decodeHarnessTurnEvent({ + messages: [firstMessage, otherConversationMessage], + }), + ), + ).rejects.toBeDefined(); +}; + +const keepsPrivateRoutingMetadataClosed = async () => { + const requestMeta = { + ...harnessReplyRequestMeta(CONVERSATION_ID), + "io.modelcontextprotocol/unrelated": true, + }; + await expect( + Effect.runPromise(decodeHarnessReplyRoute(requestMeta)), + ).resolves.toEqual({ conversationId: CONVERSATION_ID }); + await expect( + Effect.runPromise( + decodeHarnessReplyRoute({ + [HARNESS_EVENTS_EXTENSION]: { + conversationId: CONVERSATION_ID, + invented: true, + }, + }), + ), + ).rejects.toBeDefined(); +}; + +interface ObservedReply { + arguments?: unknown; + route?: HarnessReplyRoute; +} + +const makeRuntimeHandlers = (observed: ObservedReply) => { + const registrationHandler = createMcpHandler( + () => new McpServer({ name: "registration-test", version: "1.0.0" }), + { legacy: "reject" }, + ); + const harnessHandler = createMcpHandler(() => { + const server = new McpServer({ + name: "harness-runtime-test", + version: "1.0.0", + }); + server.registerTool( + HARNESS_REPLY_TOOL, + { + inputSchema: replyInputJsonSchema, + outputSchema: replyResultJsonSchema, + }, + async (input, context) => { + observed.arguments = input; + observed.route = await Effect.runPromise( + decodeHarnessReplyRoute(context.mcpReq._meta), + ); + return { + content: [{ type: "text", text: "{}" }], + structuredContent: {}, + }; + }, + ); + return server; + }); + return { harnessHandler, registrationHandler }; +}; + +const assertPayloadOnlyDiscovery = async (client: Client) => { + const replyTool = (await client.listTools()).tools.find( + (tool) => tool.name === HARNESS_REPLY_TOOL, + ); + expect(replyTool?.inputSchema).toMatchObject({ + type: "object", + properties: { payload: { type: "string" } }, + required: ["payload"], + additionalProperties: false, + }); + expect(Object.keys(replyTool?.inputSchema.properties ?? {})).toEqual([ + "payload", + ]); +}; + +const preservesPrivateRoute = async () => { + const observed: ObservedReply = {}; + const { harnessHandler, registrationHandler } = makeRuntimeHandlers(observed); + const scope = Effect.runSync(Scope.make()); + const listener = await Effect.runPromise( + acquireHarnessMcpHttpServer({ + port: 0, + registrationHandler, + harnessHandler, + }).pipe(Scope.extend(scope)), + ); + const address = listener.address(); + if (address === null || typeof address === "string") { + await Effect.runPromise(Scope.close(scope, Exit.void)); + throw new Error("expected a TCP test server address"); + } + const client = new Client( + { name: "harness-runtime-client-test", version: "1.0.0" }, + { + versionNegotiation: { + mode: { pin: MODERN_PROTOCOL_VERSION }, + }, + }, + ); + + try { + await client.connect( + new StreamableHTTPClientTransport( + new URL(`http://127.0.0.1:${address.port}/mcp`), + ), + ); + await assertPayloadOnlyDiscovery(client); + + const result = await client.callTool({ + name: HARNESS_REPLY_TOOL, + arguments: { payload: "reply text" }, + _meta: harnessReplyRequestMeta(CONVERSATION_ID), + }); + + expect(observed.arguments).toEqual({ payload: "reply text" }); + expect(observed.route).toEqual({ conversationId: CONVERSATION_ID }); + expect(result.structuredContent).toEqual({}); + } finally { + await client.close(); + await Effect.runPromise(Scope.close(scope, Exit.void)); + } +}; + +// @agent-code-guard/regression-only: these examples pin the closed runtime boundary and official SDK metadata preservation. +describe("Harness MCP runtime contract", () => { + it("decodes a nonempty batch of existing protocol messages", () => + decodesProtocolMessageBatch()); + it("keeps private routing metadata closed", () => + keepsPrivateRoutingMetadataClosed()); + it("preserves the private route through an official MCP client call", () => + preservesPrivateRoute()); +}); + +/* eslint-enable agent-code-guard/async-keyword -- Restore strict defaults after the Promise-native interoperability fixture. */ diff --git a/packages/client/src/harness/runtime.ts b/packages/client/src/harness/runtime.ts new file mode 100644 index 000000000..92e1e9c17 --- /dev/null +++ b/packages/client/src/harness/runtime.ts @@ -0,0 +1,131 @@ +import { JSONSchema, Schema } from "effect"; +import { + conversationId, + type ConversationId, +} from "@moltzap/protocol/conversation"; +import { messageReceivedNotificationDefinition } from "@moltzap/protocol/message"; + +/** Harness MCP extension carrying the runtime event contract. */ +export const HARNESS_EVENTS_EXTENSION = "xyz.moltzap/events-v1"; + +/** Subscription filter requesting reply-capable harness turns. */ +export const HARNESS_TURN_READY_FILTER = "xyz.moltzap/turnReady"; + +/** Notification method carrying one coalesced inbound turn. */ +export const HARNESS_TURN_READY_NOTIFICATION = + "notifications/xyz.moltzap/turn_ready"; + +/** Tool used for model output in the current conversation. */ +export const HARNESS_REPLY_TOOL = "reply"; + +const messageSchema = + messageReceivedNotificationDefinition.paramsSchema.fields.message; + +/** One nonempty batch of protocol messages delivered as a model turn. */ +const harnessTurnEventSchema = Schema.Struct({ + messages: Schema.NonEmptyArray(messageSchema), +}).pipe( + Schema.filter( + (turn) => + turn.messages.every( + (message) => message.conversationId === turn.messages[0].conversationId, + ) || "every turn message must belong to the same conversation", + ), +); + +/** Advertised reply arguments. Routing authority remains outside tool input. */ +const harnessReplyInputSchema = Schema.Struct({ + payload: Schema.String, +}); + +/** The reply operation has no additional result data. */ +const harnessReplyResultSchema = Schema.Struct({}); + +/** Private route nested under the harness extension key in MCP request metadata. */ +const harnessReplyRouteSchema = Schema.Struct({ + conversationId, +}); + +/** Decoded harness turn event. */ +export type HarnessTurnEvent = Schema.Schema.Type< + typeof harnessTurnEventSchema +>; + +/** Decoded reply input. */ +export type HarnessReplyInput = Schema.Schema.Type< + typeof harnessReplyInputSchema +>; + +/** Decoded reply result. */ +export type HarnessReplyResult = Schema.Schema.Type< + typeof harnessReplyResultSchema +>; + +/** Decoded private reply route. */ +export type HarnessReplyRoute = Schema.Schema.Type< + typeof harnessReplyRouteSchema +>; + +const decodeTurnEvent = Schema.decodeUnknown(harnessTurnEventSchema); +const decodeReplyRoute = Schema.decodeUnknown(harnessReplyRouteSchema); + +const strictDecodeOptions = { onExcessProperty: "error" } as const; + +/** JSON Schema advertised for the payload-only reply tool arguments. */ +export const harnessReplyInputJsonSchema = JSONSchema.make( + harnessReplyInputSchema, + { target: "jsonSchema2020-12" }, +); + +/** JSON Schema advertised for the empty reply result. */ +export const harnessReplyResultJsonSchema = JSONSchema.make( + harnessReplyResultSchema, + { target: "jsonSchema2020-12" }, +); + +/** + * Strictly decode a turn event received from the MCP boundary. + * @param value Untrusted notification parameters. + * @returns The decoded nonempty protocol-message batch. + */ +export const decodeHarnessTurnEvent = (value: unknown) => + decodeTurnEvent(value, strictDecodeOptions); + +/** + * Build the private request metadata consumed by the production harness client. + * @param originatingConversationId Conversation associated with the live turn. + * @returns Namespaced MCP request metadata containing the private route. + */ +export const harnessReplyRequestMeta = ( + originatingConversationId: ConversationId, +): Readonly> => ({ + [HARNESS_EVENTS_EXTENSION]: { + conversationId: originatingConversationId, + }, +}); + +const isUnknownRecord = ( + value: unknown, +): value is Readonly> => + typeof value === "object" && value !== null; + +/** + * Decode the private reply route while allowing unrelated MCP metadata keys. + * @param requestMeta Untrusted MCP request metadata. + * @returns The decoded conversation route. + */ +export const decodeHarnessReplyRoute = (requestMeta: unknown) => { + const extensionValue: unknown = isUnknownRecord(requestMeta) + ? requestMeta[HARNESS_EVENTS_EXTENSION] + : undefined; + return decodeReplyRoute(extensionValue, strictDecodeOptions); +}; + +/** + * Return the conversation carried by the first message in a nonempty turn. + * @param turn Decoded nonempty message batch. + * @returns Conversation carried by the first protocol message. + */ +export const harnessTurnConversationId = ( + turn: HarnessTurnEvent, +): ConversationId => turn.messages[0].conversationId; diff --git a/packages/client/src/index.ts b/packages/client/src/index.ts index c5798b1ef..384360ac8 100644 --- a/packages/client/src/index.ts +++ b/packages/client/src/index.ts @@ -13,3 +13,12 @@ export { type AgentClientOptions, type RpcCallOptions, } from "./agent-client.js"; +/** Re-exports the adapter-facing daemon client capability. */ +export { + acquireHarnessClient, + HarnessClient, + makeHarnessClientLayer, + type HarnessClientOptions, + type HarnessClientService, + type HarnessTurn, +} from "./harness-client.js"; diff --git a/packages/client/src/moltzapd.ts b/packages/client/src/moltzapd.ts index 040d1bff1..f8ccddbbb 100644 --- a/packages/client/src/moltzapd.ts +++ b/packages/client/src/moltzapd.ts @@ -2,6 +2,7 @@ import type { Implementation } from "@modelcontextprotocol/server"; import { Effect, ExecutionStrategy, Exit, Scope } from "effect"; import packageJson from "../package.json" with { type: "json" }; import { MoltZapChannelCore } from "./channel-core.js"; +import type { HarnessTurnEvent } from "./harness/index.js"; import { acquireHarnessMcpHttpServer } from "./harness-mcp-server.js"; import { makeHarnessMcpHttpHandlers } from "./harness-mcp-wire.js"; import type { @@ -37,14 +38,29 @@ const makeStatusHandler = conversations: service.getConversations().length, }); -const acquireConnectedCore = ( +const acquireCore = ( service: MoltZapService, ): Effect.Effect => // eslint-disable-next-line agent-code-guard/acquire-release-requires-scope -- the process scope owns the sole core and its network connection Effect.acquireRelease( Effect.sync(() => new MoltZapChannelCore({ service })), (core) => core.disconnect(), - ).pipe(Effect.tap((core) => core.connect())); + ); + +const installTurnPublisher = ( + core: MoltZapChannelCore, + publish: (turn: HarnessTurnEvent) => boolean, +): void => { + core.onRawInbound((messages) => + Effect.sync(() => { + const first = messages[0]; + if (first === undefined) { + return; + } + publish({ messages: [first, ...messages.slice(1)] }); + }), + ); +}; /** * Owns one registered agent's service, channel core, network connection, and @@ -59,8 +75,9 @@ const acquireConnectedCore = ( * * process->>service: make(profileName) * process->>core: construct(service) - * process->>core: connect() + * process->>core: install raw turn publisher * process->>mcp: listen(port) + * process->>core: connect() * Note over core,mcp: Scope release closes MCP before disconnecting the core * ``` * @@ -86,16 +103,20 @@ export const acquireMoltzapd = ( ); const acquire = Effect.gen(function* () { const service = yield* MoltZapService.make(options.profileName); - const core = yield* acquireConnectedCore(service); + const core = yield* acquireCore(service); const handlers = makeHarnessMcpHttpHandlers({ implementation: MCP_IMPLEMENTATION, + reply: core.sendReply.bind(core), status: makeStatusHandler(service, core), }); - return yield* acquireHarnessMcpHttpServer({ + installTurnPublisher(core, handlers.active.publish); + const server = yield* acquireHarnessMcpHttpServer({ port: options.port, registrationHandler: handlers.registration, harnessHandler: handlers.active, }); + yield* core.connect(); + return server; }).pipe(Scope.extend(daemonScope)); return yield* acquire.pipe( Effect.onExit((exit) => diff --git a/packages/openclaw-channel/AGENTS.md b/packages/openclaw-channel/AGENTS.md index c493f16d4..b1834df38 100644 --- a/packages/openclaw-channel/AGENTS.md +++ b/packages/openclaw-channel/AGENTS.md @@ -34,14 +34,9 @@ surface. `routeReply()` (`OriginatingChannel === Surface` always holds for MoltZap→MoltZap), so the deliver callback MUST send the reply via `core.sendReply(conversationId, text)`. -- `createReplyGuardedDeliver` builds one `ReplyGuard` - (`@moltzap/client/channel-base`) per inbound turn, stamped only - after the first successful `core.sendReply` so a transient failure - leaves a retry able to send. A suppressed second final reply calls - host opt-in `MoltzapChannelPluginDeps.onDuplicateReply` with the - conversation id; deliver still returns `false` per - `OpenClawDeliver: PromiseLike`, as does any send failure — - all are transient, so the host may retry. +- Each final `deliver` call sends through + `core.sendReply(conversationId, text)`. A send failure returns `false` + per `OpenClawDeliver: PromiseLike` so the host may retry. - Target resolution: `messaging.targetResolver` validates both target formats with no server round-trip; `directory` (`listPeers`, `listGroups` — named groups only) is live RPC returning `[]` on diff --git a/packages/openclaw-channel/src/MODULE.md b/packages/openclaw-channel/src/MODULE.md index 42b0d2d79..d9917de24 100644 --- a/packages/openclaw-channel/src/MODULE.md +++ b/packages/openclaw-channel/src/MODULE.md @@ -10,7 +10,7 @@ runtime entries from `index.*` at the extension root only, so the built ## Public surface -### [`createMoltzapChannelPlugin`](./openclaw-entry.ts#L1295) +### [`createMoltzapChannelPlugin`](./openclaw-entry.ts#L1226) _Function_ @@ -42,21 +42,15 @@ sequenceDiagram Core->>Plugin: enriched message arrives Plugin->>OC: dispatchReplyWithBufferedBlockDispatcher note over OC: agent pipeline → LLM - OC->>Plugin: deliver(payload, opts) — createReplyGuardedDeliver + OC->>Plugin: deliver(payload, opts) — createReplyDeliver Plugin->>Server: core.sendReply(conversationId, text) - alt second final reply for the same turn - Plugin->>Plugin: ReplyGuard already stamped
onDuplicateReply callback, return false - end OC->>Plugin: stopAccount(ctx) Plugin->>Core: core.disconnect() Plugin->>Plugin: activeClients.delete(account) ``` `deliver` returns `PromiseLike<boolean>` per openclaw contract; -false signals "not delivered" without throwing. The reply guard is -single-shot per inbound turn: a second final reply is suppressed locally -and reported through `MoltzapChannelPluginDeps.onDuplicateReply` rather -than a throw. +false signals a failed send without throwing. `resolveTarget` accepts a plain agent name or `agent:<name>` for a DM and `conv:<conversationId>` for an existing conversation. Plain names normalize @@ -64,7 +58,7 @@ to `agent:<name>`. Other colon-prefixed shapes are rejected. **Returns:** The created moltzap channel plugin. -### [`default`](./openclaw-entry.ts#L1325) +### [`default`](./openclaw-entry.ts#L1256) _Variable_ @@ -72,7 +66,7 @@ _Variable_ const plugin = ``` -### [`moltzapChannelPlugin`](./openclaw-entry.ts#L1322) +### [`moltzapChannelPlugin`](./openclaw-entry.ts#L1253) _Variable_ @@ -85,7 +79,7 @@ Shared singleton so a single registration reuses the same `activeClients` closure across `startAccount` and `sendText`. Tests import this directly to assert against that shared state. -### [`MoltzapChannelPlugin`](./openclaw-entry.ts#L1313) +### [`MoltzapChannelPlugin`](./openclaw-entry.ts#L1244) _TypeAlias_ diff --git a/packages/openclaw-channel/src/openclaw-entry.delivery.test.ts b/packages/openclaw-channel/src/openclaw-entry.delivery.test.ts index e1ac68790..581259dd8 100644 --- a/packages/openclaw-channel/src/openclaw-entry.delivery.test.ts +++ b/packages/openclaw-channel/src/openclaw-entry.delivery.test.ts @@ -119,7 +119,6 @@ class SendToAgentTestFailure extends Data.TaggedError( const mockSend = vi.fn(); const mockSendToAgent = vi.fn(); -const mockOnDuplicateReply = vi.fn<(conversationId: string) => void>(); let started: { readonly fixture: FakeChannelService; @@ -146,10 +145,7 @@ describe("Flow 6: Outbound delivery - deliver callback + sendText", () => { "does not report a rejected inbound dispatch as finished", rejectedDispatchIsNotFinished, ); - it( - "deliver callback rejects duplicate final delivery", - rejectsDuplicateFinal, - ); + it("each final delivery sends a reply", sendsEachFinalDelivery); it("deliver callback returns true for non-final replies", nonFinalIsIgnored); it("sendText sends to the right conversation", sendsToConversation); it("resolveTarget accepts agent targets", acceptsAgentTarget); @@ -162,10 +158,7 @@ describe("Flow 6: Outbound delivery - deliver callback + sendText", () => { it("sendText reports disconnected clients", reportsDisconnectedClient); it("sendText reports send failures", reportsSendFailure); it("deliver reports transient RPC send failures", sendFailureIsReported); - it( - "reply guard stays unconsumed on transient send failure", - replyGuardUnconsumedOnTransientFailure, - ); + it("a later delivery retries after a send failure", retriesAfterSendFailure); it("stopAccount removes client from active pool", stopRemovesClient); it( "property: resolveTarget normalizes generated agent names", @@ -183,7 +176,6 @@ function startGateway() { const service = createTestService(fixture); const plugin = createMoltzapChannelPlugin({ createService: () => service, - onDuplicateReply: mockOnDuplicateReply, }); const abortController = new AbortController(); abortControllers.push(abortController); @@ -399,7 +391,7 @@ function rejectedDispatchIsNotFinished() { }); } -function rejectsDuplicateFinal() { +function sendsEachFinalDelivery() { return Effect.gen(function* () { yield* emitMessage(); yield* waitForDispatchTimes(1); @@ -409,9 +401,8 @@ function rejectsDuplicateFinal() { const second = yield* deliverFinal(SECOND_REPLY_TEXT); expect(first).toBe(true); expect(sendAfterFirst).toBe(sendBefore + 1); - expect(second).toBe(false); - expect(mockSend.mock.calls.length).toBe(sendAfterFirst); - expect(mockOnDuplicateReply).toHaveBeenCalledWith(DEFAULT_CONVERSATION_ID); + expect(second).toBe(true); + expect(mockSend.mock.calls.length).toBe(sendAfterFirst + 1); }); } @@ -582,16 +573,7 @@ function sendFailureIsReported() { }); } -/** - * A transient `core.sendReply` failure MUST leave the per-turn `ReplyGuard` - * unconsumed, so a retried `deliver(...)` still drives the send path. The - * guard is stamped via `Effect.tap` only after a successful sendReply - * (`openclaw-entry.ts → createReplyGuardedDeliver` + `sendDeliveredReply`); - * stamping earlier would make a first failure permanently short-circuit - * every retry to `false` without re-calling `core.sendReply`. - * @returns The reply guard unconsumed on transient failure result. - */ -function replyGuardUnconsumedOnTransientFailure() { +function retriesAfterSendFailure() { return Effect.gen(function* () { mockSend.mockReturnValueOnce( Effect.fail( @@ -606,10 +588,6 @@ function replyGuardUnconsumedOnTransientFailure() { const first = yield* deliverFinal(FIRST_REPLY_TEXT); expect(first).toBe(false); expect(mockSend.mock.calls.length).toBe(sendBefore + 1); - // Second deliver: send is now configured to succeed (default - // `mockSend.mockImplementation(fixture.service.send)` from `startGateway`). - // The guard MUST NOT have been stamped by the first failure, so this - // retry reaches the send path and returns true. const second = yield* deliverFinal(SECOND_REPLY_TEXT); expect(second).toBe(true); expect(mockSend.mock.calls.length).toBe(sendBefore + 2); diff --git a/packages/openclaw-channel/src/openclaw-entry.ts b/packages/openclaw-channel/src/openclaw-entry.ts index ddfe8a93c..80246d8d3 100644 --- a/packages/openclaw-channel/src/openclaw-entry.ts +++ b/packages/openclaw-channel/src/openclaw-entry.ts @@ -1,5 +1,3 @@ -/* eslint-disable jsdoc/text-escaping -- mermaid sequenceDiagram blocks need literal `
` (HTML5) for renderer compatibility; the escape would render as literal text. */ - /** * OpenClaw plugin entry point for MoltZap. * @@ -19,7 +17,6 @@ import { MoltZapService, type ServiceRpcError } from "@moltzap/client"; import { drainPaginatedList } from "@moltzap/client/pagination"; import { MoltZapChannelCore, - ReplyGuard, formatCrossConv, getGroupFields, type ChannelService, @@ -299,17 +296,6 @@ interface MoltzapChannelPluginDeps { account: MoltZapAccount, ) => OpenClawClientService; readonly createCore?: (service: ChannelService) => MoltZapChannelCore; - - /** - * Invoked when a second final reply for the same inbound turn is suppressed - * locally. Deliver still returns `false` per the - * `OpenClawDeliver: PromiseLike<boolean>` contract; this callback is - * the side-channel for hosts that want to observe the duplicate without - * violating that contract. Threaded from `createMoltzapChannelPlugin` deps - * through `createGatewaySection` and `registerInboundHandler` into the - * `createReplyGuardedDeliver` closure. - */ - readonly onDuplicateReply?: (conversationId: string) => void; } interface OpenClawDirectoryParams { @@ -333,7 +319,6 @@ interface InboundHandlerParams { readonly service: OpenClawClientService; readonly contextLogDir?: string; readonly enriched: EnrichedInboundMessage; - readonly onDuplicateReply?: (conversationId: string) => void; } interface InboundRuntimeData { @@ -429,45 +414,23 @@ function sendDeliveredReply(params: { readonly conversationId: ConversationId; readonly text: string; readonly log?: OpenClawLogger; - readonly guard: ReplyGuard; }): Effect.Effect { return params.core.sendReply(params.conversationId, params.text).pipe( - // Stamp the guard only on a successful `core.sendReply`. Transient send - // failures reopen the claimed guard via `abort()` below WITHOUT - // stamping, so a retried deliver call still gets to send. - Effect.tap(() => params.guard.consume()), Effect.tap(() => logOutboundReply(params.conversationId, params.text, params.log), ), Effect.map(() => true), Effect.catchAll((err) => - params.guard - .abort() - .pipe( - Effect.zipRight( - handleReplyFailure(params.conversationId, err, params.log), - ), - ), + handleReplyFailure(params.conversationId, err, params.log), ), ); } -function createReplyGuardedDeliver(params: { +function createReplyDeliver(params: { readonly core: MoltZapChannelCore; readonly enriched: EnrichedInboundMessage; readonly log?: OpenClawLogger; - readonly onDuplicateReply?: (conversationId: string) => void; }): OpenClawDeliver { - // One ReplyGuard per inbound turn: stamped exactly once, on the FIRST - // successful `core.sendReply`. It is the only thing keeping a runtime that - // delivers twice from double-posting — the server accepts every send. - // - // Ordering matters: pre-check `guard.consumedAt` BEFORE sending so duplicate - // delivers are short-circuited; the actual stamp happens inside - // `sendDeliveredReply` via `Effect.tap(() => guard.consume())` AFTER - // `core.sendReply` succeeds. A transient send failure therefore leaves the - // guard unconsumed, and a retried deliver call still gets to send. - const guard = new ReplyGuard(); return (payload, info) => { if (info?.kind !== "final") { return Promise.resolve(true); @@ -477,31 +440,11 @@ function createReplyGuardedDeliver(params: { return Promise.resolve(true); } return Effect.runPromise( - Effect.gen(function* () { - // begin() claims the guard in one synchronous step, so two - // concurrent final delivers cannot both pass the duplicate check - // while a send is mid-flight. - const claimed = yield* guard.begin(); - if (!claimed) { - const stamped = yield* guard.consumedAt; - const detail = Option.isSome(stamped) - ? ` (first reply sent at ts=${stamped.value.toString()})` - : " (another reply is in flight)"; - params.log?.warn?.( - `MoltZap: duplicate reply rejected for ${params.enriched.conversationId}${detail}`, - ); - params.onDuplicateReply?.(params.enriched.conversationId); - // Deliver contract: PromiseLike. False signals - // "not delivered" without violating the type. - return false; - } - return yield* sendDeliveredReply({ - core: params.core, - conversationId: params.enriched.conversationId, - text, - log: params.log, - guard, - }); + sendDeliveredReply({ + core: params.core, + conversationId: params.enriched.conversationId, + text, + log: params.log, }), ); }; @@ -555,7 +498,6 @@ function dispatchInboundReply(params: { readonly input: InboundDispatchInput; readonly core: MoltZapChannelCore; readonly log?: OpenClawLogger; - readonly onDuplicateReply?: (conversationId: string) => void; }): Effect.Effect<{ queuedFinal: boolean }, unknown> { return Effect.tryPromise({ try: () => @@ -563,11 +505,10 @@ function dispatchInboundReply(params: { ctx: buildInboundDispatchContext(params.input), cfg: params.input.cfg, dispatcherOptions: { - deliver: createReplyGuardedDeliver({ + deliver: createReplyDeliver({ core: params.core, enriched: params.input.enriched, log: params.log, - onDuplicateReply: params.onDuplicateReply, }), }, }), @@ -815,7 +756,6 @@ function startGatewayAccountEffect( ctx, service, contextLogDir, - onDuplicateReply: deps.onDuplicateReply, }); registerConnectionStatus(core, ctx); activeClients.set(accountId, service); @@ -869,7 +809,6 @@ interface RegisterInboundHandlerParams { readonly ctx: OpenClawStartAccountContext; readonly service: OpenClawClientService; readonly contextLogDir?: string; - readonly onDuplicateReply?: (conversationId: string) => void; } function registerInboundHandler(params: RegisterInboundHandlerParams): void { @@ -880,7 +819,6 @@ function registerInboundHandler(params: RegisterInboundHandlerParams): void { service: params.service, contextLogDir: params.contextLogDir, enriched, - onDuplicateReply: params.onDuplicateReply, }).pipe(Effect.withSpan("createMoltzapChannelPlugin.inboundDispatch")), ); } @@ -904,7 +842,6 @@ function handleInboundMessage(params: InboundHandlerParams) { input: inboundDispatchInput(params.ctx, params.enriched, data), core: params.core, log: params.ctx.log, - onDuplicateReply: params.onDuplicateReply, }); logDispatchFinished(params.enriched, params.ctx.log); logUnqueuedDispatch(params.enriched, result, params.ctx.log); @@ -1270,21 +1207,15 @@ function sendTextEffect( * Core->>Plugin: enriched message arrives * Plugin->>OC: dispatchReplyWithBufferedBlockDispatcher * note over OC: agent pipeline → LLM - * OC->>Plugin: deliver(payload, opts) — createReplyGuardedDeliver + * OC->>Plugin: deliver(payload, opts) — createReplyDeliver * Plugin->>Server: core.sendReply(conversationId, text) - * alt second final reply for the same turn - * Plugin->>Plugin: ReplyGuard already stamped
onDuplicateReply callback, return false - * end * OC->>Plugin: stopAccount(ctx) * Plugin->>Core: core.disconnect() * Plugin->>Plugin: activeClients.delete(account) * ``` * * `deliver` returns `PromiseLike<boolean>` per openclaw contract; - * false signals "not delivered" without throwing. The reply guard is - * single-shot per inbound turn: a second final reply is suppressed locally - * and reported through `MoltzapChannelPluginDeps.onDuplicateReply` rather - * than a throw. + * false signals a failed send without throwing. * * `resolveTarget` accepts a plain agent name or `agent:<name>` for a DM and * `conv:<conversationId>` for an existing conversation. Plain names normalize @@ -1336,5 +1267,3 @@ const plugin = { // eslint-disable-next-line import-x/no-default-export -- OpenClaw discovers channel plugins through a required default module export. export default plugin; - -/* eslint-enable jsdoc/text-escaping -- Restore strict defaults after the scoped file-level exception. */ diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index f5df8a64b..5260537e1 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -86,6 +86,9 @@ importers: '@effect/typeclass': specifier: ^0.41.0 version: 0.41.0(effect@3.22.0) + '@modelcontextprotocol/client': + specifier: 2.0.0-beta.5 + version: 2.0.0-beta.5 '@modelcontextprotocol/node': specifier: 2.0.0-beta.5 version: 2.0.0-beta.5(@modelcontextprotocol/server@2.0.0-beta.5)(hono@4.12.9) @@ -102,9 +105,6 @@ importers: '@effect/vitest': specifier: ^0.30.0 version: 0.30.0(effect@3.22.0)(vitest@3.2.4(@types/debug@4.1.13)(@types/node@25.5.0)(jiti@2.7.0)(tsx@4.21.0)(yaml@2.9.0)) - '@modelcontextprotocol/client': - specifier: 2.0.0-beta.5 - version: 2.0.0-beta.5 '@moltzap/server-core': specifier: workspace:* version: link:../server diff --git a/scripts/gen-architecture-configs.mjs b/scripts/gen-architecture-configs.mjs index 15771b2c6..b15e4f7a6 100644 --- a/scripts/gen-architecture-configs.mjs +++ b/scripts/gen-architecture-configs.mjs @@ -73,6 +73,9 @@ const packageDefinitions = { client: { beforeShared: { minExportedSiblingModules: 6, + // HarnessClient is an intentional runtime-facing entrypoint alongside + // the existing package surfaces. + maxSubpathExports: 6, maxPublicExports: 29, // channel-base names the adapter primitives, and BoundedMap is one of // them; the rule counts local re-exports, so owning that module in-package