From 9a4150c090532b09d9e60e952d45158ab52f0ad4 Mon Sep 17 00:00:00 2001 From: Tapan Chugh Date: Wed, 5 Aug 2026 11:22:16 -0700 Subject: [PATCH] refactor(nanoclaw): own one Harness client for the adapter's lifetime The adapter carried two constructors and production only ever reached the one that built MoltZapService and MoltZapChannelCore directly, because nothing could resolve a daemon endpoint. The slot now carries its port, so the profile name alone resolves a client. NanoClaw registers its channel through a zero-argument factory that runs at module import, so there is no seam to inject a client through. The adapter therefore owns the scope itself: it acquires in setup and releases in teardown. That inverts the borrowed-client contract, so the borrowing constructor goes rather than surviving beside it. The simulator writes the slot and stops there. It does not start a daemon: the adapter starts its own, and a real nanoclaw checkout has no simulator to start one for it. Two daemons would bind the same slot's single port. The channel stays one file. The asset copier installs exactly this module into a downloaded checkout, so acquisition lives here or behind a client export, never in a sibling. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01P76aaa1STr3WPZ3nDascta --- packages/client/AGENTS.md | 5 +- packages/nanoclaw-channel/AGENTS.md | 53 +- .../src/__tests__/echo.integration.test.ts | 502 +++--- .../src/channels/moltzap.test.ts | 1343 +++++++++-------- .../nanoclaw-channel/src/channels/moltzap.ts | 368 ++--- .../harness-adapters.integration.test.ts | 106 +- .../simulator/src/runtime/nanoclaw/process.ts | 8 + 7 files changed, 1134 insertions(+), 1251 deletions(-) diff --git a/packages/client/AGENTS.md b/packages/client/AGENTS.md index 177f817d6..64b393869 100644 --- a/packages/client/AGENTS.md +++ b/packages/client/AGENTS.md @@ -42,8 +42,9 @@ Subpath exports: `./channel-base`, `./harness-client`, `./test-utils`, `./auth`, ## Concepts - **Channel adapter** — a package bridging MoltZap to an agent - runtime (openclaw, nanoclaw). Each wraps `MoltZapChannelCore` and - shares the channel-base primitives. + runtime (openclaw, nanoclaw). Each consumes a `HarnessClient` over its + slot's loopback MCP surface and shares the channel-base primitives. + `MoltZapChannelCore` sits behind that boundary, inside `moltzapd`. - **Turn** — one `InboundHandler` invocation. Turn-taking is endpoint-local: the server delivers every message it accepts. A single consumer fiber awaits the handler inline, so one turn runs diff --git a/packages/nanoclaw-channel/AGENTS.md b/packages/nanoclaw-channel/AGENTS.md index 8b4e6d3d1..d39d7bf29 100644 --- a/packages/nanoclaw-channel/AGENTS.md +++ b/packages/nanoclaw-channel/AGENTS.md @@ -8,11 +8,13 @@ channel plugins. ## Structure - `src/channels/moltzap.ts` — `MoltZapAdapter`, the entry point - (package `main`); implements nanoclaw's `ChannelAdapter` contract - over an injected `HarnessClient` or the transitional - `MoltZapChannelCore` path and self-registers via - `registerChannelAdapter`. The production factory remains profile/core-backed - until profile-to-MCP acquisition is available. + (package `main`); implements nanoclaw's `ChannelAdapter` contract over a + Harness client whose lifetime it owns, and self-registers via + `registerChannelAdapter`. This file is the whole channel: the simulator's + asset copier (`packages/simulator/scripts/copy-nanoclaw-assets.mjs`) + copies exactly it, so a sibling module added beside it does not exist at + nanoclaw runtime. New logic belongs in this file or behind a + `@moltzap/client` export. - `src/channels/adapter.ts`, `src/channels/channel-registry.ts`, `src/db/messaging-groups.ts`, `src/types.ts` — stub mirrors of the nanoclaw modules the channel imports, pinned to the commit in `NANOCLAW_SHA` @@ -37,16 +39,21 @@ channel plugins. ## Code -- The injected Harness path drains `HarnessClient.turns` sequentially and - retains each turn's bound `reply` closure by jid. NanoClaw may call - `deliver` asynchronously after `onInbound` returns, so the closure remains - available until a newer inbound for that conversation replaces it or the - bounded entry is evicted. -- `fromHarnessClient` borrows an already acquired client. Adapter teardown - interrupts its turn drain but does not close the caller-owned client scope. -- `MoltZapChannelError` covers host-shape failures (un-owned jid, - unknown conversation, disconnected channel); reply failures retain their - backing client's error type. +- The adapter drains `HarnessClient.turns` sequentially and retains each + turn's bound `reply` closure by jid. NanoClaw may call `deliver` + asynchronously after `onInbound` returns, so the closure remains available + until a newer inbound for that conversation replaces it or the bounded + entry is evicted. +- `fromHarnessAcquisition` is the only constructor, and the adapter owns the + acquisition's `Scope`: `setup` opens it, `teardown` closes it. NanoClaw + builds channel adapters from a zero-argument factory at module import, so + no caller exists to hold that scope. `makeMoltZapAdapter` supplies + `harnessClientForProfile(MOLTZAP_PROFILE)`, which resolves the slot into + its own `moltzapd` child, the loopback endpoint the slot names, and a + file-backed checkpoint store. +- `MoltZapChannelError` covers host-shape failures (un-owned jid, unknown + conversation, a host callback that rejects a projected turn); reply + failures retain their backing client's error type. - Inbound projection: `onMetadata` fires before `onInbound`; content is `{ text, sender, senderId }` with context blocks inlined into `text`; own (`isFromMe`) messages are dropped, not delivered. @@ -58,10 +65,12 @@ channel plugins. - `vitest.integration.globalSetup.ts` spawns the standalone server on PGlite, registers two agents, and `provide`s base/WS URLs plus per-agent IDs and API keys; inject keys are typed in - `src/__tests__/vitest-provided.d.ts`. -- The adapter currently connects once during setup and logs a nonterminal - disconnect. It does not yet drive reconnect or missed-message catch-up; - the gated full-agent evaluation covers the initial live connection path. -- Harness behavior tests use a fake `HarnessClientService` stream and bound - reply closures. Import/constructor absence remains an architecture check for - the later production-factory cutover, not a unit assertion. + `src/__tests__/vitest-provided.d.ts`. The echo suite reserves the slot's + loopback port, writes the slot, and drives `makeMoltZapAdapter` — the same + adapter nanoclaw registers — so a real `moltzapd` carries the round trip. +- The adapter connects once during setup and logs a nonterminal disconnect. + It does not drive reconnect or missed-message catch-up; the gated + full-agent evaluation covers the initial live connection path. +- Unit tests drive a fake `HarnessClientService` through a counted + acquisition, so acquire/release counts assert what `setup` and `teardown` + did to the client's lifetime. diff --git a/packages/nanoclaw-channel/src/__tests__/echo.integration.test.ts b/packages/nanoclaw-channel/src/__tests__/echo.integration.test.ts index 90acca093..445309339 100644 --- a/packages/nanoclaw-channel/src/__tests__/echo.integration.test.ts +++ b/packages/nanoclaw-channel/src/__tests__/echo.integration.test.ts @@ -1,47 +1,65 @@ /** * Echo integration test for `@moltzap/nanoclaw-channel`. * - * Boots `MoltZapAdapter` against a real MoltZap server (PGlite-backed, - * spawned by the global setup), uses a peer `MoltZapService` to drive - * inbound messages, and verifies the host-facing callbacks - * (`setup.onInbound`, `setup.onMetadata`) fire with the expected shape - * and `deliver(jid, null, message)` round-trips back to the peer. + * Drives the adapter nanoclaw itself registers: the production factory, which + * resolves a profile slot into its own `moltzapd` child, the loopback MCP + * endpoint the slot names, and a file-backed checkpoint store. A peer agent + * on the same server (PGlite-backed, spawned by the global setup) opens a DM + * and sends into it, so the assertions cover the whole path — daemon startup, + * turn projection, the host-facing callbacks, and `deliver` back to the peer. */ -/* eslint-disable agent-code-guard/no-effect-error-coalescing -- test scaffolding coalesces wire-level Service/Rpc errors into a single test-context error class for cleaner diagnostic output; production rule does not apply to integration test scaffolding. */ - -import { afterAll, beforeAll, describe, expect, inject } from "vitest"; +import { describe, expect, inject } from "vitest"; import { live as it } from "@effect/vitest"; -import { Data, Effect, Schema } from "effect"; -import { MoltZapService } from "@moltzap/client"; -import { withTestServiceConfig } from "@moltzap/client/test-utils"; +import { + Data, + Deferred, + Effect, + Fiber, + Option, + Schema, + type Scope, + Stream, +} from "effect"; +import { MoltZapAgentClient } from "@moltzap/client"; +import { + reserveTestMcpPort, + withTestServiceConfig, +} from "@moltzap/client/test-utils"; import { type AgentKey, agentKey, type AgentId, } from "@moltzap/protocol/identity"; -import type { Message } from "@moltzap/protocol/message"; -import { serverBaseUrl } from "@moltzap/protocol/network"; import { - agentConversationCreate, - type ConversationId, -} from "@moltzap/protocol/conversation"; + messageReceivedNotificationDefinition, + messagesSend, + type Message, +} from "@moltzap/protocol/message"; +import { agentConversationCreate } from "@moltzap/protocol/conversation"; import { agentId as makeAgentId } from "@moltzap/protocol/testing"; -import { MoltZapAdapter } from "../channels/moltzap.js"; +import { + makeMoltZapAdapter, + type MoltZapAdapter, +} from "../channels/moltzap.js"; import type { ChannelSetup, InboundMessage, OutboundMessage, } from "../channels/adapter.js"; -class EchoIntegrationError extends Data.TaggedError("EchoIntegrationError")<{ - readonly operation: string; - readonly cause: unknown; -}> {} +/** The production factory refused the profile slot this suite just wrote. */ +class MissingAdapterError extends Data.TaggedError("MissingAdapterError")< + Record +> { + override get message(): string { + return "the production factory returned no adapter"; + } +} interface InjectedConfig { - readonly wsUrl: string; + readonly baseUrl: string; readonly channelApiKey: AgentKey; readonly peerApiKey: AgentKey; readonly channelAgentId: AgentId; @@ -59,22 +77,8 @@ interface ChatMetadataCapture { readonly isGroup?: boolean; } -interface Harness { - readonly adapter: MoltZapAdapter; - readonly peerService: MoltZapService; - readonly inboundMessages: InboundCapture[]; - readonly chatMetadata: ChatMetadataCapture[]; - readonly peerInbox: Message[]; - readonly conversationId: ConversationId; - readonly chatJid: string; - readonly peerAgentId: string; - readonly stop: () => PromiseLike; -} - -const WAIT_FOR_TICK_MS = 25; -const INBOUND_NOTIFICATION_TIMEOUT_MS = 15_000; -const PING_ONE = "ping-one"; -const PING_TWO = "ping-two"; +const REPLY_TIMEOUT = "20 seconds"; +const PING = "ping-one"; const TEXT_TYPE = "text"; const ECHO_PREFIX = "echo-"; const CHANNEL_PROFILE_NAME = "channel-agent"; @@ -82,7 +86,13 @@ const MOLTZAP_CHANNEL_NAME = "moltzap"; const JID_PREFIX = "mz:"; const OUTBOUND_KIND_CHAT = "chat"; -let h: Harness; +const toError = (cause: unknown): Error => + cause instanceof Error ? cause : new Error(String(cause)); + +const tryPromise = ( + evaluate: () => PromiseLike, +): Effect.Effect => + Effect.tryPromise({ try: evaluate, catch: toError }); function injectString(key: string): string { return inject( @@ -90,9 +100,13 @@ function injectString(key: string): string { ); } +function decodeInjectedAgentKey(key: string): AgentKey { + return Schema.decodeUnknownSync(agentKey)(injectString(key)); +} + function injectedConfig(): InjectedConfig { return { - wsUrl: injectString("moltzapWsUrl"), + baseUrl: injectString("moltzapBaseUrl"), channelApiKey: decodeInjectedAgentKey("agentAApiKey"), peerApiKey: decodeInjectedAgentKey("agentBApiKey"), channelAgentId: makeAgentId(injectString("agentAAgentId")), @@ -100,10 +114,6 @@ function injectedConfig(): InjectedConfig { }; } -function decodeInjectedAgentKey(key: string): AgentKey { - return Schema.decodeUnknownSync(agentKey)(injectString(key)); -} - function contentText(msg: InboundMessage): string { return ( /* Safe because the test fixture establishes this asserted shape. */ @@ -111,298 +121,182 @@ function contentText(msg: InboundMessage): string { ); } -function channelSenderId(agentId: string): string { - return `${MOLTZAP_CHANNEL_NAME}:${agentId}`; +function senderIdOf(msg: InboundMessage): string { + return ( + /* Safe because the test fixture establishes this asserted shape. */ + (msg.content as { readonly senderId: string }).senderId + ); } function makeOutbound(text: string): OutboundMessage { return { kind: OUTBOUND_KIND_CHAT, content: { text } }; } -function tryPromise( - operation: string, - evaluate: () => PromiseLike, -): Effect.Effect { - return Effect.tryPromise({ - try: evaluate, - catch: (cause) => new EchoIntegrationError({ operation, cause }), - }); +function messageText(message: Message): string { + return message.parts + .flatMap((part) => (part.type === TEXT_TYPE ? [part.text] : [])) + .join(""); } -function waitFor( - predicate: () => boolean, - timeoutMs: number, - label: string, -): Effect.Effect { - return Effect.tryPromise({ - try: () => waitForPromise(predicate, timeoutMs, label), - catch: (cause) => - new EchoIntegrationError({ operation: `waitFor(${label})`, cause }), - }); +/** + * Nanoclaw's host contract: the router calls `deliver` from its own turn + * handling, so the echo runs inside `onInbound` exactly where a session's + * model output would. + * @param adapter Adapter under test. + * @param inbound Resolved with the first host-facing inbound message. + * @param metadata Chat-metadata events observed for the conversation. + * @returns The setup nanoclaw would install. + */ +function echoSetup( + adapter: MoltZapAdapter, + inbound: Deferred.Deferred, + metadata: ChatMetadataCapture[], +): ChannelSetup { + return { + onInbound: (...[jid, , msg]) => { + Effect.runSync(Deferred.succeed(inbound, { jid, msg })); + return adapter.deliver( + jid, + null, + makeOutbound(`${ECHO_PREFIX}${contentText(msg)}`), + ); + }, + onMetadata: (jid, name, isGroup) => { + metadata.push({ jid, name, isGroup }); + }, + }; } -function waitForPromise( - predicate: () => boolean, - timeoutMs: number, - label: string, -) { - return new Promise((resolve, reject) => { - const start = Date.now(); - const tick = (): void => { - if (predicate()) { - resolve(undefined); - return; - } - if (Date.now() - start > timeoutMs) { - reject(new Error(`waitFor(${label}) timed out`)); - return; +function acquireAdapter( + inbound: Deferred.Deferred, + metadata: ChatMetadataCapture[], +): Effect.Effect { + return Effect.acquireRelease( + Effect.suspend(() => { + const adapter = makeMoltZapAdapter({ + profileName: CHANNEL_PROFILE_NAME, + evalMode: false, + }); + if (adapter === null) { + return new MissingAdapterError(); } - setTimeout(tick, WAIT_FOR_TICK_MS); - }; - tick(); - }); + return tryPromise(() => + adapter.setup(echoSetup(adapter, inbound, metadata)), + ).pipe(Effect.as(adapter)); + }), + (adapter) => tryPromise(() => adapter.teardown()).pipe(Effect.ignore), + ); } -function makeAdapter( +function acquirePeer( config: InjectedConfig, - inboundMessages: InboundCapture[], - chatMetadata: ChatMetadataCapture[], -): Effect.Effect { - return Effect.gen(function* () { - const adapter = MoltZapAdapter.fromProfile(CHANNEL_PROFILE_NAME, false); - const setup: ChannelSetup = { - onInbound: (...args) => { - const [jid, , msg] = args; - inboundMessages.push({ jid, msg }); - autoEcho(adapter, jid, contentText(msg)); - }, - onMetadata: (jid, name, isGroup) => { - chatMetadata.push({ jid, name, isGroup }); - }, - }; - yield* withTestServiceConfig( - { - agentId: config.channelAgentId, - agentKey: config.channelApiKey, - serverUrl: config.wsUrl, - profileName: CHANNEL_PROFILE_NAME, - agentName: CHANNEL_PROFILE_NAME, - }, - tryPromise("adapter.setup", () => adapter.setup(setup)), - ); - return adapter; - }); -} - -function autoEcho(adapter: MoltZapAdapter, jid: string, content: string): void { - // Auto-echo failures (e.g. retries on a closed dispatch during teardown) - // are absorbed: the host-facing test asserts at the peer-inbox boundary - // so individual deliver failures do not invalidate the test. Modeled - // as a fire-and-forget Effect (the simulated host responding to inbound). - Effect.runFork( - Effect.tryPromise({ - try: () => - adapter.deliver(jid, null, makeOutbound(`${ECHO_PREFIX}${content}`)), - catch: noopOnError, - }).pipe(Effect.ignore), +): Effect.Effect { + return Effect.acquireRelease( + Effect.suspend(() => { + const peer = new MoltZapAgentClient({ + serverUrl: config.baseUrl, + agentKey: config.peerApiKey, + }); + return peer.connect().pipe(Effect.mapError(toError), Effect.as(peer)); + }), + (peer) => peer.close().pipe(Effect.ignore), ); } -function noopOnError(): void { - // Intentional no-op: auto-echo loop swallows transient failures. -} - -function bootPeerService( +function takeChannelReply( + peer: MoltZapAgentClient, config: InjectedConfig, - peerInbox: Message[], -): Effect.Effect { - return Effect.succeed( - MoltZapService.fromConfig({ - agentId: config.peerAgentId, - agentKey: config.peerApiKey, - serverUrl: serverBaseUrl(config.wsUrl), + conversationId: string, +): Effect.Effect { + return peer.subscribe(messageReceivedNotificationDefinition).pipe( + Stream.filter( + ({ message }) => + message.senderId === config.channelAgentId && + message.conversationId === conversationId, + ), + Stream.runHead, + Effect.timeoutFail({ + duration: REPLY_TIMEOUT, + onTimeout: () => new Error("timed out waiting for the adapter echo"), }), - ).pipe( - Effect.tap((peerService) => - Effect.sync(() => { - peerService.on("message", (payload) => { - peerInbox.push(payload.message); - }); + Effect.flatMap( + Option.match({ + onNone: () => + Effect.die(new Error("peer reply stream closed before delivery")), + onSome: ({ message }) => Effect.succeed(message), }), ), + Effect.mapError(toError), ); } -function createDm( - peerService: MoltZapService, - channelAgentId: AgentId, -): Effect.Effect<{ conversationId: ConversationId }, EchoIntegrationError> { - return peerService - .call(agentConversationCreate.name, { - participants: [channelAgentId], - }) - .pipe( - Effect.map((res) => ({ conversationId: res.conversation.id })), - Effect.mapError( - (cause) => new EchoIntegrationError({ operation: "createDm", cause }), - ), - ); -} - -function connectPeerService( - peerService: MoltZapService, -): Effect.Effect { - return peerService.connect().pipe( - Effect.mapError( - (cause) => - new EchoIntegrationError({ - operation: "peerService.connect", - cause, - }), - ), - ); -} - -function makeHarness( - config: InjectedConfig, -): Effect.Effect { +function runEchoExchange(config: InjectedConfig) { return Effect.gen(function* () { - const inboundMessages: InboundCapture[] = []; - const chatMetadata: ChatMetadataCapture[] = []; - const peerInbox: Message[] = []; - const adapter = yield* makeAdapter( - config, - inboundMessages, - chatMetadata, - ).pipe( - Effect.mapError( - (cause) => - new EchoIntegrationError({ operation: "makeAdapter", cause }), - ), + const inbound = yield* Deferred.make(); + const metadata: ChatMetadataCapture[] = []; + const adapter = yield* acquireAdapter(inbound, metadata); + const peer = yield* acquirePeer(config); + + const created = yield* peer + .callDefinition(agentConversationCreate, { + participants: [config.channelAgentId], + }) + .pipe(Effect.mapError(toError)); + const conversationId = created.conversation.id; + const chatJid = `${JID_PREFIX}${conversationId}`; + + const echoFiber = yield* Effect.fork( + takeChannelReply(peer, config, conversationId), ); - const peerService = yield* bootPeerService(config, peerInbox).pipe( - Effect.mapError( - (cause) => - new EchoIntegrationError({ operation: "bootPeerService", cause }), - ), + yield* peer + .callDefinition(messagesSend, { + conversationId, + parts: [{ type: TEXT_TYPE, text: PING }], + }) + .pipe(Effect.mapError(toError)); + + const delivered = yield* Deferred.await(inbound).pipe( + Effect.timeoutFail({ + duration: REPLY_TIMEOUT, + onTimeout: () => new Error("timed out waiting for the host inbound"), + }), ); - yield* connectPeerService(peerService); - const { conversationId } = yield* createDm( - peerService, - config.channelAgentId, + expect(delivered.jid).toBe(chatJid); + expect(contentText(delivered.msg)).toBe(PING); + expect(senderIdOf(delivered.msg)).toBe( + `${MOLTZAP_CHANNEL_NAME}:${config.peerAgentId}`, ); - return { - adapter, - peerService, - inboundMessages, - chatMetadata, - peerInbox, - conversationId, - chatJid: `${JID_PREFIX}${conversationId}`, - peerAgentId: config.peerAgentId, - stop: () => stopAdapterAndPeer(adapter, peerService), - }; - }); -} -function stopAdapterAndPeer( - adapter: MoltZapAdapter, - peerService: MoltZapService, -) { - return Effect.runPromise( - Effect.gen(function* () { - yield* Effect.tryPromise({ - try: () => adapter.teardown(), - catch: () => undefined, - }).pipe(Effect.ignore); - peerService.close(); - return undefined; - }), - ); -} - -beforeAll(() => Effect.runPromise(initHarness())); -afterAll(() => stopHarness()); - -function initHarness() { - return Effect.gen(function* () { - h = yield* makeHarness(injectedConfig()); - }); -} - -function stopHarness() { - return h === undefined ? Promise.resolve(undefined) : h.stop(); -} - -function messageContains(message: Message, needle: string): boolean { - return message.parts.some( - (part) => part.type === TEXT_TYPE && part.text.includes(needle), - ); -} + // Metadata precedes the inbound dispatch, so it is already recorded by + // the time the inbound deferred resolves. + expect(metadata.some((entry) => entry.jid === chatJid)).toBe(true); + expect(adapter.isConnected()).toBe(true); -function inboundHas(needle: string): boolean { - return h.inboundMessages.some((c) => contentText(c.msg).includes(needle)); -} - -function peerInboxHas(needle: string): boolean { - return h.peerInbox.some((m) => messageContains(m, needle)); -} - -function peerSend(text: string): Effect.Effect { - return h.peerService - .send(h.conversationId, text) - .pipe( - Effect.mapError( - (cause) => - new EchoIntegrationError({ operation: "peerService.send", cause }), - ), - ); + const echo = yield* Fiber.join(echoFiber); + expect(messageText(echo)).toBe(`${ECHO_PREFIX}${PING}`); + }).pipe(Effect.scoped); } describe("nanoclaw echo integration", () => { - it( - "delivers inbound messages to the host onInbound callback", - deliversInbound, - ); - it("emits a chat-metadata event for the conversation", emitsChatMetadata); - it("deliver round-trips back to the peer's inbox", roundTripsToPeer); -}); - -function deliversInbound() { - return Effect.gen(function* () { - yield* peerSend(PING_ONE); - yield* waitFor( - () => inboundHas(PING_ONE), - INBOUND_NOTIFICATION_TIMEOUT_MS, - "ping-one inbound", - ); - const seen = h.inboundMessages.find((c) => - contentText(c.msg).includes(PING_ONE), - ); - expect(seen?.jid).toBe(h.chatJid); - expect( - /* Safe because the test fixture establishes this asserted shape. */ - (seen!.msg.content as { senderId: string }).senderId, - ).toBe(channelSenderId(h.peerAgentId)); - }); -} - -function emitsChatMetadata() { - return Effect.sync(() => { - expect(h.chatMetadata.some((m) => m.jid === h.chatJid)).toBe(true); - }); -} - -function roundTripsToPeer() { - return Effect.gen(function* () { - yield* peerSend(PING_TWO); - yield* waitFor( - () => peerInboxHas(`${ECHO_PREFIX}${PING_TWO}`), - INBOUND_NOTIFICATION_TIMEOUT_MS, - "echo-pong-two on peer", + it("round-trips a peer message through the production adapter", () => { + const config = injectedConfig(); + return Effect.scoped( + Effect.gen(function* () { + // The daemon binds exactly the port its slot records, so the port is + // reserved here and written into the slot before the adapter starts. + const mcpPort = yield* reserveTestMcpPort; + return yield* withTestServiceConfig( + { + profileName: CHANNEL_PROFILE_NAME, + agentName: CHANNEL_PROFILE_NAME, + agentId: config.channelAgentId, + agentKey: config.channelApiKey, + serverUrl: config.baseUrl, + mcpPort, + }, + runEchoExchange(config), + ); + }), ); - expect(peerInboxHas(`${ECHO_PREFIX}${PING_TWO}`)).toBe(true); }); -} - -/* eslint-enable agent-code-guard/no-effect-error-coalescing -- Restore strict defaults after the scoped file-level exception. */ +}); diff --git a/packages/nanoclaw-channel/src/channels/moltzap.test.ts b/packages/nanoclaw-channel/src/channels/moltzap.test.ts index f2e8dbb79..05e6b9ee3 100644 --- a/packages/nanoclaw-channel/src/channels/moltzap.test.ts +++ b/packages/nanoclaw-channel/src/channels/moltzap.test.ts @@ -5,20 +5,21 @@ import type { HarnessClientService, HarnessTurn, } from "@moltzap/client/harness-client"; +import type { + CrossConvMessage, + EnrichedConversationMeta, +} from "@moltzap/client/channel-base"; import { - buildMessage, - createFakeChannelService, - flushDispatchChain, testAgentId, testConversationId, testMessageId, - type FakeChannelService, } from "@moltzap/client/test-utils"; import { EVAL_AGENT_GROUP_ID, makeMoltZapAdapter, MoltZapAdapter, + type HarnessClientAcquisition, } from "./moltzap.js"; import type { ChannelSetup, @@ -53,21 +54,47 @@ interface RecordedChannelSetup extends ChannelSetup { readonly callOrder: string[]; } -interface Harness { - readonly fake: FakeChannelService; - readonly config: RecordedChannelSetup; - readonly adapter: MoltZapAdapter; -} - interface HarnessClientReply { readonly route: string; readonly payload: string; } -interface HarnessClientFixture { - readonly client: HarnessClientService; +/** Counts how often the adapter opened and closed its client acquisition. */ +interface AcquisitionCounts { + acquired: number; + released: number; +} + +interface Harness { + readonly adapter: MoltZapAdapter; + readonly config: RecordedChannelSetup; + readonly counts: AcquisitionCounts; readonly replies: HarnessClientReply[]; readonly turns: Queue.Queue; + readonly signal: Queue.Queue; +} + +interface TurnOptions { + readonly conversationId: string; + readonly messageId?: string; + readonly route?: string; + readonly text?: string; + readonly senderId?: string; + readonly senderName?: string; + readonly isFromMe?: boolean; + readonly conversationMeta?: EnrichedConversationMeta; + readonly crossConversationMessages?: readonly CrossConvMessage[]; +} + +interface HarnessOptions { + readonly evalMode?: boolean; + readonly replies?: HarnessClientReply[]; + readonly turns?: Stream.Stream; + readonly config?: RecordedChannelSetup; + readonly acquire?: ( + client: HarnessClientService, + counts: AcquisitionCounts, + ) => HarnessClientAcquisition; } const AGENT_SELF = "agent-self"; @@ -94,7 +121,6 @@ const MSG_TURN_1 = "msg-turn-1"; const MSG_TURN_2 = "msg-turn-2"; const MSG_EVAL_1 = "msg-eval-1"; const MSG_EVAL_2 = "msg-eval-2"; -const HELLO_THERE = "hello there"; const FIRST_REPLY = "first reply"; const SECOND_REPLY = "second reply"; const HI_NANOCLAW = "hi nanoclaw"; @@ -108,7 +134,7 @@ const ZENDA_TEXT = "Zenda"; const CONTENT_TEXT = "content"; const MESSAGE_CREATED_AT = "2026-04-10T13:00:00.000Z"; const CROSS_CONV_TIMESTAMP = "2026-04-13T22:00:00Z"; -const PROFILE_LOADED_ON_CONNECT = "profile-loaded-on-connect"; +const PROFILE_ACQUIRED_ON_SETUP = "profile-acquired-on-setup"; const INBOUND_KIND_CHAT = "chat"; const OUTBOUND_KIND_CHAT = "chat"; const MENTIONS_NEVER = "never"; @@ -144,7 +170,10 @@ const MESSAGES_CLOSE_PATTERN = /<\/messages>/g; const NO_SENT_MESSAGE = "nope"; const FIRST_HARNESS_ROUTE = "first-harness-route"; const SECOND_HARNESS_ROUTE = "second-harness-route"; +const DEFAULT_HARNESS_ROUTE = "default-harness-route"; const HARNESS_REPLY_FAILURE_PATTERN = /HarnessReplyTestError/; +const ACQUISITION_FAILURE_PATTERN = /HarnessAcquisitionTestError/; +const DM_META: EnrichedConversationMeta = { type: "dm", participants: [] }; class MetadataCallbackTestError extends Data.TaggedError( "MetadataCallbackTestError", @@ -154,7 +183,14 @@ class HarnessReplyTestError extends Data.TaggedError("HarnessReplyTestError")< Record > {} -function createRecordedSetup(): RecordedChannelSetup { +class HarnessAcquisitionTestError extends Data.TaggedError( + "HarnessAcquisitionTestError", +)> {} + +function createRecordedSetup( + signal: Queue.Queue, + waitForInbound?: (jid: string) => Effect.Effect, +): RecordedChannelSetup { const received: ReceivedMessage[] = []; const metadata: MetadataRecord[] = []; const callOrder: string[] = []; @@ -162,6 +198,9 @@ function createRecordedSetup(): RecordedChannelSetup { onInbound: (jid, threadId, msg) => { received.push({ jid, threadId, msg }); callOrder.push(ON_INBOUND); + Queue.unsafeOffer(signal, jid); + const wait = waitForInbound?.(jid); + return wait === undefined ? undefined : Effect.runPromise(wait); }, onMetadata: (jid, name, isGroup) => { metadata.push({ jid, name, isGroup }); @@ -173,27 +212,10 @@ function createRecordedSetup(): RecordedChannelSetup { }; } -function createSignallingSetup( - signal: Queue.Queue, - waitForInbound?: (jid: string) => Effect.Effect, -): RecordedChannelSetup { - const setup = createRecordedSetup(); - return { - ...setup, - onInbound: (jid, threadId, msg) => { - setup.received.push({ jid, threadId, msg }); - setup.callOrder.push(ON_INBOUND); - Queue.unsafeOffer(signal, jid); - const wait = waitForInbound?.(jid); - return wait === undefined ? undefined : Effect.runPromise(wait); - }, - }; -} - function createMetadataFailingSetup( signal: Queue.Queue, ): RecordedChannelSetup { - const setup = createSignallingSetup(signal); + const setup = createRecordedSetup(signal); let failNext = true; return { ...setup, @@ -208,51 +230,94 @@ function createMetadataFailingSetup( }; } -function createHarnessClientFixture(): HarnessClientFixture { - const turns = Effect.runSync(Queue.unbounded()); - const replies: HarnessClientReply[] = []; +function turnSender(options: TurnOptions): HarnessTurn["sender"] { return { - client: { - agentId: testAgentId(AGENT_SELF), - startConversation: () => - Effect.dieMessage("startConversation is not used by these tests"), - turns: Stream.fromQueue(turns), - }, - replies, - turns, + id: testAgentId(options.senderId ?? AGENT_ALICE), + name: options.senderName ?? ALICE_NAME, + }; +} + +// The daemon projects context blocks before a turn reaches the adapter, so a +// fixture turn carries them the way `projectHarnessTurn` would. +function turnContextBlocks(options: TurnOptions): HarnessTurn["contextBlocks"] { + return { + ...(options.conversationMeta?.type === "group" + ? { groupMetadata: options.conversationMeta } + : {}), + ...(options.crossConversationMessages === undefined + ? {} + : { crossConversationMessages: [...options.crossConversationMessages] }), }; } function makeHarnessTurn( - fixture: HarnessClientFixture, - options: { - readonly conversationId: string; - readonly messageId: string; - readonly route: string; - readonly text?: string; - }, + replies: HarnessClientReply[], + options: TurnOptions, ): HarnessTurn { + const route = options.route ?? DEFAULT_HARNESS_ROUTE; return { - id: testMessageId(options.messageId), + id: testMessageId(options.messageId ?? MSG_ABC), conversationId: testConversationId(options.conversationId), - sender: { id: testAgentId(AGENT_ALICE), name: ALICE_NAME }, + sender: turnSender(options), text: options.text ?? HI_NANOCLAW, - isFromMe: false, + isFromMe: options.isFromMe ?? false, createdAt: MESSAGE_CREATED_AT, - conversationMeta: { type: "dm", participants: [] }, - contextBlocks: {}, + conversationMeta: options.conversationMeta ?? DM_META, + contextBlocks: turnContextBlocks(options), reply: (payload) => Effect.sync(() => { - fixture.replies.push({ route: options.route, payload }); + replies.push({ route, payload }); }), }; } -function createHarness(evalMode = false): Harness { - const fake = createFakeChannelService({ ownAgentId: AGENT_SELF }); - const config = createRecordedSetup(); - const adapter = MoltZapAdapter.fromService(fake.service, evalMode); - return { fake, config, adapter }; +/** + * Builds an adapter over a counted client acquisition. The adapter owns that + * acquisition's scope, so the counts observe exactly what `setup` and + * `teardown` did to the client's lifetime. + * @param options Eval mode plus optional pre-built replies, turns, and setup. + * @returns The adapter with the fixtures its behavior is asserted against. + */ +function createHarness(options: HarnessOptions = {}): Harness { + const turns = Effect.runSync(Queue.unbounded()); + const signal = Effect.runSync(Queue.unbounded()); + const replies = options.replies ?? []; + const counts: AcquisitionCounts = { acquired: 0, released: 0 }; + const client: HarnessClientService = { + agentId: testAgentId(AGENT_SELF), + startConversation: () => + Effect.dieMessage("startConversation is not used by these tests"), + turns: options.turns ?? Stream.fromQueue(turns), + }; + const adapter = MoltZapAdapter.fromHarnessAcquisition( + (options.acquire ?? countedAcquisition)(client, counts), + options.evalMode ?? false, + ); + return { + adapter, + config: options.config ?? createRecordedSetup(signal), + counts, + replies, + turns, + signal, + }; +} + +function countedAcquisition( + client: HarnessClientService, + counts: AcquisitionCounts, +): HarnessClientAcquisition { + // eslint-disable-next-line agent-code-guard/acquire-release-requires-scope -- The adapter under test owns the enclosing scope; that is the contract these counts assert. + return Effect.acquireRelease( + Effect.sync(() => { + counts.acquired += 1; + return client; + }), + () => + Effect.sync(() => { + counts.released += 1; + }), + ); } function asJid(conversationId: string): string { @@ -287,10 +352,6 @@ function runPromise( }); } -function flushDispatch(): Effect.Effect { - return runPromise(() => flushDispatchChain()); -} - function setup(harness: Harness): Effect.Effect { return runPromise(() => harness.adapter.setup(harness.config)); } @@ -307,6 +368,22 @@ function deliver( return runPromise(() => adapter.deliver(jid, null, makeOutbound(text))); } +/** + * Offers one turn and resolves once the adapter has dispatched it inbound. + * @param harness Adapter and fixtures under test. + * @param options Shape of the turn the client emits. + * @returns The jid the adapter dispatched that turn under. + */ +function offerTurn( + harness: Harness, + options: TurnOptions, +): Effect.Effect { + return Queue.offer( + harness.turns, + makeHarnessTurn(harness.replies, options), + ).pipe(Effect.zipRight(Queue.take(harness.signal))); +} + function expectPromiseFailure( effect: Effect.Effect, pattern: RegExp, @@ -322,67 +399,135 @@ function expectPromiseFailure( }); } -function setDmConversation(harness: Harness, conversationId: string): void { - harness.fake.state.setConversation(conversationId, { - type: "dm", - participants: [], - }); - harness.fake.state.setAgentName(AGENT_ALICE, ALICE_NAME); +function withTeardown( + harness: Harness, + effect: Effect.Effect, +): Effect.Effect { + return effect.pipe(Effect.ensuring(teardown(harness).pipe(Effect.ignore))); } -function setGroupConversation(harness: Harness): void { - harness.fake.state.setConversation(CONV_1, { +function groupMeta(name: string, members: readonly string[]) { + return { type: "group", - name: DEVS_GROUP_NAME, - participants: [`agent:${AGENT_ALICE}`], - }); - harness.fake.state.setAgentName(AGENT_ALICE, ALICE_NAME); + name, + participants: members.map((member) => `agent:${testAgentId(member)}`), + } as const satisfies EnrichedConversationMeta; } -function emitText( - harness: Harness, - conversationId: string, - text: string, -): void { - harness.fake.emit.message( - buildMessage({ - conversationId, - parts: [{ type: "text", text }], - }), - ); +function crossConvMessage(overrides: { + readonly senderName: string; + readonly senderId: string; + readonly text: string; +}): CrossConvMessage { + return { + conversationId: testConversationId(CONV_OTHER), + senderName: overrides.senderName, + senderId: testAgentId(overrides.senderId), + text: overrides.text, + timestamp: CROSS_CONV_TIMESTAMP, + }; } -function constructsSynchronouslyWithoutReadingTheProfile() { - const adapter = MoltZapAdapter.fromProfile(PROFILE_LOADED_ON_CONNECT, false); +function productionAdapter(): MoltZapAdapter { + const adapter = makeMoltZapAdapter({ + profileName: PROFILE_ACQUIRED_ON_SETUP, + evalMode: false, + }); + expect(adapter).not.toBeNull(); + return /* Safe because the profile name above is non-null, so the factory returns an adapter. */ adapter!; +} + +function constructsWithoutAcquiringItsClient() { + const adapter = productionAdapter(); expect(adapter).toBeInstanceOf(MoltZapAdapter); expect(adapter.isConnected()).toBe(false); } -function teardownBeforeSetupResolvesWithoutACore() { - const adapter = MoltZapAdapter.fromProfile(PROFILE_LOADED_ON_CONNECT, false); - return expect(adapter.teardown()).resolves.toBeUndefined(); +function teardownBeforeSetupResolvesWithoutAClient() { + return expect(productionAdapter().teardown()).resolves.toBeUndefined(); } -function setupDelegatesToCore() { +function setupAcquiresTheClientAndConnects() { const harness = createHarness(); - return Effect.gen(function* () { - expect(harness.adapter.isConnected()).toBe(false); - yield* setup(harness); - expect(harness.fake.state.connectCalls.count).toBe(1); - expect(harness.adapter.isConnected()).toBe(true); - }); + return withTeardown( + harness, + Effect.gen(function* () { + expect(harness.adapter.isConnected()).toBe(false); + yield* setup(harness); + expect(harness.counts.acquired).toBe(1); + expect(harness.adapter.isConnected()).toBe(true); + }), + ); } -function teardownDelegatesToCore() { +function setupWhileConnectedDoesNotReacquire() { + const harness = createHarness(); + return withTeardown( + harness, + Effect.gen(function* () { + yield* setup(harness); + yield* setup(harness); + expect(harness.counts.acquired).toBe(1); + expect(harness.adapter.isConnected()).toBe(true); + }), + ); +} + +function teardownClosesTheAdapterOwnedScope() { const harness = createHarness(); return Effect.gen(function* () { yield* setup(harness); + expect(harness.counts.released).toBe(0); yield* teardown(harness); - expect(harness.fake.state.closeCalls.count).toBe(1); + expect(harness.counts.released).toBe(1); expect(harness.adapter.isConnected()).toBe(false); }); } +function setupAfterTeardownAcquiresAgain() { + const harness = createHarness(); + return withTeardown( + harness, + Effect.gen(function* () { + yield* setup(harness); + yield* teardown(harness); + yield* setup(harness); + expect(harness.counts.acquired).toBe(2); + expect(harness.counts.released).toBe(1); + + expect(yield* offerTurn(harness, { conversationId: CONV_42 })).toBe( + asJid(CONV_42), + ); + }), + ); +} + +function failedAcquisitionLeavesNoScopeBehind() { + let attempts = 0; + // The first attempt fails inside the adapter-owned scope; the second + // succeeds, so a rejected setup must leave nothing half-open behind it. + const harness = createHarness({ + acquire: (client, counts) => + Effect.suspend(() => { + attempts += 1; + return attempts === 1 + ? Effect.fail(new HarnessAcquisitionTestError()) + : countedAcquisition(client, counts); + }), + }); + return withTeardown( + harness, + Effect.gen(function* () { + yield* expectPromiseFailure(setup(harness), ACQUISITION_FAILURE_PATTERN); + expect(harness.adapter.isConnected()).toBe(false); + + yield* setup(harness); + expect(harness.counts.acquired).toBe(1); + expect(harness.adapter.isConnected()).toBe(true); + }), + ); +} + function registersAdapterWithNeverMentions() { const registration = getRegisteredChannelAdapter(MOLTZAP_CHANNEL_NAME); expect(registration).toBeDefined(); @@ -397,649 +542,535 @@ function factoryReturnsNullWithoutProfile() { } function ownsPrefixedJids() { - const harness = createHarness(); - expect(harness.adapter.ownsJid(asJid(CONV_1))).toBe(true); + expect(createHarness().adapter.ownsJid(asJid(CONV_1))).toBe(true); } function rejectsOtherChannelJids() { - const harness = createHarness(); - expect(harness.adapter.ownsJid(TELEGRAM_JID)).toBe(false); - expect(harness.adapter.ownsJid(WHATSAPP_JID)).toBe(false); - expect(harness.adapter.ownsJid(RAW_CONVERSATION_JID)).toBe(false); -} - -function stripsPrefixAndForwardsSend() { - const harness = createHarness(); - return Effect.gen(function* () { - yield* setup(harness); - setDmConversation(harness, CONV_42); - harness.fake.emit.message( - buildMessage({ id: MSG_TURN_1, conversationId: CONV_42 }), - ); - yield* flushDispatch(); - yield* deliver(harness.adapter, asJid(CONV_42), HELLO_THERE); - expect(harness.fake.state.sent).toEqual([ - { - convId: testConversationId(CONV_42), - text: HELLO_THERE, - }, - ]); - }); + const { adapter } = createHarness(); + expect(adapter.ownsJid(TELEGRAM_JID)).toBe(false); + expect(adapter.ownsJid(WHATSAPP_JID)).toBe(false); + expect(adapter.ownsJid(RAW_CONVERSATION_JID)).toBe(false); } function rejectsUnownedJid() { - const harness = createHarness(); return expectPromiseFailure( - deliver(harness.adapter, TELEGRAM_JID, NO_SENT_MESSAGE), + deliver(createHarness().adapter, TELEGRAM_JID, NO_SENT_MESSAGE), OWNERSHIP_ERROR_PATTERN, ); } function rejectsDeliverWithoutInboundConversation() { - const harness = createHarness(); return expectPromiseFailure( - deliver(harness.adapter, asJid(CONV_1), NO_SENT_MESSAGE), + deliver(createHarness().adapter, asJid(CONV_1), NO_SENT_MESSAGE), UNKNOWN_CONVERSATION_PATTERN, ); } -interface GatedChannelSetup extends ChannelSetup { - readonly startedTurns: string[]; - releaseTurn(): void; -} - -/** - * Holds every inbound turn open until released, so a reply can be observed - * while its own turn is still the one in flight. - * @returns A setup whose turns block until `releaseTurn` is called. - */ -function createGatedSetup(): GatedChannelSetup { - const startedTurns: string[] = []; - const pending: Array<() => void> = []; - return { - onInbound: (jid) => { - startedTurns.push(jid); - // The host contract is promise-based, so a held turn is a pending promise. - return new Promise((resolve) => { - pending.push(() => { - resolve(undefined); - }); - }); - }, - onMetadata: () => {}, - startedTurns, - releaseTurn: () => { - pending.shift()?.(); - }, - }; -} - -function overlappingTurnsStaySerialized() { - const harness = createHarness(); - const gate = createGatedSetup(); - return Effect.gen(function* () { - yield* runPromise(() => harness.adapter.setup(gate)); - setDmConversation(harness, CONV_42); - setDmConversation(harness, CONV_43); - - harness.fake.emit.message( - buildMessage({ id: MSG_TURN_1, conversationId: CONV_42 }), - ); - yield* flushDispatch(); - expect(gate.startedTurns).toEqual([asJid(CONV_42)]); - - // A second inbound arrives while the first turn is still running. The - // core must not start it: doing so would overwrite the per-jid - // conversation entry and the still-pending first reply would address the - // wrong conversation. - harness.fake.emit.message( - buildMessage({ id: MSG_TURN_2, conversationId: CONV_43 }), - ); - yield* flushDispatch(); - expect(gate.startedTurns).toEqual([asJid(CONV_42)]); - - // The first turn replies late, and must still address its own conversation. - yield* deliver(harness.adapter, asJid(CONV_42), FIRST_REPLY); - expect(harness.fake.state.sent).toEqual([ - { convId: testConversationId(CONV_42), text: FIRST_REPLY }, - ]); - - gate.releaseTurn(); - yield* flushDispatch(); - expect(gate.startedTurns).toEqual([asJid(CONV_42), asJid(CONV_43)]); - - yield* deliver(harness.adapter, asJid(CONV_43), SECOND_REPLY); - expect(harness.fake.state.sent).toEqual([ - { convId: testConversationId(CONV_42), text: FIRST_REPLY }, - { convId: testConversationId(CONV_43), text: SECOND_REPLY }, - ]); - }); -} - function harnessRepliesUseLatestBoundTurn() { - const fixture = createHarnessClientFixture(); - const adapter = MoltZapAdapter.fromHarnessClient(fixture.client); - return Effect.gen(function* () { - const signal = yield* Queue.unbounded(); - yield* runPromise(() => adapter.setup(createSignallingSetup(signal))); - - yield* Queue.offer( - fixture.turns, - makeHarnessTurn(fixture, { - conversationId: CONV_42, - messageId: MSG_TURN_1, - route: FIRST_HARNESS_ROUTE, - }), - ); - expect(yield* Queue.take(signal)).toBe(asJid(CONV_42)); - yield* deliver(adapter, asJid(CONV_42), FIRST_REPLY); - - yield* Queue.offer( - fixture.turns, - makeHarnessTurn(fixture, { - conversationId: CONV_42, - messageId: MSG_TURN_2, - route: SECOND_HARNESS_ROUTE, - }), - ); - expect(yield* Queue.take(signal)).toBe(asJid(CONV_42)); - yield* deliver(adapter, asJid(CONV_42), SECOND_REPLY); - yield* deliver(adapter, asJid(CONV_42), SECOND_REPLY); - - expect(fixture.replies).toEqual([ - { route: FIRST_HARNESS_ROUTE, payload: FIRST_REPLY }, - { route: SECOND_HARNESS_ROUTE, payload: SECOND_REPLY }, - { route: SECOND_HARNESS_ROUTE, payload: SECOND_REPLY }, - ]); - }).pipe( - Effect.ensuring(runPromise(() => adapter.teardown()).pipe(Effect.ignore)), + const harness = createHarness(); + return withTeardown( + harness, + Effect.gen(function* () { + yield* setup(harness); + + expect( + yield* offerTurn(harness, { + conversationId: CONV_42, + messageId: MSG_TURN_1, + route: FIRST_HARNESS_ROUTE, + }), + ).toBe(asJid(CONV_42)); + yield* deliver(harness.adapter, asJid(CONV_42), FIRST_REPLY); + + expect( + yield* offerTurn(harness, { + conversationId: CONV_42, + messageId: MSG_TURN_2, + route: SECOND_HARNESS_ROUTE, + }), + ).toBe(asJid(CONV_42)); + yield* deliver(harness.adapter, asJid(CONV_42), SECOND_REPLY); + yield* deliver(harness.adapter, asJid(CONV_42), SECOND_REPLY); + + expect(harness.replies).toEqual([ + { route: FIRST_HARNESS_ROUTE, payload: FIRST_REPLY }, + { route: SECOND_HARNESS_ROUTE, payload: SECOND_REPLY }, + { route: SECOND_HARNESS_ROUTE, payload: SECOND_REPLY }, + ]); + }), ); } function harnessReplyFailureHasNoFallback() { - const fixture = createHarnessClientFixture(); + const replies: HarnessClientReply[] = []; const reply = vi .fn() .mockReturnValue(Effect.fail(new HarnessReplyTestError())); const turn = { - ...makeHarnessTurn(fixture, { + ...makeHarnessTurn(replies, { conversationId: CONV_42, messageId: MSG_TURN_1, route: FIRST_HARNESS_ROUTE, }), reply, }; - const adapter = MoltZapAdapter.fromHarnessClient({ - ...fixture.client, - turns: Stream.make(turn), - }); - return Effect.gen(function* () { - const signal = yield* Queue.unbounded(); - yield* runPromise(() => adapter.setup(createSignallingSetup(signal))); - expect(yield* Queue.take(signal)).toBe(asJid(CONV_42)); - - yield* expectPromiseFailure( - deliver(adapter, asJid(CONV_42), FIRST_REPLY), - HARNESS_REPLY_FAILURE_PATTERN, - ); - expect(reply).toHaveBeenCalledExactlyOnceWith(FIRST_REPLY); - expect(fixture.replies).toEqual([]); - }).pipe( - Effect.ensuring(runPromise(() => adapter.teardown()).pipe(Effect.ignore)), + const harness = createHarness({ replies, turns: Stream.make(turn) }); + return withTeardown( + harness, + Effect.gen(function* () { + yield* setup(harness); + expect(yield* Queue.take(harness.signal)).toBe(asJid(CONV_42)); + + yield* expectPromiseFailure( + deliver(harness.adapter, asJid(CONV_42), FIRST_REPLY), + HARNESS_REPLY_FAILURE_PATTERN, + ); + expect(reply).toHaveBeenCalledExactlyOnceWith(FIRST_REPLY); + expect(replies).toEqual([]); + }), ); } function harnessTurnsDrainSequentially() { - const fixture = createHarnessClientFixture(); - const adapter = MoltZapAdapter.fromHarnessClient(fixture.client); - return Effect.gen(function* () { - const signal = yield* Queue.unbounded(); - const releaseFirst = yield* Deferred.make(); - let firstInbound = true; - const setup = createSignallingSetup(signal, () => { - if (!firstInbound) { - return Effect.succeed(undefined); - } - firstInbound = false; - return Deferred.await(releaseFirst); - }); - yield* runPromise(() => adapter.setup(setup)); - - yield* Queue.offer( - fixture.turns, - makeHarnessTurn(fixture, { - conversationId: CONV_42, - messageId: MSG_TURN_1, - route: FIRST_HARNESS_ROUTE, - }), - ); - expect(yield* Queue.take(signal)).toBe(asJid(CONV_42)); - - yield* Queue.offer( - fixture.turns, - makeHarnessTurn(fixture, { - conversationId: CONV_43, - messageId: MSG_TURN_2, - route: SECOND_HARNESS_ROUTE, - }), - ); - yield* Effect.yieldNow(); - expect(yield* Queue.size(signal)).toBe(0); - expect(yield* Queue.size(fixture.turns)).toBe(1); - - yield* Deferred.succeed(releaseFirst, undefined); - expect(yield* Queue.take(signal)).toBe(asJid(CONV_43)); - }).pipe( - Effect.ensuring(runPromise(() => adapter.teardown()).pipe(Effect.ignore)), - ); -} - -function harnessTeardownLeavesClientCallerOwned() { - const fixture = createHarnessClientFixture(); - const adapter = MoltZapAdapter.fromHarnessClient(fixture.client); - return Effect.gen(function* () { - const signal = yield* Queue.unbounded(); - const setupConfig = createSignallingSetup(signal); - yield* runPromise(() => adapter.setup(setupConfig)); - expect(adapter.isConnected()).toBe(true); - - yield* runPromise(() => adapter.teardown()); - expect(adapter.isConnected()).toBe(false); - expect(yield* Queue.isShutdown(fixture.turns)).toBe(false); - - yield* Queue.offer( - fixture.turns, - makeHarnessTurn(fixture, { - conversationId: CONV_42, - messageId: MSG_TURN_1, - route: FIRST_HARNESS_ROUTE, - }), - ); - yield* Effect.yieldNow(); - expect(yield* Queue.size(fixture.turns)).toBe(1); - expect(yield* Queue.size(signal)).toBe(0); - - yield* runPromise(() => adapter.setup(setupConfig)); - expect(yield* Queue.take(signal)).toBe(asJid(CONV_42)); - expect(adapter.isConnected()).toBe(true); - }).pipe( - Effect.ensuring(runPromise(() => adapter.teardown()).pipe(Effect.ignore)), + const signal = Effect.runSync(Queue.unbounded()); + const releaseFirst = Effect.runSync(Deferred.make()); + let firstInbound = true; + const config = createRecordedSetup(signal, () => { + if (!firstInbound) { + return Effect.succeed(undefined); + } + firstInbound = false; + return Deferred.await(releaseFirst); + }); + const harness = { ...createHarness({ config }), signal }; + return withTeardown( + harness, + Effect.gen(function* () { + yield* setup(harness); + + yield* Queue.offer( + harness.turns, + makeHarnessTurn(harness.replies, { + conversationId: CONV_42, + messageId: MSG_TURN_1, + route: FIRST_HARNESS_ROUTE, + }), + ); + expect(yield* Queue.take(signal)).toBe(asJid(CONV_42)); + + yield* Queue.offer( + harness.turns, + makeHarnessTurn(harness.replies, { + conversationId: CONV_43, + messageId: MSG_TURN_2, + route: SECOND_HARNESS_ROUTE, + }), + ); + yield* Effect.yieldNow(); + expect(yield* Queue.size(signal)).toBe(0); + expect(yield* Queue.size(harness.turns)).toBe(1); + + yield* Deferred.succeed(releaseFirst, undefined); + expect(yield* Queue.take(signal)).toBe(asJid(CONV_43)); + }), ); } function harnessMetadataFailureDoesNotStopDrain() { - const fixture = createHarnessClientFixture(); - const adapter = MoltZapAdapter.fromHarnessClient(fixture.client); - return Effect.gen(function* () { - const signal = yield* Queue.unbounded(); - const setupConfig = createMetadataFailingSetup(signal); - yield* runPromise(() => adapter.setup(setupConfig)); - - yield* Queue.offer( - fixture.turns, - makeHarnessTurn(fixture, { - conversationId: CONV_42, - messageId: MSG_TURN_1, - route: FIRST_HARNESS_ROUTE, - }), - ); - yield* Queue.offer( - fixture.turns, - makeHarnessTurn(fixture, { - conversationId: CONV_43, - messageId: MSG_TURN_2, - route: SECOND_HARNESS_ROUTE, - }), - ); - - expect(yield* Queue.take(signal)).toBe(asJid(CONV_43)); - expect(setupConfig.received).toHaveLength(1); - expect(adapter.isConnected()).toBe(true); - }).pipe( - Effect.ensuring(runPromise(() => adapter.teardown()).pipe(Effect.ignore)), + const signal = Effect.runSync(Queue.unbounded()); + const config = createMetadataFailingSetup(signal); + const harness = { ...createHarness({ config }), signal }; + return withTeardown( + harness, + Effect.gen(function* () { + yield* setup(harness); + + yield* Queue.offer( + harness.turns, + makeHarnessTurn(harness.replies, { + conversationId: CONV_42, + messageId: MSG_TURN_1, + route: FIRST_HARNESS_ROUTE, + }), + ); + yield* Queue.offer( + harness.turns, + makeHarnessTurn(harness.replies, { + conversationId: CONV_43, + messageId: MSG_TURN_2, + route: SECOND_HARNESS_ROUTE, + }), + ); + + expect(yield* Queue.take(signal)).toBe(asJid(CONV_43)); + expect(config.received).toHaveLength(1); + expect(harness.adapter.isConnected()).toBe(true); + }), ); } function harnessLateDeliveryUsesRetainedAuthority() { - const fixture = createHarnessClientFixture(); - const turn = makeHarnessTurn(fixture, { + const replies: HarnessClientReply[] = []; + const turn = makeHarnessTurn(replies, { conversationId: CONV_42, messageId: MSG_TURN_1, route: FIRST_HARNESS_ROUTE, }); - const adapter = MoltZapAdapter.fromHarnessClient({ - ...fixture.client, - turns: Stream.make(turn), - }); - return Effect.gen(function* () { - const signal = yield* Queue.unbounded(); - yield* runPromise(() => adapter.setup(createSignallingSetup(signal))); - expect(yield* Queue.take(signal)).toBe(asJid(CONV_42)); - yield* runPromise(() => - vi.waitFor(() => { - expect(adapter.isConnected()).toBe(false); - }), - ); - - yield* deliver(adapter, asJid(CONV_42), FIRST_REPLY); - expect(fixture.replies).toEqual([ - { route: FIRST_HARNESS_ROUTE, payload: FIRST_REPLY }, - ]); - }).pipe( - Effect.ensuring(runPromise(() => adapter.teardown()).pipe(Effect.ignore)), + const harness = createHarness({ replies, turns: Stream.make(turn) }); + return withTeardown( + harness, + Effect.gen(function* () { + yield* setup(harness); + expect(yield* Queue.take(harness.signal)).toBe(asJid(CONV_42)); + yield* runPromise(() => + vi.waitFor(() => { + expect(harness.adapter.isConnected()).toBe(false); + }), + ); + + yield* deliver(harness.adapter, asJid(CONV_42), FIRST_REPLY); + expect(replies).toEqual([ + { route: FIRST_HARNESS_ROUTE, payload: FIRST_REPLY }, + ]); + }), ); } -function mapsEnrichedMessageToInboundMessage() { +function mapsTurnToInboundMessage() { const harness = createHarness(); - return Effect.gen(function* () { - yield* setup(harness); - harness.fake.state.setConversation(CONV_1, { - type: "dm", - name: "alice-dm", - participants: [], - }); - harness.fake.state.setAgentName(AGENT_ALICE, ALICE_NAME); - harness.fake.emit.message( - buildMessage({ - id: MSG_ABC, + return withTeardown( + harness, + Effect.gen(function* () { + yield* setup(harness); + yield* offerTurn(harness, { conversationId: CONV_1, - senderId: AGENT_ALICE, - parts: [{ type: "text", text: HI_NANOCLAW }], - createdAt: MESSAGE_CREATED_AT, - }), - ); - yield* flushDispatch(); - - expect(harness.config.received).toHaveLength(1); - const { jid, threadId, msg } = - /* Safe because the test fixture establishes this asserted shape. */ harness - .config.received[0]!; - expect(jid).toBe(asJid(CONV_1)); - expect(threadId).toBeNull(); - expect(msg.id).toBe(testMessageId(MSG_ABC)); - expect(msg.kind).toBe(INBOUND_KIND_CHAT); - expect(msg.timestamp).toBe(MESSAGE_CREATED_AT); - expect(msg.isGroup).toBe(false); - const content = inboundContent(msg); - expect(content.text).toBe(HI_NANOCLAW); - expect(content.sender).toBe(ALICE_NAME); - expect(content.senderId).toBe(senderIdFor(AGENT_ALICE)); - }); + messageId: MSG_ABC, + conversationMeta: { type: "dm", name: "alice-dm", participants: [] }, + }); + + expect(harness.config.received).toHaveLength(1); + const received = + /* Safe because the assertion above established the entry exists. */ harness + .config.received[0]!; + expect(received).toMatchObject({ jid: asJid(CONV_1), threadId: null }); + expect(received.msg).toMatchObject({ + id: testMessageId(MSG_ABC), + kind: INBOUND_KIND_CHAT, + timestamp: MESSAGE_CREATED_AT, + isGroup: false, + }); + expect(inboundContent(received.msg)).toEqual({ + text: HI_NANOCLAW, + sender: ALICE_NAME, + senderId: senderIdFor(AGENT_ALICE), + }); + }), + ); } function emitsMetadataBeforeMessage() { const harness = createHarness(); - return Effect.gen(function* () { - yield* setup(harness); - setGroupConversation(harness); - harness.fake.emit.message(buildMessage()); - yield* flushDispatch(); - - expect(harness.config.callOrder).toEqual([ON_METADATA, ON_INBOUND]); - expect(harness.config.metadata).toHaveLength(1); - expect(harness.config.metadata[0]).toMatchObject({ - jid: asJid(CONV_1), - name: DEVS_GROUP_NAME, - isGroup: true, - }); - }); + return withTeardown( + harness, + Effect.gen(function* () { + yield* setup(harness); + yield* offerTurn(harness, { + conversationId: CONV_1, + conversationMeta: groupMeta(DEVS_GROUP_NAME, [AGENT_ALICE]), + }); + + expect(harness.config.callOrder).toEqual([ON_METADATA, ON_INBOUND]); + expect(harness.config.metadata).toHaveLength(1); + expect(harness.config.metadata[0]).toMatchObject({ + jid: asJid(CONV_1), + name: DEVS_GROUP_NAME, + isGroup: true, + }); + }), + ); } function dropsMessagesFromOwnAgent() { const harness = createHarness(); - return Effect.gen(function* () { - yield* setup(harness); - harness.fake.emit.message( - buildMessage({ conversationId: CONV_1, senderId: AGENT_SELF }), - ); - yield* flushDispatch(); - expect(harness.config.received).toHaveLength(0); - expect(harness.config.callOrder).toHaveLength(0); - }); + return withTeardown( + harness, + Effect.gen(function* () { + yield* setup(harness); + yield* Queue.offer( + harness.turns, + makeHarnessTurn(harness.replies, { + conversationId: CONV_1, + senderId: AGENT_SELF, + isFromMe: true, + }), + ); + // The dropped turn signals nothing, so a following turn that does + // dispatch is what proves the drain consumed and discarded the first. + expect(yield* offerTurn(harness, { conversationId: CONV_42 })).toBe( + asJid(CONV_42), + ); + + expect(harness.config.received).toHaveLength(1); + expect(harness.config.received[0]?.jid).toBe(asJid(CONV_42)); + }), + ); } function doesNotCreateWiringWithoutEvalMode() { - const harness = createHarness(false); - return Effect.gen(function* () { - yield* setup(harness); - setDmConversation(harness, CONV_EVAL_OFF); - harness.fake.emit.message(buildMessage({ conversationId: CONV_EVAL_OFF })); - yield* flushDispatch(); - expect( - getMessagingGroupByPlatform(MOLTZAP_CHANNEL_NAME, asJid(CONV_EVAL_OFF)), - ).toBeUndefined(); - }); + const harness = createHarness({ evalMode: false }); + return withTeardown( + harness, + Effect.gen(function* () { + yield* setup(harness); + yield* offerTurn(harness, { conversationId: CONV_EVAL_OFF }); + expect( + getMessagingGroupByPlatform(MOLTZAP_CHANNEL_NAME, asJid(CONV_EVAL_OFF)), + ).toBeUndefined(); + }), + ); } function autoRegistersEvalWiring() { - const harness = createHarness(true); - return Effect.gen(function* () { - yield* setup(harness); - setDmConversation(harness, CONV_EVAL_ON); - harness.fake.emit.message(buildMessage({ conversationId: CONV_EVAL_ON })); - yield* flushDispatch(); - - const jid = asJid(CONV_EVAL_ON); - const group = getMessagingGroupByPlatform(MOLTZAP_CHANNEL_NAME, jid); - expect(group).toBeDefined(); - expect( - /* Safe because the test fixture establishes this asserted shape. */ group! - .platform_id, - ).toBe(jid); - expect( - /* Safe because the test fixture establishes this asserted shape. */ group! - .unknown_sender_policy, - ).toBe(UNKNOWN_SENDER_PUBLIC); - - const wiring = getMessagingGroupAgentByPair( - /* Safe because the test fixture establishes this asserted shape. */ group! - .id, - EVAL_AGENT_GROUP_ID, - ); - expect(wiring).toBeDefined(); - expect( - /* Safe because the test fixture establishes this asserted shape. */ wiring! - .engage_mode, - ).toBe(ENGAGE_MODE_PATTERN); - expect( - /* Safe because the test fixture establishes this asserted shape. */ wiring! - .engage_pattern, - ).toBe(ENGAGE_PATTERN_DOT); - expect( - /* Safe because the test fixture establishes this asserted shape. */ wiring! - .sender_scope, - ).toBe(SENDER_SCOPE_ALL); - expect( - /* Safe because the test fixture establishes this asserted shape. */ wiring! - .ignored_message_policy, - ).toBe(IGNORED_MESSAGE_POLICY_DROP); - expect( - /* Safe because the test fixture establishes this asserted shape. */ wiring! - .session_mode, - ).toBe(SESSION_MODE_SHARED); - expect( - /* Safe because the test fixture establishes this asserted shape. */ wiring! - .priority, - ).toBe(DEFAULT_WIRING_PRIORITY); - }); + const harness = createHarness({ evalMode: true }); + return withTeardown( + harness, + Effect.gen(function* () { + yield* setup(harness); + yield* offerTurn(harness, { conversationId: CONV_EVAL_ON }); + + const jid = asJid(CONV_EVAL_ON); + const group = getMessagingGroupByPlatform(MOLTZAP_CHANNEL_NAME, jid); + expect(group).toMatchObject({ + platform_id: jid, + unknown_sender_policy: UNKNOWN_SENDER_PUBLIC, + }); + + const wiring = getMessagingGroupAgentByPair( + /* Safe because the assertion above established the group exists. */ group! + .id, + EVAL_AGENT_GROUP_ID, + ); + // Every persisted policy field comes from the channel's declared + // defaults, so the wiring row cannot drift from the contract. + expect(wiring).toMatchObject({ + engage_mode: ENGAGE_MODE_PATTERN, + engage_pattern: ENGAGE_PATTERN_DOT, + sender_scope: SENDER_SCOPE_ALL, + ignored_message_policy: IGNORED_MESSAGE_POLICY_DROP, + session_mode: SESSION_MODE_SHARED, + priority: DEFAULT_WIRING_PRIORITY, + }); + }), + ); } function doesNotRecreateExistingEvalWiring() { - const harness = createHarness(true); + const harness = createHarness({ evalMode: true }); const jid = asJid(CONV_EVAL_IDEMPOTENT); - return Effect.gen(function* () { - yield* setup(harness); - setDmConversation(harness, CONV_EVAL_IDEMPOTENT); - harness.fake.emit.message( - buildMessage({ id: MSG_EVAL_1, conversationId: CONV_EVAL_IDEMPOTENT }), - ); - yield* flushDispatch(); - const firstGroup = getMessagingGroupByPlatform(MOLTZAP_CHANNEL_NAME, jid); - expect(firstGroup).toBeDefined(); - - harness.fake.emit.message( - buildMessage({ id: MSG_EVAL_2, conversationId: CONV_EVAL_IDEMPOTENT }), - ); - yield* flushDispatch(); - const secondGroup = getMessagingGroupByPlatform(MOLTZAP_CHANNEL_NAME, jid); - // Same stored object — the second inbound short-circuits before recreating. - expect(secondGroup).toBe(firstGroup); - }); + return withTeardown( + harness, + Effect.gen(function* () { + yield* setup(harness); + yield* offerTurn(harness, { + conversationId: CONV_EVAL_IDEMPOTENT, + messageId: MSG_EVAL_1, + }); + const firstGroup = getMessagingGroupByPlatform(MOLTZAP_CHANNEL_NAME, jid); + expect(firstGroup).toBeDefined(); + + yield* offerTurn(harness, { + conversationId: CONV_EVAL_IDEMPOTENT, + messageId: MSG_EVAL_2, + }); + const secondGroup = getMessagingGroupByPlatform( + MOLTZAP_CHANNEL_NAME, + jid, + ); + // Same stored object — the second inbound short-circuits before recreating. + expect(secondGroup).toBe(firstGroup); + }), + ); } function inlinesGroupMetadataBlock() { const harness = createHarness(); - return Effect.gen(function* () { - yield* setup(harness); - harness.fake.state.setConversation(CONV_1, { - type: "group", - name: DEVS_GROUP_NAME, - participants: [`agent:${AGENT_ALICE}`, `agent:${AGENT_BOB}`], - }); - harness.fake.state.setAgentName(AGENT_ALICE, ALICE_NAME); - emitText(harness, CONV_1, HI_TEAM); - yield* flushDispatch(); - - const content = firstReceivedContent(harness); - expect(content).toContain(SYSTEM_REMINDER_OPEN); - expect(content).toContain(GROUP_CONVERSATION_TEXT); - expect(content).toContain(GROUP_NAME_DEVS_TEXT); - expect(content).toContain( - `Participants (2): agent:${testAgentId(AGENT_ALICE)}, agent:${testAgentId(AGENT_BOB)}`, - ); - expect(content).toContain(SYSTEM_REMINDER_CLOSE); - expect(content).toMatch(GROUP_ENDS_WITH_HI_TEAM); - }); + return withTeardown( + harness, + Effect.gen(function* () { + yield* setup(harness); + yield* offerTurn(harness, { + conversationId: CONV_1, + text: HI_TEAM, + conversationMeta: groupMeta(DEVS_GROUP_NAME, [AGENT_ALICE, AGENT_BOB]), + }); + + const content = firstReceivedContent(harness); + expect(content).toContain(SYSTEM_REMINDER_OPEN); + expect(content).toContain(GROUP_CONVERSATION_TEXT); + expect(content).toContain(GROUP_NAME_DEVS_TEXT); + expect(content).toContain( + `Participants (2): agent:${testAgentId(AGENT_ALICE)}, agent:${testAgentId(AGENT_BOB)}`, + ); + expect(content).toContain(SYSTEM_REMINDER_CLOSE); + expect(content).toMatch(GROUP_ENDS_WITH_HI_TEAM); + }), + ); } function omitsGroupBlockForDmConversations() { const harness = createHarness(); - return Effect.gen(function* () { - yield* setup(harness); - harness.fake.state.setConversation(CONV_1, { - type: "dm", - name: "alice-dm", - participants: [`agent:${AGENT_ALICE}`, `agent:${AGENT_SELF}`], - }); - harness.fake.state.setAgentName(AGENT_ALICE, ALICE_NAME); - emitText(harness, CONV_1, JUST_A_DM); - yield* flushDispatch(); - expect(firstReceivedContent(harness)).toBe(JUST_A_DM); - }); + return withTeardown( + harness, + Effect.gen(function* () { + yield* setup(harness); + yield* offerTurn(harness, { + conversationId: CONV_1, + text: JUST_A_DM, + conversationMeta: { + type: "dm", + name: "alice-dm", + participants: [ + `agent:${testAgentId(AGENT_ALICE)}`, + `agent:${testAgentId(AGENT_SELF)}`, + ], + }, + }); + expect(firstReceivedContent(harness)).toBe(JUST_A_DM); + }), + ); } function inlinesCrossConversationMessages() { const harness = createHarness(); - return Effect.gen(function* () { - yield* setup(harness); - setDmConversation(harness, CONV_1); - harness.fake.state.setFullMessages(CONV_1, [ - { - conversationId: CONV_OTHER, - senderName: BOB_NAME, - senderId: AGENT_BOB, - text: FREEDONIA_TEXT, - timestamp: CROSS_CONV_TIMESTAMP, - }, - ]); - emitText(harness, CONV_1, QUESTION_TEXT); - yield* flushDispatch(); - - const content = firstReceivedContent(harness); - expect(content).toContain(MESSAGES_OPEN); - expect(content).toContain(SENDER_BOB_ATTRIBUTE); - expect(content).toContain(ZENDA_TEXT); - expect(content).toMatch(QUESTION_ENDS_CONTENT); - }); + return withTeardown( + harness, + Effect.gen(function* () { + yield* setup(harness); + yield* offerTurn(harness, { + conversationId: CONV_1, + text: QUESTION_TEXT, + crossConversationMessages: [ + crossConvMessage({ + senderName: BOB_NAME, + senderId: AGENT_BOB, + text: FREEDONIA_TEXT, + }), + ], + }); + + const content = firstReceivedContent(harness); + expect(content).toContain(MESSAGES_OPEN); + expect(content).toContain(SENDER_BOB_ATTRIBUTE); + expect(content).toContain(ZENDA_TEXT); + expect(content).toMatch(QUESTION_ENDS_CONTENT); + }), + ); } function ordersContextBlocksBeforeRawText() { const harness = createHarness(); - return Effect.gen(function* () { - yield* setup(harness); - setGroupConversation(harness); - harness.fake.state.setFullMessages(CONV_1, [ - { - conversationId: CONV_OTHER, - senderName: BOB_NAME, - senderId: AGENT_BOB, - text: CROSS_CONV_CANARY, - timestamp: CROSS_CONV_TIMESTAMP, - }, - ]); - emitText(harness, CONV_1, ACTUAL_MESSAGE_TEXT); - yield* flushDispatch(); - - const content = firstReceivedContent(harness); - const xconvIdx = content.indexOf(CROSS_CONV_CANARY); - const groupIdx = content.indexOf(GROUP_CONVERSATION_TEXT); - const textIdx = content.indexOf(ACTUAL_MESSAGE_TEXT); - expect(xconvIdx).toBeGreaterThanOrEqual(0); - expect(groupIdx).toBeGreaterThan(xconvIdx); - expect(textIdx).toBeGreaterThan(groupIdx); - }); + return withTeardown( + harness, + Effect.gen(function* () { + yield* setup(harness); + yield* offerTurn(harness, { + conversationId: CONV_1, + text: ACTUAL_MESSAGE_TEXT, + conversationMeta: groupMeta(DEVS_GROUP_NAME, [AGENT_ALICE]), + crossConversationMessages: [ + crossConvMessage({ + senderName: BOB_NAME, + senderId: AGENT_BOB, + text: CROSS_CONV_CANARY, + }), + ], + }); + + const content = firstReceivedContent(harness); + const xconvIdx = content.indexOf(CROSS_CONV_CANARY); + const groupIdx = content.indexOf(GROUP_CONVERSATION_TEXT); + const textIdx = content.indexOf(ACTUAL_MESSAGE_TEXT); + expect(xconvIdx).toBeGreaterThanOrEqual(0); + expect(groupIdx).toBeGreaterThan(xconvIdx); + expect(textIdx).toBeGreaterThan(groupIdx); + }), + ); } function sanitizesGroupMetadata() { const harness = createHarness(); - return Effect.gen(function* () { - yield* setup(harness); - harness.fake.state.setConversation(CONV_1, { - type: "group", - name: MALICIOUS_GROUP_NAME, - participants: [`agent:${AGENT_ALICE}`], - }); - harness.fake.state.setAgentName(AGENT_ALICE, ALICE_NAME); - harness.fake.emit.message(buildMessage()); - yield* flushDispatch(); - - const content = firstReceivedContent(harness); - expect(content).not.toContain(MALICIOUS_GROUP_FRAGMENT); - expect(content).toContain(ESCAPED_GROUP_FRAGMENT); - expect(content.match(SYSTEM_REMINDER_OPEN_PATTERN)).toHaveLength(1); - expect(content.match(SYSTEM_REMINDER_CLOSE_PATTERN)).toHaveLength(1); - }); + return withTeardown( + harness, + Effect.gen(function* () { + yield* setup(harness); + yield* offerTurn(harness, { + conversationId: CONV_1, + conversationMeta: groupMeta(MALICIOUS_GROUP_NAME, [AGENT_ALICE]), + }); + + const content = firstReceivedContent(harness); + expect(content).not.toContain(MALICIOUS_GROUP_FRAGMENT); + expect(content).toContain(ESCAPED_GROUP_FRAGMENT); + expect(content.match(SYSTEM_REMINDER_OPEN_PATTERN)).toHaveLength(1); + expect(content.match(SYSTEM_REMINDER_CLOSE_PATTERN)).toHaveLength(1); + }), + ); } function sanitizesCrossConversationSenderName() { const harness = createHarness(); - return Effect.gen(function* () { - yield* setup(harness); - setDmConversation(harness, CONV_1); - harness.fake.state.setFullMessages(CONV_1, [ - { - conversationId: CONV_OTHER, - senderName: MALICIOUS_SENDER, - senderId: AGENT_MALLORY, - text: CONTENT_TEXT, - timestamp: CROSS_CONV_TIMESTAMP, - }, - ]); - harness.fake.emit.message(buildMessage()); - yield* flushDispatch(); - - const content = firstReceivedContent(harness); - expect(content).not.toContain(MALICIOUS_MESSAGES_FRAGMENT); - expect(content).toContain(ESCAPED_MESSAGES_FRAGMENT); - expect(content.match(MESSAGES_OPEN_PATTERN)).toHaveLength(1); - expect(content.match(MESSAGES_CLOSE_PATTERN)).toHaveLength(1); - }); + return withTeardown( + harness, + Effect.gen(function* () { + yield* setup(harness); + yield* offerTurn(harness, { + conversationId: CONV_1, + crossConversationMessages: [ + crossConvMessage({ + senderName: MALICIOUS_SENDER, + senderId: AGENT_MALLORY, + text: CONTENT_TEXT, + }), + ], + }); + + const content = firstReceivedContent(harness); + expect(content).not.toContain(MALICIOUS_MESSAGES_FRAGMENT); + expect(content).toContain(ESCAPED_MESSAGES_FRAGMENT); + expect(content.match(MESSAGES_OPEN_PATTERN)).toHaveLength(1); + expect(content.match(MESSAGES_CLOSE_PATTERN)).toHaveLength(1); + }), + ); } describe("MoltZapAdapter lifecycle", () => { vitestIt( - "constructs synchronously without reading the profile", - constructsSynchronouslyWithoutReadingTheProfile, + "constructs without acquiring its client", + constructsWithoutAcquiringItsClient, ); vitestIt( - "teardown before setup resolves without a core", - teardownBeforeSetupResolvesWithoutACore, + "teardown before setup resolves without a client", + teardownBeforeSetupResolvesWithoutAClient, ); - it("setup delegates to the core and marks connected", setupDelegatesToCore); it( - "teardown delegates to the core and clears connected", - teardownDelegatesToCore, + "setup acquires the client and marks connected", + setupAcquiresTheClientAndConnects, + ); + it( + "setup while connected does not reacquire the client", + setupWhileConnectedDoesNotReacquire, + ); + it( + "teardown closes the adapter-owned client scope", + teardownClosesTheAdapterOwnedScope, + ); + it( + "setup after teardown acquires a fresh client and drains it", + setupAfterTeardownAcquiresAgain, + ); + it( + "a failed acquisition leaves no scope behind for the next setup", + failedAcquisitionLeavesNoScopeBehind, ); }); @@ -1060,10 +1091,6 @@ describe("MoltZapAdapter ownership", () => { }); describe("MoltZapAdapter deliver basics", () => { - it( - "strips the mz prefix and forwards to core.sendReply", - stripsPrefixAndForwardsSend, - ); it("rejects a JID not owned by this channel", rejectsUnownedJid); it( "rejects when no inbound established a conversation for the JID", @@ -1071,13 +1098,6 @@ describe("MoltZapAdapter deliver basics", () => { ); }); -describe("MoltZapAdapter turn serialization", () => { - it( - "serializes overlapping turns so each reply keeps its own conversation", - overlappingTurnsStaySerialized, - ); -}); - // @agent-code-guard/regression-only: controlled queues and callbacks pin the exact asynchronous NanoClaw delivery and drain lifecycle. describe("MoltZapAdapter HarnessClient behavior", () => { it( @@ -1085,17 +1105,10 @@ describe("MoltZapAdapter HarnessClient behavior", () => { harnessRepliesUseLatestBoundTurn, ); it( - "propagates reply failure without a legacy fallback", + "propagates reply failure with no other route", harnessReplyFailureHasNoFallback, ); - it( - "drains injected Harness turns sequentially", - harnessTurnsDrainSequentially, - ); - it( - "can restart its drain without closing the caller-owned client", - harnessTeardownLeavesClientCallerOwned, - ); + it("drains Harness turns sequentially", harnessTurnsDrainSequentially); it( "continues after a synchronous metadata callback failure", harnessMetadataFailureDoesNotStopDrain, @@ -1108,8 +1121,8 @@ describe("MoltZapAdapter HarnessClient behavior", () => { describe("MoltZapAdapter inbound projection", () => { it( - "maps enriched message to InboundMessage with mz prefix", - mapsEnrichedMessageToInboundMessage, + "maps a Harness turn to InboundMessage with mz prefix", + mapsTurnToInboundMessage, ); it("calls onMetadata before onInbound", emitsMetadataBeforeMessage); it( diff --git a/packages/nanoclaw-channel/src/channels/moltzap.ts b/packages/nanoclaw-channel/src/channels/moltzap.ts index 18e6bb147..91f50ee7d 100644 --- a/packages/nanoclaw-channel/src/channels/moltzap.ts +++ b/packages/nanoclaw-channel/src/channels/moltzap.ts @@ -4,25 +4,22 @@ import { ConfigProvider, Data, Effect, + Exit, Fiber, - Match, Option, + Scope, Stream, } from "effect"; -import { MoltZapService, type ServiceRpcError } from "@moltzap/client"; +import { harnessClientForProfile } from "@moltzap/client"; import type { HarnessClientService, HarnessTurn, } from "@moltzap/client/harness-client"; -import type { ConversationId } from "@moltzap/protocol/conversation"; import { BoundedMap, - MoltZapChannelCore, formatCrossConv, formatGroupBlock, getGroupFields, - type ChannelService, - type EnrichedInboundMessage, } from "@moltzap/client/channel-base"; import type { @@ -41,8 +38,8 @@ import { import type { MessagingGroupAgent } from "../types.js"; // `MoltZapChannelError` covers nanoclaw's host-shape failures: un-owned jid, -// unknown conversation, disconnected channel. Reply failures keep the error -// type supplied by their backing client. +// unknown conversation, and a host callback that rejects a projected turn. +// Reply failures keep the error type supplied by their backing client. class MoltZapChannelError extends Data.TaggedError("MoltZapChannelError")<{ readonly reason: string; }> { @@ -92,6 +89,17 @@ const moltZapChannelEnv = Config.all({ evalMode: moltZapEvalModeEnv, }); +/** + * A scoped acquisition of the adapter-facing Harness capability. The adapter + * opens the scope in `setup` and closes it in `teardown`, so the acquisition + * describes the whole client lifetime rather than a borrowed connection. + */ +export type HarnessClientAcquisition = Effect.Effect< + HarnessClientService, + Error, + Scope.Scope +>; + /** * MoltZap conversationId → nanoclaw platform id. The router addresses * conversations by `(channelType, platformId)`; this channel uses @@ -135,47 +143,32 @@ function extractOutboundText(message: OutboundMessage): string | null { return null; } -interface MoltZapAdapterState { - readonly core: MoltZapChannelCore | null; - readonly harnessClient: HarnessClientService | null; - readonly ownAgentId: string; - readonly evalMode: boolean; - readonly profileName: string | null; -} - -type ConversationReplyRoute = - | { - readonly _tag: "legacy"; - readonly conversationId: ConversationId; - } - | { - readonly _tag: "harness"; - readonly reply: HarnessTurn["reply"]; - }; - /** - * Nanoclaw channel adapter for MoltZap. Presents Nanoclaw's - * `ChannelAdapter` contract over either the transitional channel core or an - * injected `HarnessClient`. + * Nanoclaw channel adapter for MoltZap. Presents Nanoclaw's `ChannelAdapter` + * contract over a Harness client whose lifetime this adapter owns. * * ```mermaid * sequenceDiagram - * participant Source as HarnessClient or MoltZapChannelCore - * participant Handler as handleInbound (this adapter) + * participant Nano as nanoclaw channel host + * participant Adapter as MoltZapAdapter + * participant Client as HarnessClient * participant Router as nanoclaw router - * Source->>Handler: HarnessTurn or enriched inbound - * note over Handler: Step 1 — jidFromConversationId
platformId = "mz:" + conversationId - * note over Handler: Step 2 — rememberReplyRoute
latest reply authority retained by jid - * note over Handler: Step 3 — ensureEvalWiring (eval mode only)
conversation rows target the harness-seeded agent - * Handler->>Router: Step 4 — setup.onMetadata(jid, name, isGroup) - * Handler->>Router: Step 5 — setup.onInbound(jid, null, message) - * Router-->>Handler: Step 6 — callback resolves + * Nano->>Adapter: setup(config) + * note over Adapter: Step 1 — Scope.make
the adapter owns the client scope + * Adapter->>Client: Step 2 — acquire within that scope + * Client-->>Adapter: Step 3 — HarnessTurn per inbound + * note over Adapter: Step 4 — jidFromConversationId
platformId = "mz:" + conversationId + * note over Adapter: Step 5 — rememberReplyRoute
turn.reply retained by jid + * note over Adapter: Step 6 — ensureEvalWiring (eval mode only)
conversation rows target the harness-seeded agent + * Adapter->>Router: Step 7 — setup.onMetadata(jid, name, isGroup) + * Adapter->>Router: Step 8 — setup.onInbound(jid, null, message) + * Nano->>Adapter: teardown() + * note over Adapter: Step 9 — Scope.close
the client and its daemon go with it * ``` * * Nanoclaw writes model output to its session outbox and calls `deliver` * asynchronously, after the inbound callback may already have returned. The - * per-jid entry therefore retains the newest bound reply authority instead of - * reducing a Harness turn back to a generic conversation send. + * per-jid entry therefore retains the newest bound reply authority. */ export class MoltZapAdapter implements ChannelAdapter { readonly name = MOLTZAP_CHANNEL; @@ -189,118 +182,66 @@ export class MoltZapAdapter implements ChannelAdapter { // next inbound refreshes the entry. private readonly replyRoutesByJid = new BoundedMap< string, - ConversationReplyRoute + HarnessTurn["reply"] >(MAX_TRACKED_CONVERSATIONS); - private ownAgentId: string; - private core: MoltZapChannelCore | null; - private readonly harnessClient: HarnessClientService | null; + private readonly acquireClient: HarnessClientAcquisition; + private readonly evalMode: boolean; + private ownAgentId = ""; + private harnessScope: Scope.CloseableScope | null = null; private harnessDrainFiber: Fiber.RuntimeFiber | null = null; private harnessConnected = false; private setupConfig: ChannelSetup | null = null; - private readonly evalMode: boolean; - private readonly profileName: string | null; - - private constructor(state: MoltZapAdapterState) { - this.core = state.core; - this.harnessClient = state.harnessClient; - this.ownAgentId = state.ownAgentId; - this.evalMode = state.evalMode; - this.profileName = state.profileName; - if (state.core !== null) { - this.attachCore(state.core); - } - } - static fromService( - service: ChannelService, - evalMode = false, - ): MoltZapAdapter { - return new MoltZapAdapter({ - core: new MoltZapChannelCore({ service }), - harnessClient: null, - ownAgentId: service.ownAgentId ?? "", - evalMode, - profileName: null, - }); - } - - static fromProfile(profileName: string, evalMode = false): MoltZapAdapter { - return new MoltZapAdapter({ - core: null, - harnessClient: null, - ownAgentId: "", - evalMode, - profileName, - }); + private constructor( + acquireClient: HarnessClientAcquisition, + evalMode: boolean, + ) { + this.acquireClient = acquireClient; + this.evalMode = evalMode; } /** - * Creates an adapter over an already acquired Harness client. The caller - * owns the client's scope; adapter teardown interrupts only this adapter's - * turn drain. - * @param harnessClient Adapter-facing Harness capability. + * Creates an adapter that owns one Harness client acquisition. + * + * Nanoclaw builds channel adapters from a zero-argument factory at module + * import, so no caller exists to hold a `Scope` across the adapter's + * lifetime. The acquisition stays a lazy description here and is run inside + * an adapter-owned scope by `setup`. + * @param acquireClient Scoped acquisition of the adapter-facing capability. * @param evalMode Whether first inbound creates NanoClaw eval wiring. * @returns An adapter whose replies use authorities carried by Harness turns. */ - static fromHarnessClient( - harnessClient: HarnessClientService, + static fromHarnessAcquisition( + acquireClient: HarnessClientAcquisition, evalMode = false, ): MoltZapAdapter { - return new MoltZapAdapter({ - core: null, - harnessClient, - ownAgentId: harnessClient.agentId, - evalMode, - profileName: null, - }); + return new MoltZapAdapter(acquireClient, evalMode); } setup(config: ChannelSetup) { this.setupConfig = config; - const harnessClient = this.harnessClient; - if (harnessClient !== null) { - return Effect.runPromise( - this.startHarnessDrain(harnessClient).pipe( - Effect.tap(() => - Effect.logInfo("MoltZap connected").pipe( - Effect.annotateLogs({ channel: MOLTZAP_CHANNEL }), - ), - ), - ), - ); - } return Effect.runPromise( - this.initializeCore().pipe( - Effect.flatMap((core) => core.connect()), + this.connect().pipe( Effect.tap(() => Effect.logInfo("MoltZap connected").pipe( Effect.annotateLogs({ channel: MOLTZAP_CHANNEL }), ), ), - Effect.asVoid, ), ); } teardown() { - if (this.harnessClient !== null) { - return Effect.runPromise(this.stopHarnessDrain()); - } - const core = this.core; - return Effect.runPromise( - core === null ? Effect.void : core.disconnect().pipe(Effect.asVoid), - ); + return Effect.runPromise(this.disconnect()); } isConnected(): boolean { - return this.harnessClient === null - ? (this.core?.isConnected() ?? false) - : this.harnessConnected; + return this.harnessConnected; } /** - * Outbound reply path: the reply uses the route retained by the jid's most - * recent inbound. Harness-backed routes keep their exact bound closure. + * Outbound reply path: the reply uses the authority retained by the jid's + * most recent inbound, keeping its exact bound closure. * @param platformId Value supplied to the operation. * @param args Thread identifier and outbound message supplied by Nanoclaw. * @returns The text result. @@ -320,48 +261,41 @@ export class MoltZapAdapter implements ChannelAdapter { return jid.startsWith(MOLTZAP_JID_PREFIX); } - private initializeCore() { + // The scope is opened here rather than at construction so a channel the + // host never starts spawns no daemon, and a failed acquisition leaves the + // adapter with no half-open scope to close. + private connect(): Effect.Effect { return Effect.gen( function* (this: MoltZapAdapter) { - if (this.core !== null) { - return this.core; - } - const profileName = this.profileName; - if (profileName === null) { - return yield* new MoltZapChannelError({ - reason: "MoltZap channel has no profile for initialization", - }); + if (this.harnessScope !== null) { + return; } - const service = yield* MoltZapService.make(profileName); - const core = new MoltZapChannelCore({ service }); - this.core = core; - this.ownAgentId = service.ownAgentId ?? ""; - this.attachCore(core); - return core; + const scope = yield* Scope.make(); + const client = yield* Scope.extend(this.acquireClient, scope).pipe( + Effect.onError((cause) => Scope.close(scope, Exit.failCause(cause))), + ); + this.harnessScope = scope; + this.ownAgentId = client.agentId; + yield* this.startHarnessDrain(client); }.bind(this), ); } - private attachCore(core: MoltZapChannelCore): void { - core.onInbound((msg: EnrichedInboundMessage) => - this.handleInbound(msg, { - _tag: "legacy", - conversationId: msg.conversationId, - }), + private disconnect(): Effect.Effect { + return this.stopHarnessDrain().pipe( + Effect.zipRight( + Effect.suspend(() => { + const scope = this.harnessScope; + this.harnessScope = null; + return scope === null + ? Effect.void + : Scope.close(scope, Exit.succeed(undefined)); + }), + ), ); - core.onDisconnect(() => { - Effect.runFork( - Effect.logWarning("MoltZap disconnected").pipe( - Effect.annotateLogs({ channel: MOLTZAP_CHANNEL }), - ), - ); - }); } - private deliverEffect( - jid: string, - text: string, - ): Effect.Effect { + private deliverEffect(jid: string, text: string): Effect.Effect { return Effect.gen( function* (this: MoltZapAdapter) { if (!this.ownsJid(jid)) { @@ -369,57 +303,29 @@ export class MoltZapAdapter implements ChannelAdapter { reason: `MoltZap channel does not own jid: ${jid}`, }); } - const route = this.replyRoutesByJid.get(jid); - if (route === undefined) { + const reply = this.replyRoutesByJid.get(jid); + if (reply === undefined) { return yield* new MoltZapChannelError({ reason: `MoltZap channel has no conversation for jid: ${jid}`, }); } - yield* Match.value(route).pipe( - Match.tag("harness", ({ reply }) => - this.deliverHarnessReply(reply, text), - ), - Match.tag("legacy", ({ conversationId }) => - this.deliverLegacyReply(conversationId, text), - ), - Match.exhaustive, - ); + yield* reply(text); }.bind(this), ); } - private deliverHarnessReply( - reply: HarnessTurn["reply"], - text: string, - ): Effect.Effect { - return reply(text); - } - - private deliverLegacyReply( - conversationId: ConversationId, - text: string, - ): Effect.Effect { - const core = this.core; - return core === null - ? new MoltZapChannelError({ - reason: "MoltZap channel is not connected", - }) - : core.sendReply(conversationId, text); - } - - private rememberReplyRoute(jid: string, route: ConversationReplyRoute): void { - this.replyRoutesByJid.set(jid, route); + private rememberReplyRoute(jid: string, reply: HarnessTurn["reply"]): void { + this.replyRoutesByJid.set(jid, reply); } private handleInbound( - enriched: EnrichedInboundMessage, - route: ConversationReplyRoute, + turn: HarnessTurn, ): Effect.Effect { return Effect.gen( function* (this: MoltZapAdapter) { // Own outbound replies echo back through the notification stream; the // router has no is-from-me concept, so they are dropped here. - if (enriched.isFromMe) { + if (turn.isFromMe) { return; } const config = this.setupConfig; @@ -428,16 +334,16 @@ export class MoltZapAdapter implements ChannelAdapter { } const prepared = yield* Effect.try({ try: () => { - const jid = jidFromConversationId(enriched.conversationId); - this.rememberReplyRoute(jid, route); - const isGroup = enriched.conversationMeta?.type === "group"; + const jid = jidFromConversationId(turn.conversationId); + this.rememberReplyRoute(jid, turn.reply); + const isGroup = turn.conversationMeta?.type === "group"; if (this.evalMode) { - this.ensureEvalWiring(jid, enriched, isGroup); + this.ensureEvalWiring(jid, turn, isGroup); } - config.onMetadata(jid, enriched.conversationMeta?.name, isGroup); + config.onMetadata(jid, turn.conversationMeta?.name, isGroup); return { jid, - message: this.toInboundMessage(enriched, isGroup), + message: this.toInboundMessage(turn, isGroup), }; }, catch: (cause) => @@ -470,10 +376,7 @@ export class MoltZapAdapter implements ChannelAdapter { const fiber = Effect.runFork( harnessClient.turns.pipe( Stream.runForEach((turn) => - this.handleInbound(turn, { - _tag: "harness", - reply: turn.reply, - }).pipe( + this.handleInbound(turn).pipe( Effect.catchAll((cause) => this.logHarnessTurnFailure(turn, cause), ), @@ -520,57 +423,59 @@ export class MoltZapAdapter implements ChannelAdapter { } private stopHarnessDrain(): Effect.Effect { - const fiber = this.harnessDrainFiber; - this.harnessConnected = false; - return fiber === null - ? Effect.void - : Fiber.interrupt(fiber).pipe( - Effect.ensuring( - Effect.sync(() => { - this.clearHarnessDrain(fiber); - }), - ), - Effect.asVoid, - ); + return Effect.suspend(() => { + const fiber = this.harnessDrainFiber; + this.harnessConnected = false; + return fiber === null + ? Effect.void + : Fiber.interrupt(fiber).pipe( + Effect.ensuring( + Effect.sync(() => { + this.clearHarnessDrain(fiber); + }), + ), + Effect.asVoid, + ); + }); } // Nanoclaw's router consumes the content text verbatim into prompt XML, // so structured context blocks are rendered as `` markup // here via channel-base's `xml-system-reminder` variant. - private contentFor(enriched: EnrichedInboundMessage): string { + private contentFor(turn: HarnessTurn): string { const blocks: string[] = []; const crossConv = formatCrossConv( - enriched.contextBlocks.crossConversationMessages ?? [], + turn.contextBlocks.crossConversationMessages ?? [], { ownAgentId: this.ownAgentId, markup: "xml-system-reminder" }, ); if (crossConv !== null) { blocks.push(crossConv); } - const groupFields = getGroupFields(enriched.contextBlocks.groupMetadata); + const groupFields = getGroupFields(turn.contextBlocks.groupMetadata); if (groupFields !== null) { blocks.push( formatGroupBlock(groupFields, { markup: "xml-system-reminder" }), ); } if (blocks.length === 0) { - return enriched.text; + return turn.text; } - return `${blocks.join("\n\n")}\n\n${enriched.text}`; + return `${blocks.join("\n\n")}\n\n${turn.text}`; } private toInboundMessage( - enriched: EnrichedInboundMessage, + turn: HarnessTurn, isGroup: boolean, ): InboundMessage { return { - id: enriched.id, + id: turn.id, kind: "chat", content: { - text: this.contentFor(enriched), - sender: enriched.sender.name ?? enriched.sender.id, - senderId: `${MOLTZAP_CHANNEL}:${enriched.sender.id}`, + text: this.contentFor(turn), + sender: turn.sender.name ?? turn.sender.id, + senderId: `${MOLTZAP_CHANNEL}:${turn.sender.id}`, }, - timestamp: enriched.createdAt, + timestamp: turn.createdAt, isGroup, }; } @@ -582,18 +487,18 @@ export class MoltZapAdapter implements ChannelAdapter { * container config before startup; NanoClaw's sender resolver owns user * rows. Production registrations stay out of band. * @param jid Value supplied to the operation. - * @param enriched Value supplied to the operation. + * @param turn Value supplied to the operation. * @param isGroup Value supplied to the operation. */ private ensureEvalWiring( jid: string, - enriched: EnrichedInboundMessage, + turn: HarnessTurn, isGroup: boolean, ): void { if (getMessagingGroupByPlatform(MOLTZAP_CHANNEL, jid) !== undefined) { return; } - this.createEvalWiring(jid, enriched, isGroup); + this.createEvalWiring(jid, turn, isGroup); } // Persisted policy fields come from MOLTZAP_CONTEXT_DEFAULTS so the wiring @@ -601,23 +506,23 @@ export class MoltZapAdapter implements ChannelAdapter { // the full conversation id, making the platform lookup the freshness guard. private createEvalWiring( jid: string, - enriched: EnrichedInboundMessage, + turn: HarnessTurn, isGroup: boolean, ): void { const now = new Date().toISOString(); - const shortId = enriched.conversationId.slice(0, EVAL_NAME_ID_CHARS); - const messagingGroupId = `mg-eval-${enriched.conversationId}`; + const shortId = turn.conversationId.slice(0, EVAL_NAME_ID_CHARS); + const messagingGroupId = `mg-eval-${turn.conversationId}`; createMessagingGroup({ id: messagingGroupId, channel_type: MOLTZAP_CHANNEL, platform_id: jid, - name: enriched.conversationMeta?.name ?? `eval-${shortId}`, + name: turn.conversationMeta?.name ?? `eval-${shortId}`, is_group: isGroup ? 1 : 0, unknown_sender_policy: MOLTZAP_CONTEXT_DEFAULTS.unknownSenderPolicy, created_at: now, }); createMessagingGroupAgent({ - id: `mga-eval-${enriched.conversationId}`, + id: `mga-eval-${turn.conversationId}`, messaging_group_id: messagingGroupId, agent_group_id: EVAL_AGENT_GROUP_ID, engage_mode: MOLTZAP_CONTEXT_DEFAULTS.engageMode, @@ -632,8 +537,11 @@ export class MoltZapAdapter implements ChannelAdapter { } /** - * Creates molt zap adapter. - * @param env Value supplied to the operation. + * Builds the adapter nanoclaw registers for this channel. The profile name is + * the only input the production composition needs: the slot carries the + * loopback port its daemon binds, and `harnessClientForProfile` derives the + * daemon child, the endpoint, and the checkpoint store from it. + * @param env Channel environment; read from the process environment when omitted. * @returns The created molt zap adapter. */ export function makeMoltZapAdapter( @@ -643,8 +551,8 @@ export function makeMoltZapAdapter( if (resolvedEnv.profileName === null) { return null; } - return MoltZapAdapter.fromProfile( - resolvedEnv.profileName, + return MoltZapAdapter.fromHarnessAcquisition( + harnessClientForProfile(resolvedEnv.profileName), resolvedEnv.evalMode, ); } diff --git a/packages/simulator/src/runtime/harness-adapters.integration.test.ts b/packages/simulator/src/runtime/harness-adapters.integration.test.ts index 8b07795dd..87e324b4a 100644 --- a/packages/simulator/src/runtime/harness-adapters.integration.test.ts +++ b/packages/simulator/src/runtime/harness-adapters.integration.test.ts @@ -15,9 +15,10 @@ import { type ConnectedHarnessAgent, type RegisterResponse, } from "@moltzap/client/test-utils"; -import { MoltZapAdapter } from "@moltzap/nanoclaw-channel"; +import { makeMoltZapAdapter } from "@moltzap/nanoclaw-channel"; import { createMoltzapChannelPlugin } from "@moltzap/openclaw-channel"; import { + agentConversationCreate, conversationList, type ConversationId, } from "@moltzap/protocol/conversation"; @@ -33,6 +34,7 @@ import { type CoreTestServer, } from "@moltzap/server-core/test-utils"; import { + Data, Deferred, Duration, Effect, @@ -90,6 +92,14 @@ interface OpenClawExchange extends AdapterExchange { readonly profileName: string; } +/** What one case needs before it decides who acquires the slot's client. */ +interface CaseInput { + readonly profileName: string; + readonly peerName: string; + readonly owner: RegisterResponse; + readonly peer: ConnectedHarnessAgent; +} + const OPENCLAW_CASE: AdapterCase = { kind: "openclaw", profileName: OPENCLAW_PROFILE, @@ -104,6 +114,15 @@ const NANOCLAW_CASE: AdapterCase = { peerName: NANOCLAW_PEER_NAME, }; +/** The NanoClaw factory refused the profile slot this case just wrote. */ +class MissingNanoClawAdapterError extends Data.TaggedError( + "MissingNanoClawAdapterError", +)> { + override get message(): string { + return "the NanoClaw factory returned no adapter"; + } +} + const toError = (cause: unknown): Error => cause instanceof Error ? cause : new Error(String(cause)); @@ -409,10 +428,53 @@ const readNanoClawText = (content: unknown): string => { return content.text; }; -const runNanoClawExchange = (exchange: AdapterExchange) => +const createPeerDm = ( + peer: ConnectedHarnessAgent, + owner: RegisterResponse, +): Effect.Effect => + peer.client + .sendRpc(agentConversationCreate, { participants: [owner.agentId] }) + .pipe( + Effect.map((created) => created.conversation.id), + Effect.mapError(toError), + ); + +// The OpenClaw plugin takes an injected client, so the test acquires the +// slot's client itself and asserts the conversation boundary through it. +const runOpenClawCase = (input: CaseInput) => + Effect.gen(function* () { + // The production composition end to end: the slot's own daemon, the + // endpoint derived from the slot, and a real file-backed checkpoint + // store — no test-only acquisition path. + const harness = yield* harnessClientForProfile(input.profileName); + const conversationId = yield* assertConversationBoundary( + harness, + input.owner, + input.peer, + input.peerName, + ); + yield* runOpenClawExchange({ + harness, + peer: input.peer, + owner: input.owner, + conversationId, + profileName: input.profileName, + }); + }); + +// The NanoClaw adapter acquires the slot's client itself, and one slot names +// one loopback port, so nothing else here may open a second daemon against +// it. The peer therefore opens the conversation. +const runNanoClawCase = (input: CaseInput) => Effect.gen(function* () { const inboundText = yield* Deferred.make(); - const adapter = MoltZapAdapter.fromHarnessClient(exchange.harness); + const adapter = makeMoltZapAdapter({ + profileName: input.profileName, + evalMode: false, + }); + if (adapter === null) { + return yield* new MissingNanoClawAdapterError(); + } yield* tryPromise(() => adapter.setup({ onInbound: (...[jid, , message]) => { @@ -431,10 +493,11 @@ const runNanoClawExchange = (exchange: AdapterExchange) => tryPromise(() => adapter.teardown()).pipe(Effect.ignore), ); + const conversationId = yield* createPeerDm(input.peer, input.owner); yield* runPeerExchange({ - peer: exchange.peer, - owner: exchange.owner, - conversationId: exchange.conversationId, + peer: input.peer, + owner: input.owner, + conversationId, expectedReply: NANOCLAW_REPLY, inboundText, }); @@ -451,6 +514,12 @@ const runAdapterCase = (adapterCase: AdapterCase) => // chosen here and written into the slot before the child starts. const mcpPort = yield* Effect.scoped(reserveTestMcpPort); + const input: CaseInput = { + profileName: adapterCase.profileName, + peerName: adapterCase.peerName, + owner, + peer, + }; yield* withTestServiceConfig( { profileName: adapterCase.profileName, @@ -461,28 +530,9 @@ const runAdapterCase = (adapterCase: AdapterCase) => mcpPort, }, Effect.scoped( - Effect.gen(function* () { - // The production composition end to end: the slot's own daemon, - // the endpoint derived from the slot, and a real file-backed - // checkpoint store — no test-only acquisition path. - const harness = yield* harnessClientForProfile( - adapterCase.profileName, - ); - const conversationId = yield* assertConversationBoundary( - harness, - owner, - peer, - adapterCase.peerName, - ); - - const exchange = { harness, peer, owner, conversationId }; - yield* adapterCase.kind === "openclaw" - ? runOpenClawExchange({ - ...exchange, - profileName: adapterCase.profileName, - }) - : runNanoClawExchange(exchange); - }), + adapterCase.kind === "openclaw" + ? runOpenClawCase(input) + : runNanoClawCase(input), ), ); }), diff --git a/packages/simulator/src/runtime/nanoclaw/process.ts b/packages/simulator/src/runtime/nanoclaw/process.ts index b338ba59b..738264682 100644 --- a/packages/simulator/src/runtime/nanoclaw/process.ts +++ b/packages/simulator/src/runtime/nanoclaw/process.ts @@ -253,8 +253,16 @@ function buildNanoclawChildEnvironment( ): Readonly> { return { ...baseEnvironment, + // The channel adapter resolves its slot by name out of this config home + // and owns the daemon it starts from that slot, so these two locate the + // whole endpoint; no MCP endpoint is passed, because the slot names the + // loopback port. MOLTZAP_PROFILE: SIMULATOR_PROFILE_NAME, MOLTZAP_CONFIG_HOME: join(runtimeDir, ".moltzap"), + // No code in this child reads the server address: the slot carries + // identity and port but not a server, so the daemon this child starts + // inherits the address through the environment and reads it there. + // Dropping it silently retargets the agent at the public default. MOLTZAP_SERVER_URL: httpBaseUrl(opts.serverUrl), MOLTZAP_EVAL_MODE: opts.autoRegisterConversations ? "1" : "0", CONTAINER_RUNTIME: "docker",