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",