diff --git a/docs/modules/client/src.mdx b/docs/modules/client/src.mdx index d23eb28c3..e3b4a2b38 100644 --- a/docs/modules/client/src.mdx +++ b/docs/modules/client/src.mdx @@ -13,7 +13,7 @@ Public barrel for the MoltZap client package. ## Public surface -### [`acquireHarnessClient`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/harness-client.ts#L211) +### [`acquireHarnessClient`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/harness-client.ts#L201) _Function_ @@ -76,6 +76,18 @@ export interface ConversationMeta { Describes conversation meta. +### [`ConversationWithParticipants`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/harness/runtime.ts#L95) + +_TypeAlias_ + +```ts +export type ConversationWithParticipants = Schema.Schema.Type< + typeof conversationWithParticipantsSchema +>; +``` + +Conversation projection carried only between the daemon and HarnessClient. + ### [`HarnessClient`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/harness-client.ts#L58) _Class_ @@ -114,7 +126,7 @@ export interface HarnessClientService { readonly startConversation: ( otherAgentNames: readonly AgentName[], initialContent: string, - ) => Effect.Effect; + ) => Effect.Effect; /** The sole receive stream owned by this scoped client. */ readonly turns: Stream.Stream; } @@ -135,7 +147,7 @@ export interface HarnessTurn extends EnrichedInboundMessage { Existing adapter presentation with reply authority bound to its live turn. -### [`makeHarnessClientLayer`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/harness-client.ts#L242) +### [`makeHarnessClientLayer`](https://github.com/chughtapan/moltzap/blob/main/packages/client/src/harness-client.ts#L232) _Function_ @@ -327,5 +339,6 @@ to that method's errors at the `call` site. ## Files - `harness-client.ts` +- `runtime.ts` - `state.ts` - `service.ts` diff --git a/docs/modules/openclaw-channel/src.mdx b/docs/modules/openclaw-channel/src.mdx index 59259f4b6..b460aa30e 100644 --- a/docs/modules/openclaw-channel/src.mdx +++ b/docs/modules/openclaw-channel/src.mdx @@ -15,7 +15,7 @@ runtime entries from `index.*` at the extension root only, so the built ## Public surface -### [`createMoltzapChannelPlugin`](https://github.com/chughtapan/moltzap/blob/main/packages/openclaw-channel/src/openclaw-entry.ts#L1226) +### [`createMoltzapChannelPlugin`](https://github.com/chughtapan/moltzap/blob/main/packages/openclaw-channel/src/openclaw-entry.ts#L1318) _Function_ @@ -38,20 +38,27 @@ and `resolveTarget` for openclaw's targeting layer. sequenceDiagram participant OC as openclaw runtime participant Plugin as moltzap plugin + participant Harness as caller-owned HarnessClient participant Core as MoltZapChannelCore participant Server as MoltZap server OC->>Plugin: startAccount(ctx) - Plugin->>Core: new MoltZapAgentClient → MoltZapChannelCore - Plugin->>Core: core.connect() — WS auth - Plugin->>Core: core.onInbound(handler) — register dispatch - Core->>Plugin: enriched message arrives + alt HarnessClient is injected + Plugin->>Harness: drain turns sequentially + Harness-->>Plugin: originating HarnessTurn + else legacy profile client + Plugin->>Core: new MoltZapAgentClient → MoltZapChannelCore + Plugin->>Core: core.connect() — WS auth + Plugin->>Core: core.onInbound(handler) — register dispatch + Core->>Plugin: enriched message arrives + Plugin->>Plugin: bind HarnessTurn reply authority + end Plugin->>OC: dispatchReplyWithBufferedBlockDispatcher note over OC: agent pipeline → LLM - OC->>Plugin: deliver(payload, opts) — createReplyDeliver - Plugin->>Server: core.sendReply(conversationId, text) + OC->>Plugin: deliver(payload, opts) — createHarnessReplyDeliver + Plugin->>Plugin: turn.reply(text) + Plugin->>Server: core ingress bridge sends reply OC->>Plugin: stopAccount(ctx) - Plugin->>Core: core.disconnect() - Plugin->>Plugin: activeClients.delete(account) + Plugin->>Plugin: stop owned drain or disconnect owned core ``` `deliver` returns `PromiseLike<boolean>` per openclaw contract; @@ -63,7 +70,7 @@ to `agent:<name>`. Other colon-prefixed shapes are rejected. **Returns:** The created moltzap channel plugin. -### [`default`](https://github.com/chughtapan/moltzap/blob/main/packages/openclaw-channel/src/openclaw-entry.ts#L1256) +### [`default`](https://github.com/chughtapan/moltzap/blob/main/packages/openclaw-channel/src/openclaw-entry.ts#L1349) _Variable_ @@ -71,7 +78,7 @@ _Variable_ const plugin = ``` -### [`moltzapChannelPlugin`](https://github.com/chughtapan/moltzap/blob/main/packages/openclaw-channel/src/openclaw-entry.ts#L1253) +### [`moltzapChannelPlugin`](https://github.com/chughtapan/moltzap/blob/main/packages/openclaw-channel/src/openclaw-entry.ts#L1346) _Variable_ @@ -84,7 +91,7 @@ Shared singleton so a single registration reuses the same `activeClients` closure across `startAccount` and `sendText`. Tests import this directly to assert against that shared state. -### [`MoltzapChannelPlugin`](https://github.com/chughtapan/moltzap/blob/main/packages/openclaw-channel/src/openclaw-entry.ts#L1244) +### [`MoltzapChannelPlugin`](https://github.com/chughtapan/moltzap/blob/main/packages/openclaw-channel/src/openclaw-entry.ts#L1337) _TypeAlias_ diff --git a/packages/client/src/MODULE.md b/packages/client/src/MODULE.md index c0e98bcd9..cd586d399 100644 --- a/packages/client/src/MODULE.md +++ b/packages/client/src/MODULE.md @@ -8,7 +8,7 @@ Public barrel for the MoltZap client package. ## Public surface -### [`acquireHarnessClient`](./harness-client.ts#L211) +### [`acquireHarnessClient`](./harness-client.ts#L201) _Function_ @@ -71,6 +71,18 @@ export interface ConversationMeta { Describes conversation meta. +### [`ConversationWithParticipants`](./harness/runtime.ts#L95) + +_TypeAlias_ + +```ts +export type ConversationWithParticipants = Schema.Schema.Type< + typeof conversationWithParticipantsSchema +>; +``` + +Conversation projection carried only between the daemon and HarnessClient. + ### [`HarnessClient`](./harness-client.ts#L58) _Class_ @@ -109,7 +121,7 @@ export interface HarnessClientService { readonly startConversation: ( otherAgentNames: readonly AgentName[], initialContent: string, - ) => Effect.Effect; + ) => Effect.Effect; /** The sole receive stream owned by this scoped client. */ readonly turns: Stream.Stream; } @@ -130,7 +142,7 @@ export interface HarnessTurn extends EnrichedInboundMessage { Existing adapter presentation with reply authority bound to its live turn. -### [`makeHarnessClientLayer`](./harness-client.ts#L242) +### [`makeHarnessClientLayer`](./harness-client.ts#L232) _Function_ @@ -322,5 +334,6 @@ to that method's errors at the `call` site. ## Files - `harness-client.ts` +- `runtime.ts` - `state.ts` - `service.ts` diff --git a/packages/client/src/harness-client.test.ts b/packages/client/src/harness-client.test.ts index 4c5d8cf23..8573ab67b 100644 --- a/packages/client/src/harness-client.test.ts +++ b/packages/client/src/harness-client.test.ts @@ -569,7 +569,7 @@ const rejectsUnexpectedTurnFields = async () => { } }; -const startsConversationWithCanonicalProjection = async () => { +const startsConversationWithMcpLocalParticipants = async () => { const observedStarts: HarnessStartConversationInput[] = []; const running = await startHarnessServer( makeHarnessHandler([], true, observedStarts), @@ -594,14 +594,7 @@ const startsConversationWithCanonicalProjection = async () => { initialContent: INITIAL_CONTENT, }, ]); - expect(started).toEqual({ - id: STARTED_CONVERSATION.id, - name: STARTED_CONVERSATION.name, - createdBy: STARTED_CONVERSATION.createdBy, - createdAt: STARTED_CONVERSATION.createdAt, - updatedAt: STARTED_CONVERSATION.updatedAt, - }); - expect(started).not.toHaveProperty("participants"); + expect(started).toEqual(STARTED_CONVERSATION); } finally { await Effect.runPromise(Scope.close(running.scope, Exit.void)); } @@ -675,10 +668,10 @@ const abortsReplyCallWhenInterrupted = async () => { } }; -// @agent-code-guard/regression-only: the scoped loopback boundary pins the canonical start projection and every reply closure to its originating turn without suppression. +// @agent-code-guard/regression-only: the scoped loopback boundary preserves local participant enrichment and pins every reply closure to its originating turn without suppression. describe("HarnessClient", () => { - it("starts a conversation and projects its MCP-local result to the canonical shape", () => - startsConversationWithCanonicalProjection()); + it("starts a conversation and preserves MCP-local participants", () => + startsConversationWithMcpLocalParticipants()); it("sends every reply through the originating conversation after later turns", () => preservesBoundConversation()); it("rejects a server without the harness events extension", () => diff --git a/packages/client/src/harness-client.ts b/packages/client/src/harness-client.ts index 992fd1931..1a0656804 100644 --- a/packages/client/src/harness-client.ts +++ b/packages/client/src/harness-client.ts @@ -1,9 +1,6 @@ import * as KeyValueStore from "@effect/platform/KeyValueStore"; import { Context, Effect, Layer, Schema, Stream, type Scope } from "effect"; -import type { - Conversation, - conversationSearch, -} from "@moltzap/protocol/conversation"; +import type { conversationSearch } from "@moltzap/protocol/conversation"; import { agentsSearch, type AgentId, @@ -35,6 +32,9 @@ import { } from "./harness/index.js"; import { statusCommandRpc } from "./local-daemon-rpc.js"; +/** MCP-local conversation projection including participant identities. */ +export type { ConversationWithParticipants } from "./harness/index.js"; + /** Existing adapter presentation with reply authority bound to its live turn. */ export interface HarnessTurn extends EnrichedInboundMessage { /** Sends model output through the MCP reply route captured by this turn. */ @@ -49,7 +49,7 @@ export interface HarnessClientService { readonly startConversation: ( otherAgentNames: readonly AgentName[], initialContent: string, - ) => Effect.Effect; + ) => Effect.Effect; /** The sole receive stream owned by this scoped client. */ readonly turns: Stream.Stream; } @@ -121,21 +121,11 @@ const readActiveAgentId = ( }), ); -const projectConversation = ( - conversation: ConversationWithParticipants, -): Conversation => ({ - id: conversation.id, - ...(conversation.name === undefined ? {} : { name: conversation.name }), - createdBy: conversation.createdBy, - createdAt: conversation.createdAt, - updatedAt: conversation.updatedAt, -}); - const startConversation = ( session: HarnessClientInternalService, otherAgentNames: readonly AgentName[], initialContent: string, -): Effect.Effect => +): Effect.Effect => session .callTool(HARNESS_START_CONVERSATION_TOOL, { otherAgentNames, @@ -143,7 +133,7 @@ const startConversation = ( }) .pipe( Effect.flatMap(decodeHarnessStartConversationResult), - Effect.map(({ conversation }) => projectConversation(conversation)), + Effect.map(({ conversation }) => conversation), Effect.mapError(asError), ); diff --git a/packages/client/src/index.ts b/packages/client/src/index.ts index 384360ac8..9f74a5368 100644 --- a/packages/client/src/index.ts +++ b/packages/client/src/index.ts @@ -21,4 +21,5 @@ export { type HarnessClientOptions, type HarnessClientService, type HarnessTurn, + type ConversationWithParticipants, } from "./harness-client.js"; diff --git a/packages/nanoclaw-channel/AGENTS.md b/packages/nanoclaw-channel/AGENTS.md index 4c788e53c..8b4e6d3d1 100644 --- a/packages/nanoclaw-channel/AGENTS.md +++ b/packages/nanoclaw-channel/AGENTS.md @@ -9,8 +9,10 @@ channel plugins. - `src/channels/moltzap.ts` — `MoltZapAdapter`, the entry point (package `main`); implements nanoclaw's `ChannelAdapter` contract - over `MoltZapChannelCore` from `@moltzap/client/channel-base` and - self-registers via `registerChannelAdapter`. + 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. - `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` @@ -23,7 +25,7 @@ channel plugins. - **Platform id (JID)** — channel-level addressing string, `mz:`; `jidFromConversationId` converts one way, and - replies read the branded conversation id back from the per-jid map. + replies read the latest bound route back from the per-jid map. - **Wiring** — nanoclaw routes by `(channel_type, platform_id)` → `messaging_groups` → `messaging_group_agents`. Production wirings are provisioned out of band. @@ -35,13 +37,16 @@ channel plugins. ## Code -- `handleInbound` awaits the host turn rather than forking it. That - binds a reply to the turn that produced it: the per-jid - conversation entry holds the newest inbound, so a reply outliving - its own turn would address the wrong conversation. +- 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); send failures keep - their `ServiceRpcError` type. + unknown conversation, disconnected channel); 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. @@ -57,3 +62,6 @@ channel plugins. - 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. diff --git a/packages/nanoclaw-channel/src/channels/moltzap.test.ts b/packages/nanoclaw-channel/src/channels/moltzap.test.ts index e19d59655..f2e8dbb79 100644 --- a/packages/nanoclaw-channel/src/channels/moltzap.test.ts +++ b/packages/nanoclaw-channel/src/channels/moltzap.test.ts @@ -1,6 +1,10 @@ -import { describe, expect, it as vitestIt } from "vitest"; +import { describe, expect, it as vitestIt, vi } from "vitest"; import { live as it } from "@effect/vitest"; -import { Effect, Either } from "effect"; +import { Data, Deferred, Effect, Either, Queue, Stream } from "effect"; +import type { + HarnessClientService, + HarnessTurn, +} from "@moltzap/client/harness-client"; import { buildMessage, createFakeChannelService, @@ -55,6 +59,17 @@ interface Harness { readonly adapter: MoltZapAdapter; } +interface HarnessClientReply { + readonly route: string; + readonly payload: string; +} + +interface HarnessClientFixture { + readonly client: HarnessClientService; + readonly replies: HarnessClientReply[]; + readonly turns: Queue.Queue; +} + const AGENT_SELF = "agent-self"; const AGENT_ALICE = "agent-alice"; const AGENT_BOB = "agent-bob"; @@ -127,6 +142,17 @@ const SYSTEM_REMINDER_CLOSE_PATTERN = /<\/system-reminder>/g; const MESSAGES_OPEN_PATTERN = //g; 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 HARNESS_REPLY_FAILURE_PATTERN = /HarnessReplyTestError/; + +class MetadataCallbackTestError extends Data.TaggedError( + "MetadataCallbackTestError", +)> {} + +class HarnessReplyTestError extends Data.TaggedError("HarnessReplyTestError")< + Record +> {} function createRecordedSetup(): RecordedChannelSetup { const received: ReceivedMessage[] = []; @@ -147,6 +173,81 @@ 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); + let failNext = true; + return { + ...setup, + onMetadata: (jid, name, isGroup) => { + if (failNext) { + failNext = false; + throw new MetadataCallbackTestError(); + } + setup.metadata.push({ jid, name, isGroup }); + setup.callOrder.push(ON_METADATA); + }, + }; +} + +function createHarnessClientFixture(): HarnessClientFixture { + const turns = Effect.runSync(Queue.unbounded()); + const replies: HarnessClientReply[] = []; + return { + client: { + agentId: testAgentId(AGENT_SELF), + startConversation: () => + Effect.dieMessage("startConversation is not used by these tests"), + turns: Stream.fromQueue(turns), + }, + replies, + turns, + }; +} + +function makeHarnessTurn( + fixture: HarnessClientFixture, + options: { + readonly conversationId: string; + readonly messageId: string; + readonly route: string; + readonly text?: string; + }, +): HarnessTurn { + return { + id: testMessageId(options.messageId), + conversationId: testConversationId(options.conversationId), + sender: { id: testAgentId(AGENT_ALICE), name: ALICE_NAME }, + text: options.text ?? HI_NANOCLAW, + isFromMe: false, + createdAt: MESSAGE_CREATED_AT, + conversationMeta: { type: "dm", participants: [] }, + contextBlocks: {}, + reply: (payload) => + Effect.sync(() => { + fixture.replies.push({ route: options.route, payload }); + }), + }; +} + function createHarness(evalMode = false): Harness { const fake = createFakeChannelService({ ownAgentId: AGENT_SELF }); const config = createRecordedSetup(); @@ -415,6 +516,220 @@ function overlappingTurnsStaySerialized() { }); } +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)), + ); +} + +function harnessReplyFailureHasNoFallback() { + const fixture = createHarnessClientFixture(); + const reply = vi + .fn() + .mockReturnValue(Effect.fail(new HarnessReplyTestError())); + const turn = { + ...makeHarnessTurn(fixture, { + 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)), + ); +} + +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)), + ); +} + +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)), + ); +} + +function harnessLateDeliveryUsesRetainedAuthority() { + const fixture = createHarnessClientFixture(); + const turn = makeHarnessTurn(fixture, { + 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)), + ); +} + function mapsEnrichedMessageToInboundMessage() { const harness = createHarness(); return Effect.gen(function* () { @@ -763,6 +1078,34 @@ describe("MoltZapAdapter turn serialization", () => { ); }); +// @agent-code-guard/regression-only: controlled queues and callbacks pin the exact asynchronous NanoClaw delivery and drain lifecycle. +describe("MoltZapAdapter HarnessClient behavior", () => { + it( + "routes every deliver call through the latest bound turn reply", + harnessRepliesUseLatestBoundTurn, + ); + it( + "propagates reply failure without a legacy fallback", + harnessReplyFailureHasNoFallback, + ); + it( + "drains injected Harness turns sequentially", + harnessTurnsDrainSequentially, + ); + it( + "can restart its drain without closing the caller-owned client", + harnessTeardownLeavesClientCallerOwned, + ); + it( + "continues after a synchronous metadata callback failure", + harnessMetadataFailureDoesNotStopDrain, + ); + it( + "uses a retained reply authority after its receive stream completes", + harnessLateDeliveryUsesRetainedAuthority, + ); +}); + describe("MoltZapAdapter inbound projection", () => { it( "maps enriched message to InboundMessage with mz prefix", diff --git a/packages/nanoclaw-channel/src/channels/moltzap.ts b/packages/nanoclaw-channel/src/channels/moltzap.ts index c11d23b74..18e6bb147 100644 --- a/packages/nanoclaw-channel/src/channels/moltzap.ts +++ b/packages/nanoclaw-channel/src/channels/moltzap.ts @@ -1,6 +1,19 @@ /* eslint-disable jsdoc/text-escaping -- mermaid sequenceDiagram blocks need literal `
` (HTML5) for renderer compatibility; the escape would render as literal text. */ -import { Config, ConfigProvider, Data, Effect, Option } from "effect"; +import { + Config, + ConfigProvider, + Data, + Effect, + Fiber, + Match, + Option, + Stream, +} from "effect"; import { MoltZapService, type ServiceRpcError } from "@moltzap/client"; +import type { + HarnessClientService, + HarnessTurn, +} from "@moltzap/client/harness-client"; import type { ConversationId } from "@moltzap/protocol/conversation"; import { BoundedMap, @@ -28,8 +41,8 @@ import { import type { MessagingGroupAgent } from "../types.js"; // `MoltZapChannelError` covers nanoclaw's host-shape failures: un-owned jid, -// unknown conversation, disconnected channel. Send failures keep their own -// `ServiceRpcError` type. +// unknown conversation, disconnected channel. Reply failures keep the error +// type supplied by their backing client. class MoltZapChannelError extends Data.TaggedError("MoltZapChannelError")<{ readonly reason: string; }> { @@ -82,8 +95,8 @@ const moltZapChannelEnv = Config.all({ /** * MoltZap conversationId → nanoclaw platform id. The router addresses * conversations by `(channelType, platformId)`; this channel uses - * `mz:` platform ids, and replies read the branded - * conversation id back from the per-jid map rather than re-parsing the jid. + * `mz:` platform ids, and replies read their bound route from + * the per-jid map rather than re-parsing the jid. * @param conversationId Value supplied to the operation. * @returns The jid from conversation id result. */ @@ -124,33 +137,45 @@ function extractOutboundText(message: OutboundMessage): string | 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. Wraps `MoltZapChannelCore` from - * `@moltzap/client` and presents nanoclaw's `ChannelAdapter` contract. + * Nanoclaw channel adapter for MoltZap. Presents Nanoclaw's + * `ChannelAdapter` contract over either the transitional channel core or an + * injected `HarnessClient`. * * ```mermaid * sequenceDiagram - * participant Core as MoltZapChannelCore (@moltzap/client) + * participant Source as HarnessClient or MoltZapChannelCore * participant Handler as handleInbound (this adapter) * participant Router as nanoclaw router - * Core->>Handler: onInbound(enriched)
WS frame decoded + enriched + * Source->>Handler: HarnessTurn or enriched inbound * note over Handler: Step 1 — jidFromConversationId
platformId = "mz:" + conversationId - * note over Handler: Step 2 — rememberConversation
conversationsByJid.set(jid, 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 — turn resolves
awaited, so the reply binds to its own turn + * Router-->>Handler: Step 6 — callback resolves * ``` * - * The per-jid conversation entry is only sound because Step 6 is awaited: it - * holds the newest inbound, so a reply that outlived its own turn would - * address a conversation it did not come from. Awaiting keeps at most one turn - * per adapter in flight, which matches the core's single-fiber inbound drain. + * 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. */ export class MoltZapAdapter implements ChannelAdapter { readonly name = MOLTZAP_CHANNEL; @@ -158,22 +183,26 @@ export class MoltZapAdapter implements ChannelAdapter { readonly supportsThreads = false; readonly defaults = MOLTZAP_DEFAULTS; - // Per-jid memory of the branded conversation id from the most recent - // inbound. Keeping the branded id avoids re-decoding it on every reply. - // Bounded: an evicted conversation degrades to the unknown-jid deliver - // error until its next inbound refreshes the entry. - private readonly conversationsByJid = new BoundedMap< + // Nanoclaw delivers model output asynchronously through a jid, so the + // newest inbound for that jid retains its exact reply route. Bounded: an + // evicted conversation degrades to the unknown-jid deliver error until its + // next inbound refreshes the entry. + private readonly replyRoutesByJid = new BoundedMap< string, - { readonly conversationId: ConversationId } + ConversationReplyRoute >(MAX_TRACKED_CONVERSATIONS); private ownAgentId: string; private core: MoltZapChannelCore | null; + private readonly harnessClient: HarnessClientService | 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; @@ -188,6 +217,7 @@ export class MoltZapAdapter implements ChannelAdapter { ): MoltZapAdapter { return new MoltZapAdapter({ core: new MoltZapChannelCore({ service }), + harnessClient: null, ownAgentId: service.ownAgentId ?? "", evalMode, profileName: null, @@ -197,14 +227,48 @@ export class MoltZapAdapter implements ChannelAdapter { static fromProfile(profileName: string, evalMode = false): MoltZapAdapter { return new MoltZapAdapter({ core: null, + harnessClient: null, ownAgentId: "", evalMode, profileName, }); } + /** + * 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. + * @param evalMode Whether first inbound creates NanoClaw eval wiring. + * @returns An adapter whose replies use authorities carried by Harness turns. + */ + static fromHarnessClient( + harnessClient: HarnessClientService, + evalMode = false, + ): MoltZapAdapter { + return new MoltZapAdapter({ + core: null, + harnessClient, + ownAgentId: harnessClient.agentId, + evalMode, + profileName: null, + }); + } + 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()), @@ -219,6 +283,9 @@ export class MoltZapAdapter implements ChannelAdapter { } 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), @@ -226,13 +293,14 @@ export class MoltZapAdapter implements ChannelAdapter { } isConnected(): boolean { - return this.core?.isConnected() ?? false; + return this.harnessClient === null + ? (this.core?.isConnected() ?? false) + : this.harnessConnected; } /** - * Outbound reply path: the reply addresses the conversation recorded by the - * jid's most recent inbound, which is the turn the router is answering - * because `handleInbound` awaits that turn. + * Outbound reply path: the reply uses the route retained by the jid's most + * recent inbound. Harness-backed routes keep their exact bound closure. * @param platformId Value supplied to the operation. * @param args Thread identifier and outbound message supplied by Nanoclaw. * @returns The text result. @@ -275,7 +343,12 @@ export class MoltZapAdapter implements ChannelAdapter { } private attachCore(core: MoltZapChannelCore): void { - core.onInbound((msg: EnrichedInboundMessage) => this.handleInbound(msg)); + core.onInbound((msg: EnrichedInboundMessage) => + this.handleInbound(msg, { + _tag: "legacy", + conversationId: msg.conversationId, + }), + ); core.onDisconnect(() => { Effect.runFork( Effect.logWarning("MoltZap disconnected").pipe( @@ -288,7 +361,7 @@ export class MoltZapAdapter implements ChannelAdapter { private deliverEffect( jid: string, text: string, - ): Effect.Effect { + ): Effect.Effect { return Effect.gen( function* (this: MoltZapAdapter) { if (!this.ownsJid(jid)) { @@ -296,76 +369,171 @@ export class MoltZapAdapter implements ChannelAdapter { reason: `MoltZap channel does not own jid: ${jid}`, }); } - const conversation = this.conversationsByJid.get(jid); - if (conversation === undefined) { + const route = this.replyRoutesByJid.get(jid); + if (route === undefined) { return yield* new MoltZapChannelError({ reason: `MoltZap channel has no conversation for jid: ${jid}`, }); } - const core = this.core; - if (core === null) { - return yield* new MoltZapChannelError({ - reason: "MoltZap channel is not connected", - }); - } - yield* core.sendReply(conversation.conversationId, text); + yield* Match.value(route).pipe( + Match.tag("harness", ({ reply }) => + this.deliverHarnessReply(reply, text), + ), + Match.tag("legacy", ({ conversationId }) => + this.deliverLegacyReply(conversationId, text), + ), + Match.exhaustive, + ); }.bind(this), ); } - private rememberConversation( - jid: string, - enriched: EnrichedInboundMessage, - ): void { - this.conversationsByJid.set(jid, { - conversationId: enriched.conversationId, - }); + 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); } - // The host turn is awaited rather than forked, which is what keeps a reply - // bound to the turn that produced it. The core drains inbound work on a - // single fiber, so returning before the turn finishes would let a later - // inbound overwrite the per-jid conversation entry while the earlier reply - // is still pending, and that reply would then address the newer - // conversation. Awaiting costs conversation-level concurrency, which the - // core does not offer anyway. private handleInbound( enriched: EnrichedInboundMessage, + route: ConversationReplyRoute, ): Effect.Effect { - return Effect.suspend(() => { - // 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) { - return Effect.void; - } - const config = this.setupConfig; - if (config === null) { - return Effect.void; - } - const jid = jidFromConversationId(enriched.conversationId); - this.rememberConversation(jid, enriched); - const isGroup = enriched.conversationMeta?.type === "group"; - if (this.evalMode) { - this.ensureEvalWiring(jid, enriched, isGroup); - } - config.onMetadata(jid, enriched.conversationMeta?.name, isGroup); - return Effect.tryPromise({ - try: () => - Promise.resolve( - config.onInbound( + 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) { + return; + } + const config = this.setupConfig; + if (config === null) { + return; + } + const prepared = yield* Effect.try({ + try: () => { + const jid = jidFromConversationId(enriched.conversationId); + this.rememberReplyRoute(jid, route); + const isGroup = enriched.conversationMeta?.type === "group"; + if (this.evalMode) { + this.ensureEvalWiring(jid, enriched, isGroup); + } + config.onMetadata(jid, enriched.conversationMeta?.name, isGroup); + return { jid, - null, - this.toInboundMessage(enriched, isGroup), + message: this.toInboundMessage(enriched, isGroup), + }; + }, + catch: (cause) => + new MoltZapChannelError({ + reason: `nanoclaw inbound projection failed: ${String(cause)}`, + }), + }); + yield* Effect.tryPromise({ + try: () => + Promise.resolve( + config.onInbound(prepared.jid, null, prepared.message), + ), + catch: (cause) => + new MoltZapChannelError({ + reason: `nanoclaw inbound dispatch failed for ${prepared.jid}: ${String(cause)}`, + }), + }); + }.bind(this), + ); + } + + private startHarnessDrain( + harnessClient: HarnessClientService, + ): Effect.Effect { + return Effect.sync(() => { + if (this.harnessDrainFiber !== null) { + return; + } + this.harnessConnected = true; + const fiber = Effect.runFork( + harnessClient.turns.pipe( + Stream.runForEach((turn) => + this.handleInbound(turn, { + _tag: "harness", + reply: turn.reply, + }).pipe( + Effect.catchAll((cause) => + this.logHarnessTurnFailure(turn, cause), + ), + Effect.catchAllDefect((cause) => + this.logHarnessTurnFailure(turn, cause), + ), + ), + ), + Effect.catchAll((cause) => + Effect.logWarning("MoltZap disconnected").pipe( + Effect.annotateLogs({ + channel: MOLTZAP_CHANNEL, + cause: String(cause), + }), ), ), - catch: (cause) => - new MoltZapChannelError({ - reason: `nanoclaw inbound dispatch failed for ${jid}: ${String(cause)}`, - }), - }).pipe(Effect.asVoid); + ), + ); + this.harnessDrainFiber = fiber; + fiber.addObserver(() => { + this.clearHarnessDrain(fiber); + }); }); } + private logHarnessTurnFailure( + turn: HarnessTurn, + cause: unknown, + ): Effect.Effect { + return Effect.logError("MoltZap inbound dispatch failed").pipe( + Effect.annotateLogs({ + channel: MOLTZAP_CHANNEL, + conversationId: turn.conversationId, + cause: String(cause), + }), + ); + } + + private clearHarnessDrain(fiber: Fiber.RuntimeFiber): void { + if (this.harnessDrainFiber === fiber) { + this.harnessDrainFiber = null; + this.harnessConnected = false; + } + } + + 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, + ); + } + // 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. diff --git a/packages/openclaw-channel/AGENTS.md b/packages/openclaw-channel/AGENTS.md index b1834df38..c4509586b 100644 --- a/packages/openclaw-channel/AGENTS.md +++ b/packages/openclaw-channel/AGENTS.md @@ -10,9 +10,12 @@ surface. - `src/openclaw-entry.ts` — the plugin: gateway `startAccount`, notification routing, wraps `MoltZapChannelCore` (`@moltzap/client/channel-base`) for inbound enrichment and - turn ordering, projects `EnrichedInboundMessage` into + turn ordering, binds that ingress to `HarnessTurn`, and projects it into OpenClaw's `DispatchContext`, deliver callback. - `src/context-log.ts` — `writeOpenClawContextLog`. +- `src/openclaw-target.ts` — target validation and normalization. +- `src/harness-turn-delivery.ts` — bound Harness reply delivery. +- `src/openclaw-gateway-lifecycle.ts` — single-account gateway ownership. - `src/*.test.ts` — unit tests. `src/__tests__/` — integration tests, `spawn-server.ts`, echo-server fixture. @@ -33,17 +36,26 @@ surface. (`channelRuntime.reply`); OpenClaw calls `deliver` directly, never `routeReply()` (`OriginatingChannel === Surface` always holds for MoltZap→MoltZap), so the deliver callback MUST send the reply via - `core.sendReply(conversationId, text)`. -- Each final `deliver` call sends through - `core.sendReply(conversationId, text)`. A send failure returns `false` - per `OpenClawDeliver: PromiseLike` so the host may retry. + the originating `HarnessTurn.reply(text)` authority. Core-backed ingress + binds that closure to its private conversation route. +- Each final `deliver` call invokes the bound reply. A send failure returns + `false` per `OpenClawDeliver: PromiseLike` so the host may retry. +- A caller may inject an already-acquired `HarnessClientService` for an + account. The gateway owns only the sequential turn-drain fiber: stop and + abort interrupt that fiber but never close the client scope. Production + profile-to-MCP acquisition remains outside this package. Each account has + one active gateway binding; restarting it stops the prior Harness drain or + closes the prior legacy service before activating the replacement. +- Harness-backed outbound supports only agent targets, which call + `startConversation([agentName], initialContent)`. Existing-conversation + targets fail without falling back to the legacy generic send path. - Target resolution: `messaging.targetResolver` validates both target formats with no server round-trip; `directory` (`listPeers`, `listGroups` — named groups only) is live RPC returning `[]` on failure; `outbound.resolveTarget` requires a non-empty target and - rejects `:`-containing targets in no known format — a colon-free - string passes resolution and `parseConversationTarget` reads it as a - bare conversation id. + rejects `:`-containing targets in no known format. A colon-free string is + normalized to `agent:`; existing conversations require an explicit + `conv:` target. - Notification routing keys on the typed definitions from `@moltzap/protocol`: `agent/message/received` enters dispatch, non-message notifications update channel state. Sender identity diff --git a/packages/openclaw-channel/src/MODULE.md b/packages/openclaw-channel/src/MODULE.md index d9917de24..d339c2923 100644 --- a/packages/openclaw-channel/src/MODULE.md +++ b/packages/openclaw-channel/src/MODULE.md @@ -10,7 +10,7 @@ runtime entries from `index.*` at the extension root only, so the built ## Public surface -### [`createMoltzapChannelPlugin`](./openclaw-entry.ts#L1226) +### [`createMoltzapChannelPlugin`](./openclaw-entry.ts#L1318) _Function_ @@ -33,20 +33,27 @@ and `resolveTarget` for openclaw's targeting layer. sequenceDiagram participant OC as openclaw runtime participant Plugin as moltzap plugin + participant Harness as caller-owned HarnessClient participant Core as MoltZapChannelCore participant Server as MoltZap server OC->>Plugin: startAccount(ctx) - Plugin->>Core: new MoltZapAgentClient → MoltZapChannelCore - Plugin->>Core: core.connect() — WS auth - Plugin->>Core: core.onInbound(handler) — register dispatch - Core->>Plugin: enriched message arrives + alt HarnessClient is injected + Plugin->>Harness: drain turns sequentially + Harness-->>Plugin: originating HarnessTurn + else legacy profile client + Plugin->>Core: new MoltZapAgentClient → MoltZapChannelCore + Plugin->>Core: core.connect() — WS auth + Plugin->>Core: core.onInbound(handler) — register dispatch + Core->>Plugin: enriched message arrives + Plugin->>Plugin: bind HarnessTurn reply authority + end Plugin->>OC: dispatchReplyWithBufferedBlockDispatcher note over OC: agent pipeline → LLM - OC->>Plugin: deliver(payload, opts) — createReplyDeliver - Plugin->>Server: core.sendReply(conversationId, text) + OC->>Plugin: deliver(payload, opts) — createHarnessReplyDeliver + Plugin->>Plugin: turn.reply(text) + Plugin->>Server: core ingress bridge sends reply OC->>Plugin: stopAccount(ctx) - Plugin->>Core: core.disconnect() - Plugin->>Plugin: activeClients.delete(account) + Plugin->>Plugin: stop owned drain or disconnect owned core ``` `deliver` returns `PromiseLike<boolean>` per openclaw contract; @@ -58,7 +65,7 @@ to `agent:<name>`. Other colon-prefixed shapes are rejected. **Returns:** The created moltzap channel plugin. -### [`default`](./openclaw-entry.ts#L1256) +### [`default`](./openclaw-entry.ts#L1349) _Variable_ @@ -66,7 +73,7 @@ _Variable_ const plugin = ``` -### [`moltzapChannelPlugin`](./openclaw-entry.ts#L1253) +### [`moltzapChannelPlugin`](./openclaw-entry.ts#L1346) _Variable_ @@ -79,7 +86,7 @@ Shared singleton so a single registration reuses the same `activeClients` closure across `startAccount` and `sendText`. Tests import this directly to assert against that shared state. -### [`MoltzapChannelPlugin`](./openclaw-entry.ts#L1244) +### [`MoltzapChannelPlugin`](./openclaw-entry.ts#L1337) _TypeAlias_ diff --git a/packages/openclaw-channel/src/README.md b/packages/openclaw-channel/src/README.md new file mode 100644 index 000000000..2890c539d --- /dev/null +++ b/packages/openclaw-channel/src/README.md @@ -0,0 +1,16 @@ +# OpenClaw channel source + +This tree adapts MoltZap conversations to OpenClaw's channel plugin contract. + +- `openclaw-entry.ts` composes account lifecycle, directory, inbound dispatch, + and outbound delivery. +- `openclaw-target.ts` validates and normalizes agent and conversation targets. +- `harness-turn-delivery.ts` binds OpenClaw final output to the originating + Harness turn reply. +- `openclaw-gateway-lifecycle.ts` keeps one active adapter binding per account. +- `context-log.ts` writes the optional presentation-context log. +- `__tests__/` and `test-utils/` contain integration fixtures; adjacent test + files pin the public plugin behavior. + +Consumers load the package entrypoint. These source modules remain internal +composition details. diff --git a/packages/openclaw-channel/src/harness-turn-delivery.test.ts b/packages/openclaw-channel/src/harness-turn-delivery.test.ts new file mode 100644 index 000000000..504e0533c --- /dev/null +++ b/packages/openclaw-channel/src/harness-turn-delivery.test.ts @@ -0,0 +1,127 @@ +import { live as it } from "@effect/vitest"; +import type { HarnessTurn } from "@moltzap/client/harness-client"; +import { testConversationId } from "@moltzap/client/test-utils"; +import { Data, Effect } from "effect"; +import { describe, expect, vi } from "vitest"; +import { + createHarnessReplyDeliver, + type HarnessReplyDeliver, +} from "./harness-turn-delivery.js"; + +const FIRST_CONVERSATION_ID = testConversationId( + "550e8400-e29b-41d4-a716-446655440701", +); +const SECOND_CONVERSATION_ID = testConversationId( + "550e8400-e29b-41d4-a716-446655440702", +); +const FIRST_REPLY = "first reply"; +const SECOND_REPLY = "second reply"; +const RETRY_REPLY = "retry reply"; +const PARTIAL_REPLY = "partial reply"; + +type Reply = HarnessTurn["reply"]; + +class HarnessDeliveryTestError extends Data.TaggedError( + "HarnessDeliveryTestError", +)<{ readonly cause?: unknown }> {} + +const makeTurn = ( + conversationId: HarnessTurn["conversationId"], + reply: Reply, +): HarnessTurn => ({ + id: `message-${conversationId}`, + conversationId, + sender: { id: "sender-id", name: "Sender" }, + text: "incoming text", + isFromMe: false, + createdAt: "2026-08-04T00:00:00.000Z", + contextBlocks: {}, + reply, +}); + +const invoke = ( + deliver: HarnessReplyDeliver, + payload: { readonly text?: string; readonly body?: string }, + kind: string, +): Effect.Effect => + Effect.tryPromise({ + try: () => Promise.resolve(deliver(payload, { kind })), + catch: (cause) => new HarnessDeliveryTestError({ cause }), + }); + +const invokesEveryFinalDelivery = () => + Effect.gen(function* () { + const reply = vi.fn().mockReturnValue(Effect.void); + const deliver = createHarnessReplyDeliver({ + turn: makeTurn(FIRST_CONVERSATION_ID, reply), + }); + + expect(yield* invoke(deliver, { text: FIRST_REPLY }, "final")).toBe(true); + expect(yield* invoke(deliver, { text: SECOND_REPLY }, "final")).toBe(true); + expect(reply.mock.calls).toEqual([[FIRST_REPLY], [SECOND_REPLY]]); + }); + +const keepsOriginatingAuthority = () => + Effect.gen(function* () { + const firstReply = vi.fn().mockReturnValue(Effect.void); + const secondReply = vi.fn().mockReturnValue(Effect.void); + const firstDeliver = createHarnessReplyDeliver({ + turn: makeTurn(FIRST_CONVERSATION_ID, firstReply), + }); + const secondDeliver = createHarnessReplyDeliver({ + turn: makeTurn(SECOND_CONVERSATION_ID, secondReply), + }); + + yield* invoke(secondDeliver, { text: SECOND_REPLY }, "final"); + yield* invoke(firstDeliver, { text: FIRST_REPLY }, "final"); + + expect(firstReply).toHaveBeenCalledExactlyOnceWith(FIRST_REPLY); + expect(secondReply).toHaveBeenCalledExactlyOnceWith(SECOND_REPLY); + }); + +const ignoresNonFinalAndEmptyDelivery = () => + Effect.gen(function* () { + const reply = vi.fn().mockReturnValue(Effect.void); + const deliver = createHarnessReplyDeliver({ + turn: makeTurn(FIRST_CONVERSATION_ID, reply), + }); + + expect(yield* invoke(deliver, { text: PARTIAL_REPLY }, "tool")).toBe(true); + expect(yield* invoke(deliver, {}, "final")).toBe(true); + expect(reply).not.toHaveBeenCalled(); + }); + +const retriesSameAuthorityAfterFailure = () => + Effect.gen(function* () { + const reply = vi + .fn() + .mockReturnValueOnce(Effect.fail(new HarnessDeliveryTestError({}))) + .mockReturnValue(Effect.void); + const deliver = createHarnessReplyDeliver({ + turn: makeTurn(FIRST_CONVERSATION_ID, reply), + }); + + expect(yield* invoke(deliver, { text: RETRY_REPLY }, "final")).toBe(false); + expect(yield* invoke(deliver, { text: RETRY_REPLY }, "final")).toBe(true); + expect(reply.mock.calls).toEqual([[RETRY_REPLY], [RETRY_REPLY]]); + }); + +// @agent-code-guard/regression-only: these examples pin OpenClaw's fixed final-delivery callback contract at the Harness boundary. +describe("Harness turn reply delivery", () => { + it( + "invokes the bound reply for every final delivery", + invokesEveryFinalDelivery, + ); + it( + "keeps delivery authority bound to its originating turn", + keepsOriginatingAuthority, + ); + it( + "does not invoke reply for non-final or empty delivery", + ignoresNonFinalAndEmptyDelivery, + ); + it( + "reports failure and invokes the same authority again on retry", + retriesSameAuthorityAfterFailure, + ); +}); diff --git a/packages/openclaw-channel/src/harness-turn-delivery.ts b/packages/openclaw-channel/src/harness-turn-delivery.ts new file mode 100644 index 000000000..35bba09de --- /dev/null +++ b/packages/openclaw-channel/src/harness-turn-delivery.ts @@ -0,0 +1,74 @@ +import type { HarnessTurn } from "@moltzap/client/harness-client"; +import { Effect } from "effect"; + +const OUTBOUND_LOG_PREVIEW_CHARS = 80; + +interface HarnessReplyLogger { + readonly info?: (message: string) => void; + readonly error?: (message: string) => void; +} + +/** OpenClaw's Promise-based delivery callback bound to one Harness turn. */ +export type HarnessReplyDeliver = ( + payload: { readonly text?: string; readonly body?: string }, + info?: { readonly kind?: string }, +) => PromiseLike; + +const logOutboundReply = ( + turn: HarnessTurn, + text: string, + log?: HarnessReplyLogger, +): Effect.Effect => + Effect.sync(() => { + log?.info?.( + `MoltZap: outbound reply to ${turn.conversationId}: ${text.slice(0, OUTBOUND_LOG_PREVIEW_CHARS)}`, + ); + }); + +const handleReplyFailure = ( + turn: HarnessTurn, + error: Error, + log?: HarnessReplyLogger, +): Effect.Effect => + Effect.sync(() => { + log?.error?.( + `MoltZap: failed to send reply to ${turn.conversationId}: ${error}`, + ); + return false; + }); + +const sendDeliveredReply = ( + turn: HarnessTurn, + text: string, + log?: HarnessReplyLogger, +): Effect.Effect => + turn.reply(text).pipe( + Effect.tap(() => logOutboundReply(turn, text, log)), + Effect.as(true), + Effect.catchAll((error) => handleReplyFailure(turn, error, log)), + ); + +/** + * Binds OpenClaw model output to the private reply authority carried by one + * Harness turn. Conversation routing never becomes delivery input. + * + * @param params Live turn and optional channel logger. + * @param params.turn Turn carrying the private reply authority. + * @param params.log Optional channel logger. + * @returns OpenClaw's delivery callback for that turn. + */ +export const createHarnessReplyDeliver = + (params: { + readonly turn: HarnessTurn; + readonly log?: HarnessReplyLogger; + }): HarnessReplyDeliver => + (payload, info) => { + if (info?.kind !== "final") { + return Promise.resolve(true); + } + const text = payload.text ?? payload.body; + if (!text) { + return Promise.resolve(true); + } + return Effect.runPromise(sendDeliveredReply(params.turn, text, params.log)); + }; diff --git a/packages/openclaw-channel/src/openclaw-entry.harness-client.test.ts b/packages/openclaw-channel/src/openclaw-entry.harness-client.test.ts new file mode 100644 index 000000000..50c7bbb5d --- /dev/null +++ b/packages/openclaw-channel/src/openclaw-entry.harness-client.test.ts @@ -0,0 +1,600 @@ +import { live as it } from "@effect/vitest"; +import type { + HarnessClientService, + HarnessTurn, +} from "@moltzap/client/harness-client"; +import { + createFakeChannelService, + testAgentId, + testConversationId, + testMessageId, +} from "@moltzap/client/test-utils"; +import { agentName } from "@moltzap/protocol/testing"; +import { Data, Effect, Fiber, Queue, Stream } from "effect"; +import { describe, expect, vi } from "vitest"; +import { createMoltzapChannelPlugin } from "./openclaw-entry.js"; + +const ACCOUNT_ID = "harness-account"; +const ACCOUNT_AGENT_NAME = "harness-agent"; +const SELF_AGENT_ID = testAgentId("550e8400-e29b-41d4-a716-446655440801"); +const SENDER_AGENT_ID = testAgentId("550e8400-e29b-41d4-a716-446655440802"); +const CONVERSATION_ID = testConversationId( + "550e8400-e29b-41d4-a716-446655440803", +); +const MESSAGE_ID = testMessageId("550e8400-e29b-41d4-a716-446655440804"); +const STARTED_CONVERSATION_ID = testConversationId( + "550e8400-e29b-41d4-a716-446655440805", +); +const CREATED_AT = "2026-08-04T00:00:00.000Z"; +const INBOUND_TEXT = "injected inbound"; +const IDENTICAL_REPLY = "same successful reply"; +const TARGET_AGENT_NAME = agentName("target-agent"); +const TARGET_AGENT = `agent:${TARGET_AGENT_NAME}`; +const TARGET_CONVERSATION = `conv:${CONVERSATION_ID}`; +const INITIAL_CONTENT = "begin through Harness"; + +type StartConversation = HarnessClientService["startConversation"]; +type TurnReply = HarnessTurn["reply"]; +type Plugin = ReturnType; +type Dispatch = ReturnType; + +interface DispatchCall { + readonly ctx: Record; + readonly dispatcherOptions: { + readonly deliver: ( + payload: { readonly text?: string; readonly body?: string }, + info?: { readonly kind?: string }, + ) => PromiseLike; + }; +} + +class HarnessClientTestError extends Data.TaggedError( + "HarnessClientTestError", +)<{ + readonly message: string; + readonly cause?: unknown; +}> {} + +class UnexpectedLegacyConstructionError extends Data.TaggedError( + "UnexpectedLegacyConstructionError", +)> {} + +function makeAccount() { + return { id: ACCOUNT_ID, agentName: ACCOUNT_AGENT_NAME }; +} + +function makeConfig() { + return { + channels: { + moltzap: { + accounts: [makeAccount()], + }, + }, + }; +} + +function makeTurn(reply: TurnReply): HarnessTurn { + return { + id: MESSAGE_ID, + conversationId: CONVERSATION_ID, + sender: { id: SENDER_AGENT_ID, name: "sender-agent" }, + text: INBOUND_TEXT, + isFromMe: false, + createdAt: CREATED_AT, + conversationMeta: { + type: "dm", + participants: [`agent:${SELF_AGENT_ID}`, `agent:${SENDER_AGENT_ID}`], + }, + contextBlocks: {}, + reply, + }; +} + +function createHarnessFixture() { + const turns = Effect.runSync(Queue.unbounded()); + const reply = vi.fn().mockReturnValue(Effect.void); + const startConversation = vi.fn().mockReturnValue( + Effect.succeed({ + id: STARTED_CONVERSATION_ID, + createdBy: SELF_AGENT_ID, + createdAt: CREATED_AT, + updatedAt: CREATED_AT, + participants: [SELF_AGENT_ID, SENDER_AGENT_ID], + }), + ); + const callerClose = vi.fn(); + const client: HarnessClientService & { readonly close: () => void } = { + agentId: SELF_AGENT_ID, + startConversation, + turns: Stream.fromQueue(turns), + close: callerClose, + }; + return { callerClose, client, reply, startConversation, turns }; +} + +function startPluginHarnessGateway(plugin: Plugin, setStatus?: Dispatch) { + const dispatch = vi.fn().mockResolvedValue({ queuedFinal: true }); + const reportStatus = setStatus ?? vi.fn(); + const abortController = new AbortController(); + const startFiber = Effect.runFork( + runPromise("start Harness gateway", () => + plugin.gateway.startAccount({ + cfg: makeConfig(), + accountId: ACCOUNT_ID, + account: makeAccount(), + abortSignal: abortController.signal, + setStatus: reportStatus, + channelRuntime: { + reply: { dispatchReplyWithBufferedBlockDispatcher: dispatch }, + }, + }), + ), + ); + return { + abortController, + dispatch, + plugin, + setStatus: reportStatus, + startFiber, + }; +} + +function startHarnessGateway(fixture: ReturnType) { + const createService = vi.fn(() => { + throw new UnexpectedLegacyConstructionError(); + }); + const harnessClientForAccount = vi.fn(() => fixture.client); + const plugin = createMoltzapChannelPlugin({ + createService, + harnessClientForAccount, + }); + return { + ...startPluginHarnessGateway(plugin), + createService, + harnessClientForAccount, + }; +} + +function runPromise( + message: string, + operation: () => PromiseLike, +): Effect.Effect { + return Effect.tryPromise({ + try: () => Promise.resolve(operation()), + catch: (cause) => new HarnessClientTestError({ message, cause }), + }); +} + +function waitForExpectation(assertion: () => void, message: string) { + return runPromise(message, () => vi.waitFor(assertion)); +} + +function waitForGatewayStart(started: { readonly setStatus: Dispatch }) { + return waitForExpectation(() => { + expect(started.setStatus).toHaveBeenCalledWith( + expect.objectContaining({ accountId: ACCOUNT_ID, connected: true }), + ); + }, "wait for Harness gateway start"); +} + +function firstDispatchCall(dispatch: Dispatch): DispatchCall { + return /* Safe because the fixture waits until dispatch has one call. */ dispatch + .mock.calls[0]?.[0] as DispatchCall; +} + +function sendText(plugin: Plugin, to: string, text: string) { + return runPromise("send Harness text", () => + plugin.outbound.sendText({ + cfg: makeConfig(), + accountId: ACCOUNT_ID, + to, + text, + }), + ); +} + +function stopAccount(plugin: Plugin) { + return runPromise("stop Harness account", () => + plugin.gateway.stopAccount({ accountId: ACCOUNT_ID }), + ); +} + +function cleanUpStart(started: ReturnType) { + return Effect.sync(() => { + started.abortController.abort(); + }).pipe(Effect.zipRight(Fiber.interrupt(started.startFiber)), Effect.asVoid); +} + +const injectedIngress = () => { + const fixture = createHarnessFixture(); + const started = startHarnessGateway(fixture); + return Effect.gen(function* () { + yield* waitForGatewayStart(started); + yield* Queue.offer(fixture.turns, makeTurn(fixture.reply)); + yield* waitForExpectation(() => { + expect(started.dispatch).toHaveBeenCalledTimes(1); + }, "wait for injected dispatch"); + + expect(firstDispatchCall(started.dispatch).ctx).toMatchObject({ + AccountId: ACCOUNT_ID, + Body: INBOUND_TEXT, + From: `agent:${SENDER_AGENT_ID}`, + OriginatingTo: TARGET_CONVERSATION, + SenderName: "sender-agent", + }); + expect(started.harnessClientForAccount).toHaveBeenCalledWith( + ACCOUNT_ID, + makeAccount(), + ); + expect(started.createService).not.toHaveBeenCalled(); + }).pipe(Effect.ensuring(cleanUpStart(started))); +}; + +const identicalSuccessfulRepliesAreSentTwice = () => { + const fixture = createHarnessFixture(); + const started = startHarnessGateway(fixture); + return Effect.gen(function* () { + yield* waitForGatewayStart(started); + yield* Queue.offer(fixture.turns, makeTurn(fixture.reply)); + yield* waitForExpectation(() => { + expect(started.dispatch).toHaveBeenCalledTimes(1); + }, "wait for reply dispatch"); + const deliver = firstDispatchCall(started.dispatch).dispatcherOptions + .deliver; + + expect( + yield* runPromise("deliver first identical reply", () => + deliver({ text: IDENTICAL_REPLY }, { kind: "final" }), + ), + ).toBe(true); + expect( + yield* runPromise("deliver second identical reply", () => + deliver({ text: IDENTICAL_REPLY }, { kind: "final" }), + ), + ).toBe(true); + expect(fixture.reply.mock.calls).toEqual([ + [IDENTICAL_REPLY], + [IDENTICAL_REPLY], + ]); + }).pipe(Effect.ensuring(cleanUpStart(started))); +}; + +const failedTurnDoesNotStopDrain = () => { + const fixture = createHarnessFixture(); + const started = startHarnessGateway(fixture); + started.dispatch.mockRejectedValueOnce( + new HarnessClientTestError({ message: "first dispatch rejected" }), + ); + return Effect.gen(function* () { + yield* waitForGatewayStart(started); + yield* Queue.offer(fixture.turns, makeTurn(fixture.reply)); + yield* Queue.offer(fixture.turns, makeTurn(fixture.reply)); + yield* waitForExpectation(() => { + expect(started.dispatch).toHaveBeenCalledTimes(2); + }, "wait for dispatch after one rejected turn"); + }).pipe(Effect.ensuring(cleanUpStart(started))); +}; + +const agentOutboundStartsConversation = () => { + const fixture = createHarnessFixture(); + const started = startHarnessGateway(fixture); + return Effect.gen(function* () { + yield* waitForGatewayStart(started); + const result = yield* sendText( + started.plugin, + TARGET_AGENT, + INITIAL_CONTENT, + ); + + if (!result.ok) { + return yield* new HarnessClientTestError({ + message: result.error.message, + cause: result.error, + }); + } + expect(fixture.startConversation).toHaveBeenCalledExactlyOnceWith( + [TARGET_AGENT_NAME], + INITIAL_CONTENT, + ); + }).pipe(Effect.ensuring(cleanUpStart(started))); +}; + +const conversationOutboundHasNoFallback = () => { + const fixture = createHarnessFixture(); + const started = startHarnessGateway(fixture); + return Effect.gen(function* () { + yield* waitForGatewayStart(started); + const result = yield* sendText( + started.plugin, + TARGET_CONVERSATION, + INITIAL_CONTENT, + ); + + expect(result.ok).toBe(false); + expect(fixture.startConversation).not.toHaveBeenCalled(); + expect(started.createService).not.toHaveBeenCalled(); + }).pipe(Effect.ensuring(cleanUpStart(started))); +}; + +const stopLeavesClientCallerOwned = () => { + const fixture = createHarnessFixture(); + const started = startHarnessGateway(fixture); + return Effect.gen(function* () { + yield* waitForGatewayStart(started); + yield* stopAccount(started.plugin); + yield* Fiber.join(started.startFiber); + + expect(fixture.callerClose).not.toHaveBeenCalled(); + expect(yield* Queue.isShutdown(fixture.turns)).toBe(false); + yield* Queue.offer(fixture.turns, makeTurn(fixture.reply)); + yield* Effect.yieldNow(); + expect(yield* Queue.size(fixture.turns)).toBe(1); + expect(started.dispatch).not.toHaveBeenCalled(); + }).pipe(Effect.ensuring(cleanUpStart(started))); +}; + +const abortLeavesClientCallerOwned = () => { + const fixture = createHarnessFixture(); + const started = startHarnessGateway(fixture); + return Effect.gen(function* () { + yield* waitForGatewayStart(started); + started.abortController.abort(); + yield* Fiber.join(started.startFiber); + + expect(fixture.callerClose).not.toHaveBeenCalled(); + expect(yield* Queue.isShutdown(fixture.turns)).toBe(false); + yield* Queue.offer(fixture.turns, makeTurn(fixture.reply)); + yield* Effect.yieldNow(); + expect(yield* Queue.size(fixture.turns)).toBe(1); + expect(started.dispatch).not.toHaveBeenCalled(); + }).pipe(Effect.ensuring(cleanUpStart(started))); +}; + +const replacingAccountStopsPreviousDrain = () => { + const firstFixture = createHarnessFixture(); + const secondFixture = createHarnessFixture(); + const createService = vi.fn(() => { + throw new UnexpectedLegacyConstructionError(); + }); + const harnessClientForAccount = vi + .fn() + .mockReturnValueOnce(firstFixture.client) + .mockReturnValueOnce(secondFixture.client); + const plugin = createMoltzapChannelPlugin({ + createService, + harnessClientForAccount, + }); + const firstStart = startPluginHarnessGateway(plugin); + let secondStart: ReturnType | undefined; + return Effect.gen(function* () { + yield* waitForGatewayStart(firstStart); + secondStart = startPluginHarnessGateway(plugin); + yield* waitForGatewayStart(secondStart); + yield* Fiber.join(firstStart.startFiber); + + yield* Queue.offer(firstFixture.turns, makeTurn(firstFixture.reply)); + yield* Queue.offer(secondFixture.turns, makeTurn(secondFixture.reply)); + yield* waitForExpectation(() => { + expect(secondStart?.dispatch).toHaveBeenCalledTimes(1); + }, "wait for replacement gateway dispatch"); + + expect(firstStart.dispatch).not.toHaveBeenCalled(); + expect(yield* Queue.size(firstFixture.turns)).toBe(1); + expect(createService).not.toHaveBeenCalled(); + yield* stopAccount(plugin); + yield* Fiber.join(secondStart.startFiber); + }).pipe( + Effect.ensuring( + Effect.suspend(() => + secondStart === undefined + ? cleanUpStart(firstStart) + : Effect.all([cleanUpStart(firstStart), cleanUpStart(secondStart)], { + discard: true, + }), + ), + ), + ); +}; + +const statusFailureReleasesGateway = () => { + const fixture = createHarnessFixture(); + const statusFailure = new HarnessClientTestError({ + message: "status callback failed", + }); + const setStatus = vi.fn(() => { + throw statusFailure; + }); + const plugin = createMoltzapChannelPlugin({ + harnessClientForAccount: () => fixture.client, + }); + const abortController = new AbortController(); + return Effect.gen(function* () { + yield* Effect.flip( + runPromise("start gateway with failed status callback", () => + plugin.gateway.startAccount({ + cfg: makeConfig(), + accountId: ACCOUNT_ID, + account: makeAccount(), + abortSignal: abortController.signal, + setStatus, + channelRuntime: { + reply: { dispatchReplyWithBufferedBlockDispatcher: vi.fn() }, + }, + }), + ), + ); + + const result = yield* sendText(plugin, TARGET_AGENT, INITIAL_CONTENT); + expect(result.ok).toBe(false); + expect(setStatus).toHaveBeenCalledTimes(1); + expect(fixture.callerClose).not.toHaveBeenCalled(); + }).pipe( + Effect.ensuring( + Effect.sync(() => { + abortController.abort(); + }), + ), + ); +}; + +const preAbortedStartDoesNotPublishConnected = () => { + const fixture = createHarnessFixture(); + const setStatus = vi.fn(); + const harnessClientForAccount = vi.fn(() => fixture.client); + const plugin = createMoltzapChannelPlugin({ harnessClientForAccount }); + const abortController = new AbortController(); + abortController.abort(); + return Effect.gen(function* () { + yield* runPromise("start pre-aborted gateway", () => + plugin.gateway.startAccount({ + cfg: makeConfig(), + accountId: ACCOUNT_ID, + account: makeAccount(), + abortSignal: abortController.signal, + setStatus, + channelRuntime: { + reply: { dispatchReplyWithBufferedBlockDispatcher: vi.fn() }, + }, + }), + ); + + expect(setStatus).not.toHaveBeenCalled(); + expect(harnessClientForAccount).toHaveBeenCalledExactlyOnceWith( + ACCOUNT_ID, + makeAccount(), + ); + expect((yield* sendText(plugin, TARGET_AGENT, INITIAL_CONTENT)).ok).toBe( + false, + ); + }); +}; + +const staleLegacyAbortKeepsReplacement = () => { + const firstFixture = createFakeChannelService({ + ownAgentId: SELF_AGENT_ID, + }); + const secondFixture = createFakeChannelService({ + ownAgentId: SELF_AGENT_ID, + }); + const createService = vi + .fn() + .mockReturnValueOnce(firstFixture.service) + .mockReturnValueOnce(secondFixture.service); + const plugin = createMoltzapChannelPlugin({ createService }); + const firstStart = startPluginHarnessGateway(plugin); + let secondStart: ReturnType | undefined; + return Effect.gen(function* () { + yield* waitForGatewayStart(firstStart); + secondStart = startPluginHarnessGateway(plugin); + yield* waitForGatewayStart(secondStart); + + expect(firstFixture.state.closeCalls.count).toBe(1); + firstStart.abortController.abort(); + yield* Fiber.join(firstStart.startFiber); + + const result = yield* sendText( + plugin, + TARGET_CONVERSATION, + INITIAL_CONTENT, + ); + expect(result.ok).toBe(true); + expect(secondFixture.state.sent).toHaveLength(1); + yield* stopAccount(plugin); + }).pipe( + Effect.ensuring( + Effect.suspend(() => + secondStart === undefined + ? cleanUpStart(firstStart) + : Effect.all([cleanUpStart(firstStart), cleanUpStart(secondStart)], { + discard: true, + }), + ), + ), + ); +}; + +const replacingLegacyWithHarnessLeavesNoFallback = () => { + const legacyFixture = createFakeChannelService({ + ownAgentId: SELF_AGENT_ID, + }); + const harnessFixture = createHarnessFixture(); + let selectHarness = false; + const plugin = createMoltzapChannelPlugin({ + createService: () => legacyFixture.service, + harnessClientForAccount: () => + selectHarness ? harnessFixture.client : undefined, + }); + const legacyStart = startPluginHarnessGateway(plugin); + let harnessStart: ReturnType | undefined; + return Effect.gen(function* () { + yield* waitForGatewayStart(legacyStart); + selectHarness = true; + harnessStart = startPluginHarnessGateway(plugin); + yield* waitForGatewayStart(harnessStart); + + expect(legacyFixture.state.closeCalls.count).toBe(1); + legacyStart.abortController.abort(); + yield* Fiber.join(legacyStart.startFiber); + yield* stopAccount(plugin); + yield* Fiber.join(harnessStart.startFiber); + + const result = yield* sendText( + plugin, + TARGET_CONVERSATION, + INITIAL_CONTENT, + ); + expect(result.ok).toBe(false); + expect(legacyFixture.state.sent).toEqual([]); + }).pipe( + Effect.ensuring( + Effect.suspend(() => + harnessStart === undefined + ? cleanUpStart(legacyStart) + : Effect.all( + [cleanUpStart(legacyStart), cleanUpStart(harnessStart)], + { discard: true }, + ), + ), + ), + ); +}; + +// @agent-code-guard/regression-only: these examples pin the caller-owned HarnessClient seam at OpenClaw's fixed gateway contract. +describe("OpenClaw HarnessClient gateway", () => { + it("dispatches turns from an injected client", injectedIngress); + it( + "sends two identical successful replies twice", + identicalSuccessfulRepliesAreSentTwice, + ); + it("continues after one turn dispatch fails", failedTurnDoesNotStopDrain); + it( + "starts a conversation for agent outbound", + agentOutboundStartsConversation, + ); + it( + "rejects conversation outbound without fallback", + conversationOutboundHasNoFallback, + ); + it("leaves the client caller-owned on stop", stopLeavesClientCallerOwned); + it("leaves the client caller-owned on abort", abortLeavesClientCallerOwned); + it( + "stops the previous drain when an account restarts", + replacingAccountStopsPreviousDrain, + ); + it( + "releases the gateway when status reporting fails", + statusFailureReleasesGateway, + ); + it( + "does not publish connected for a pre-aborted start", + preAbortedStartDoesNotPublishConnected, + ); + it( + "keeps a replacement legacy account after a stale abort", + staleLegacyAbortKeepsReplacement, + ); + it( + "removes legacy fallback when Harness replaces an account", + replacingLegacyWithHarnessLeavesNoFallback, + ); +}); diff --git a/packages/openclaw-channel/src/openclaw-entry.ts b/packages/openclaw-channel/src/openclaw-entry.ts index 80246d8d3..52cc318a5 100644 --- a/packages/openclaw-channel/src/openclaw-entry.ts +++ b/packages/openclaw-channel/src/openclaw-entry.ts @@ -14,6 +14,10 @@ */ import { MoltZapService, type ServiceRpcError } from "@moltzap/client"; +import type { + HarnessClientService, + HarnessTurn, +} from "@moltzap/client/harness-client"; import { drainPaginatedList } from "@moltzap/client/pagination"; import { MoltZapChannelCore, @@ -28,16 +32,32 @@ import { Config, ConfigProvider, Data, + Deferred, Effect, JSONSchema, Option, Schema, + Stream, } from "effect"; import { writeOpenClawContextLog, type OpenClawContextLogInput, } from "./context-log.js"; -import { agentName, agentsList } from "@moltzap/protocol/identity"; +import { createHarnessReplyDeliver } from "./harness-turn-delivery.js"; +import { + disconnectLegacyGateway, + finishHarnessClient, + registerLegacyGatewayAbort, + stopActiveGatewayAccount, + type ActiveHarnessClient, +} from "./openclaw-gateway-lifecycle.js"; +import { + isMoltZapTarget, + normalizeMoltZapTarget, + TARGET_HINT, + TARGET_PREFIX_CONVERSATION, +} from "./openclaw-target.js"; +import { agentsList } from "@moltzap/protocol/identity"; import { type ConversationId, conversationId, @@ -46,70 +66,18 @@ import { import type { ResultOf } from "@moltzap/protocol/rpc"; const CHANNEL_ID = "moltzap"; -const TARGET_PREFIX_AGENT = "agent:"; -const TARGET_PREFIX_CONVERSATION = "conv:"; -const TARGET_HINT = - 'Use an agent name or "agent:" for DMs or "conv:" for conversations'; const INBOUND_LOG_PREVIEW_CHARS = 80; const BODY_FOR_AGENT_LOG_PREVIEW_CHARS = 500; -const OUTBOUND_LOG_PREVIEW_CHARS = 80; class DispatchInboundError extends Data.TaggedError("DispatchInboundError")<{ readonly cause: unknown; readonly message: string; }> {} -const isAgentName = Schema.is(agentName); const openClawContextLogDir = Config.option( Config.string("MOLTZAP_OPENCLAW_CONTEXT_LOG_DIR"), ); -interface ResolvedMoltZapTarget { - readonly to: string; - readonly kind: "user" | "group"; - readonly display: string; -} - -function normalizeConversationTarget( - target: string, -): ResolvedMoltZapTarget | null | undefined { - if (!target.startsWith(TARGET_PREFIX_CONVERSATION)) { - return undefined; - } - const id = target.slice(TARGET_PREFIX_CONVERSATION.length); - return id.length === 0 || id.includes(":") - ? null - : { to: target, kind: "group", display: id }; -} - -function normalizeAgentTarget(target: string): ResolvedMoltZapTarget | null { - let name: string | null; - if (target.startsWith(TARGET_PREFIX_AGENT)) { - name = target.slice(TARGET_PREFIX_AGENT.length); - } else if (target.includes(":")) { - name = null; - } else { - name = target; - } - return name === null || !isAgentName(name) - ? null - : { to: `${TARGET_PREFIX_AGENT}${name}`, kind: "user", display: name }; -} - -function normalizeMoltZapTarget(raw: string): ResolvedMoltZapTarget | null { - const target = raw.trim(); - const conversation = normalizeConversationTarget(target); - if (conversation !== undefined) { - return conversation; - } - return normalizeAgentTarget(target); -} - -function isMoltZapTarget(raw: string): boolean { - const target = raw.trim(); - return normalizeMoltZapTarget(target)?.to === target; -} - function readOpenClawContextLogDir(): string | undefined { return Option.getOrUndefined( Effect.runSync( @@ -140,6 +108,16 @@ class MoltZapAgentTargetUnsupportedError extends Data.TaggedError( } } +class MoltZapConversationTargetUnsupportedError extends Data.TaggedError( + "MoltZapConversationTargetUnsupportedError", +)<{ + readonly accountId: string; +}> { + override get message(): string { + return `MoltZap Harness client for account ${this.accountId} cannot send into an existing conversation`; + } +} + class MoltZapAccountProfileMissingError extends Data.TaggedError( "MoltZapAccountProfileMissingError", )> { @@ -278,7 +256,7 @@ interface InboundDispatchInput { readonly bodyForAgent: string; readonly groupMembers?: string; readonly groupSubject?: string; - readonly enriched: EnrichedInboundMessage; + readonly turn: HarnessTurn; } interface OpenClawClientService extends ChannelService { @@ -296,6 +274,15 @@ interface MoltzapChannelPluginDeps { account: MoltZapAccount, ) => OpenClawClientService; readonly createCore?: (service: ChannelService) => MoltZapChannelCore; + /** + * Selects a caller-acquired client for one configured account. The plugin + * owns only its turn drain and never discovers, acquires, or closes the + * returned client. + */ + readonly harnessClientForAccount?: ( + profileName: string, + account: MoltZapAccount, + ) => HarnessClientService | undefined; } interface OpenClawDirectoryParams { @@ -315,10 +302,9 @@ interface OpenClawResolveTargetParams { interface InboundHandlerParams { readonly ctx: OpenClawStartAccountContext; - readonly core: MoltZapChannelCore; - readonly service: OpenClawClientService; + readonly ownAgentId?: string; readonly contextLogDir?: string; - readonly enriched: EnrichedInboundMessage; + readonly turn: HarnessTurn; } interface InboundRuntimeData { @@ -377,110 +363,37 @@ const waitForAbort = (signal: AbortSignal): Effect.Effect => ); }); -function logOutboundReply( - conversationId: string, - text: string, - log?: OpenClawLogger, -): Effect.Effect { - return Effect.sync(() => { - log?.info?.( - `MoltZap: outbound reply to ${conversationId}: ${text.slice(0, OUTBOUND_LOG_PREVIEW_CHARS)}`, - ); - }); -} - -/** - * Project a failed reply into the deliver contract's boolean. Every send - * failure is transient from the plugin's side, so the reply reports - * not-delivered and the host may retry. - * @param conversationId Value supplied to the operation. - * @param err Error to inspect. - * @param log Value supplied to the operation. - * @returns Whether the reply is considered handled. - */ -function handleReplyFailure( - conversationId: string, - err: unknown, - log?: OpenClawLogger, -): Effect.Effect { - return Effect.sync(() => { - log?.error?.(`MoltZap: failed to send reply to ${conversationId}: ${err}`); - return false; - }); -} - -function sendDeliveredReply(params: { - readonly core: MoltZapChannelCore; - readonly conversationId: ConversationId; - readonly text: string; - readonly log?: OpenClawLogger; -}): Effect.Effect { - return params.core.sendReply(params.conversationId, params.text).pipe( - Effect.tap(() => - logOutboundReply(params.conversationId, params.text, params.log), - ), - Effect.map(() => true), - Effect.catchAll((err) => - handleReplyFailure(params.conversationId, err, params.log), - ), - ); -} - -function createReplyDeliver(params: { - readonly core: MoltZapChannelCore; - readonly enriched: EnrichedInboundMessage; - readonly log?: OpenClawLogger; -}): OpenClawDeliver { - return (payload, info) => { - if (info?.kind !== "final") { - return Promise.resolve(true); - } - const text = payload.text ?? payload.body; - if (!text) { - return Promise.resolve(true); - } - return Effect.runPromise( - sendDeliveredReply({ - core: params.core, - conversationId: params.enriched.conversationId, - text, - log: params.log, - }), - ); - }; -} - /** * Render the reply-to target for an inbound message. The conversation is the * whole address. - * @param enriched Value supplied to the operation. + * @param turn Value supplied to the operation. * @returns The originating target string. */ -function originatingTarget(enriched: EnrichedInboundMessage): string { - return `${TARGET_PREFIX_CONVERSATION}${enriched.conversationId}`; +function originatingTarget(turn: HarnessTurn): string { + return `${TARGET_PREFIX_CONVERSATION}${turn.conversationId}`; } function buildInboundDispatchContext( input: InboundDispatchInput, ): Record { return { - Body: input.enriched.text, + Body: input.turn.text, BodyForAgent: input.bodyForAgent, From: input.fromId, To: input.account.agentName ?? input.accountId, - SessionKey: `agent:main:moltzap:${input.chatType === "group" ? "group" : "dm"}:${input.enriched.conversationId}`, + SessionKey: `agent:main:moltzap:${input.chatType === "group" ? "group" : "dm"}:${input.turn.conversationId}`, AccountId: input.accountId, Provider: CHANNEL_ID, Surface: CHANNEL_ID, OriginatingChannel: CHANNEL_ID, - OriginatingTo: originatingTarget(input.enriched), + OriginatingTo: originatingTarget(input.turn), ChatType: input.chatType, ...(input.groupSubject ? { GroupSubject: input.groupSubject } : {}), ...(input.groupMembers ? { GroupMembers: input.groupMembers } : {}), - ...(input.enriched.conversationMeta?.name - ? { ConversationLabel: input.enriched.conversationMeta.name } + ...(input.turn.conversationMeta?.name + ? { ConversationLabel: input.turn.conversationMeta.name } : {}), - SenderName: input.enriched.sender.name, + SenderName: input.turn.sender.name, }; } @@ -496,7 +409,6 @@ function logDispatchError( function dispatchInboundReply(params: { readonly dispatch: OpenClawReplyDispatcher; readonly input: InboundDispatchInput; - readonly core: MoltZapChannelCore; readonly log?: OpenClawLogger; }): Effect.Effect<{ queuedFinal: boolean }, unknown> { return Effect.tryPromise({ @@ -505,9 +417,8 @@ function dispatchInboundReply(params: { ctx: buildInboundDispatchContext(params.input), cfg: params.input.cfg, dispatcherOptions: { - deliver: createReplyDeliver({ - core: params.core, - enriched: params.input.enriched, + deliver: createHarnessReplyDeliver({ + turn: params.input.turn, log: params.log, }), }, @@ -520,20 +431,6 @@ function dispatchInboundReply(params: { }).pipe(Effect.tapError((err) => logDispatchError(err, params.log))); } -function disconnectCoreOnAbort( - core: MoltZapChannelCore, - activeClients: Map, - accountId: string, -): void { - Effect.runFork( - core - .disconnect() - .pipe( - Effect.ensuring(Effect.sync(() => activeClients.delete(accountId))), - ), - ); -} - function createPluginMeta() { return { id: CHANNEL_ID, @@ -716,14 +613,20 @@ function createConfigSection() { function createGatewaySection( activeClients: Map, + activeHarnessClients: Map, deps: MoltzapChannelPluginDeps, ) { return { startAccount(ctx: OpenClawStartAccountContext) { - return startGatewayAccount(ctx, activeClients, deps); + return startGatewayAccount( + ctx, + activeClients, + activeHarnessClients, + deps, + ); }, stopAccount(ctx: OpenClawStopAccountContext) { - return stopGatewayAccount(ctx, activeClients); + return stopGatewayAccount(ctx, activeClients, activeHarnessClients); }, }; } @@ -731,14 +634,18 @@ function createGatewaySection( function startGatewayAccount( ctx: OpenClawStartAccountContext, activeClients: Map, + activeHarnessClients: Map, deps: MoltzapChannelPluginDeps, ) { - return Effect.runPromise(startGatewayAccountEffect(ctx, activeClients, deps)); + return Effect.runPromise( + startGatewayAccountEffect(ctx, activeClients, activeHarnessClients, deps), + ); } function startGatewayAccountEffect( ctx: OpenClawStartAccountContext, activeClients: Map, + activeHarnessClients: Map, deps: MoltzapChannelPluginDeps, ): Effect.Effect { const { accountId, account, abortSignal, log, setStatus } = ctx; @@ -749,8 +656,20 @@ function startGatewayAccountEffect( if (profileName.length === 0) { return yield* new MoltZapAccountProfileMissingError(); } + const harnessClient = deps.harnessClientForAccount?.(profileName, account); + if (harnessClient !== undefined) { + if (abortSignal.aborted) { + return; + } + return yield* runHarnessGateway(ctx, harnessClient, { + activeClients, + activeHarnessClients, + contextLogDir, + }); + } const service = yield* createGatewayService(profileName, account, deps); const core = createGatewayCore(service, deps); + const binding = { core, service }; registerInboundHandler({ core, ctx, @@ -758,21 +677,103 @@ function startGatewayAccountEffect( contextLogDir, }); registerConnectionStatus(core, ctx); - activeClients.set(accountId, service); + yield* stopActiveGatewayAccount( + activeClients, + activeHarnessClients, + accountId, + ); + yield* Effect.sync(() => activeClients.set(accountId, service)); if (abortSignal.aborted) { - return yield* disconnectAndRemove(core, activeClients, accountId); + return yield* disconnectLegacyGateway(binding, activeClients, accountId); } - abortSignal.addEventListener( - "abort", - () => { - disconnectCoreOnAbort(core, activeClients, accountId); - }, - { once: true }, - ); + registerLegacyGatewayAbort(abortSignal, binding, activeClients, accountId); yield* connectGatewayCore(core, service, ctx, setStatus); }); } +interface HarnessGatewayRuntime { + readonly activeClients: Map; + readonly activeHarnessClients: Map; + readonly contextLogDir?: string; +} + +function runHarnessGateway( + ctx: OpenClawStartAccountContext, + client: HarnessClientService, + runtime: HarnessGatewayRuntime, +): Effect.Effect { + const { activeClients, activeHarnessClients, contextLogDir } = runtime; + return Effect.gen(function* () { + const stopSignal = yield* Deferred.make(); + const active = { client, stopSignal }; + yield* stopActiveGatewayAccount( + activeClients, + activeHarnessClients, + ctx.accountId, + ); + yield* Effect.sync(() => activeHarnessClients.set(ctx.accountId, active)); + yield* reportHarnessConnected(client, ctx).pipe( + Effect.zipRight( + Effect.raceFirst( + client.turns.pipe( + Stream.runForEach((turn) => + handleInboundMessage({ + ctx, + ownAgentId: client.agentId, + contextLogDir, + turn, + }).pipe( + Effect.withSpan("createMoltzapChannelPlugin.inboundDispatch"), + Effect.catchAll((cause) => + logHarnessTurnFailure(turn, cause, ctx.log), + ), + Effect.catchAllDefect((cause) => + logHarnessTurnFailure(turn, cause, ctx.log), + ), + ), + ), + ), + Effect.raceFirst( + waitForAbort(ctx.abortSignal), + Deferred.await(stopSignal), + ), + ), + ), + Effect.ensuring( + finishHarnessClient(activeHarnessClients, ctx.accountId, active), + ), + ); + }); +} + +function logHarnessTurnFailure( + turn: HarnessTurn, + cause: unknown, + log?: OpenClawLogger, +): Effect.Effect { + return Effect.sync(() => { + log?.error?.( + `MoltZap: inbound dispatch failed for ${turn.conversationId}: ${String(cause)}`, + ); + }); +} + +function reportHarnessConnected( + client: HarnessClientService, + ctx: OpenClawStartAccountContext, +): Effect.Effect { + return Effect.sync(() => { + ctx.log?.info?.( + `MoltZap: connected as ${ctx.account.agentName} (${client.agentId})`, + ); + ctx.setStatus({ + accountId: ctx.accountId, + connected: true, + lastConnectedAt: Date.now(), + }); + }); +} + function createGatewayService( profileName: string, account: MoltZapAccount, @@ -794,16 +795,6 @@ function createGatewayCore( return new MoltZapChannelCore({ service }); } -function disconnectAndRemove( - core: MoltZapChannelCore, - activeClients: Map, - accountId: string, -) { - return core - .disconnect() - .pipe(Effect.tap(() => Effect.sync(() => activeClients.delete(accountId)))); -} - interface RegisterInboundHandlerParams { readonly core: MoltZapChannelCore; readonly ctx: OpenClawStartAccountContext; @@ -812,67 +803,76 @@ interface RegisterInboundHandlerParams { } function registerInboundHandler(params: RegisterInboundHandlerParams): void { - params.core.onInbound((enriched) => - handleInboundMessage({ + params.core.onInbound((enriched) => { + const turn = bindCoreInboundTurn(params.core, enriched); + return handleInboundMessage({ ctx: params.ctx, - core: params.core, - service: params.service, + ownAgentId: params.service.ownAgentId, contextLogDir: params.contextLogDir, - enriched, - }).pipe(Effect.withSpan("createMoltzapChannelPlugin.inboundDispatch")), - ); + turn, + }).pipe(Effect.withSpan("createMoltzapChannelPlugin.inboundDispatch")); + }); +} + +function bindCoreInboundTurn( + core: MoltZapChannelCore, + enriched: EnrichedInboundMessage, +): HarnessTurn { + return { + ...enriched, + reply: (payload) => core.sendReply(enriched.conversationId, payload), + }; } function handleInboundMessage(params: InboundHandlerParams) { return Effect.gen(function* () { - const data = inboundRuntimeData(params.enriched, params.service); - logInboundMessage(params.enriched, data.fromId, params.ctx.log); + const data = inboundRuntimeData(params.turn, params.ownAgentId); + logInboundMessage(params.turn, data.fromId, params.ctx.log); touchInboundStatus(params.ctx); yield* writeInboundContextLog(params, data); - logCrossConversationContext(params.enriched, data, params.ctx.log); + logCrossConversationContext(params.turn, data, params.ctx.log); const dispatch = params.ctx.channelRuntime?.reply ?.dispatchReplyWithBufferedBlockDispatcher; if (!dispatch) { - logMissingDispatcher(params.enriched.conversationId, params.ctx.log); + logMissingDispatcher(params.turn.conversationId, params.ctx.log); return; } const result = yield* dispatchInboundReply({ dispatch, - input: inboundDispatchInput(params.ctx, params.enriched, data), - core: params.core, + input: inboundDispatchInput(params.ctx, params.turn, data), log: params.ctx.log, }); - logDispatchFinished(params.enriched, params.ctx.log); - logUnqueuedDispatch(params.enriched, result, params.ctx.log); + logDispatchFinished(params.turn, params.ctx.log); + logUnqueuedDispatch(params.turn, result, params.ctx.log); }); } function inboundRuntimeData( - enriched: EnrichedInboundMessage, - service: OpenClawClientService, + turn: HarnessTurn, + ownAgentId?: string, ): InboundRuntimeData { - const groupFields = getGroupFields(enriched.conversationMeta); - const crossConversationMessages = crossConversationMessagesFor(enriched); + const groupFields = getGroupFields(turn.conversationMeta); + const crossConversationMessages = crossConversationMessagesFor(turn); const crossConvBlock = formatCrossConv(crossConversationMessages, { - ownAgentId: service.ownAgentId ?? "", + ownAgentId: ownAgentId ?? "", markup: "json-header", }); return { chatType: groupFields !== null ? "group" : "direct", - fromId: `agent:${enriched.sender.id}`, + fromId: `agent:${turn.sender.id}`, crossConvBlock, crossConversationMessages, - bodyForAgent: bodyForAgent(enriched.text, crossConvBlock), + bodyForAgent: bodyForAgent(turn.text, crossConvBlock), groupSubject: groupFields?.name, groupMembers: groupMembersFor(groupFields), }; } function crossConversationMessagesFor( - enriched: EnrichedInboundMessage, + turn: HarnessTurn, ): readonly CrossConvMessage[] { - return enriched.contextBlocks.crossConversationMessages ?? []; + return turn.contextBlocks.crossConversationMessages ?? []; } function bodyForAgent(text: string, crossConvBlock: string | null): string { @@ -891,12 +891,12 @@ function groupMembersFor(fields: GroupFields | null): string | undefined { } function logInboundMessage( - enriched: EnrichedInboundMessage, + turn: HarnessTurn, fromId: string, log?: OpenClawLogger, ): void { log?.info?.( - `MoltZap: inbound from ${fromId}: ${enriched.text.slice(0, INBOUND_LOG_PREVIEW_CHARS)}`, + `MoltZap: inbound from ${fromId}: ${turn.text.slice(0, INBOUND_LOG_PREVIEW_CHARS)}`, ); } @@ -917,13 +917,13 @@ function writeInboundContextLog( logDir: params.contextLogDir, accountId: params.ctx.accountId, accountAgentName: params.ctx.account.agentName, - ownAgentId: params.service.ownAgentId, - conversationId: params.enriched.conversationId, - conversationName: params.enriched.conversationMeta?.name, + ownAgentId: params.ownAgentId, + conversationId: params.turn.conversationId, + conversationName: params.turn.conversationMeta?.name, conversationType: data.chatType, from: data.fromId, to: params.ctx.account.agentName ?? params.ctx.accountId, - body: params.enriched.text, + body: params.turn.text, bodyForAgent: data.bodyForAgent, crossConversationMessages: data.crossConversationMessages, }, @@ -932,7 +932,7 @@ function writeInboundContextLog( } function logCrossConversationContext( - enriched: EnrichedInboundMessage, + turn: HarnessTurn, data: InboundRuntimeData, log?: OpenClawLogger, ): void { @@ -940,7 +940,7 @@ function logCrossConversationContext( return; } log?.info?.( - `MoltZap: BodyForAgent has cross-conv context (${data.crossConversationMessages.length} msgs) for ${enriched.conversationId}: ${data.bodyForAgent.slice(0, BODY_FOR_AGENT_LOG_PREVIEW_CHARS)}`, + `MoltZap: BodyForAgent has cross-conv context (${data.crossConversationMessages.length} msgs) for ${turn.conversationId}: ${data.bodyForAgent.slice(0, BODY_FOR_AGENT_LOG_PREVIEW_CHARS)}`, ); } @@ -953,7 +953,7 @@ function logMissingDispatcher( function inboundDispatchInput( ctx: OpenClawStartAccountContext, - enriched: EnrichedInboundMessage, + turn: HarnessTurn, data: InboundRuntimeData, ): InboundDispatchInput { return { @@ -965,21 +965,18 @@ function inboundDispatchInput( bodyForAgent: data.bodyForAgent, groupMembers: data.groupMembers, groupSubject: data.groupSubject, - enriched, + turn, }; } -function logDispatchFinished( - enriched: EnrichedInboundMessage, - log?: OpenClawLogger, -): void { +function logDispatchFinished(turn: HarnessTurn, log?: OpenClawLogger): void { log?.info?.( - `MoltZap: dispatch finished for ${enriched.conversationId} message ${enriched.id}`, + `MoltZap: dispatch finished for ${turn.conversationId} message ${turn.id}`, ); } function logUnqueuedDispatch( - enriched: EnrichedInboundMessage, + turn: HarnessTurn, result: { readonly queuedFinal: boolean }, log?: OpenClawLogger, ): void { @@ -987,7 +984,7 @@ function logUnqueuedDispatch( return; } log?.debug?.( - `MoltZap: dispatch completed without final reply for ${enriched.conversationId}`, + `MoltZap: dispatch completed without final reply for ${turn.conversationId}`, ); } @@ -1044,18 +1041,26 @@ function logConnectionFailure(err: unknown, log?: OpenClawLogger) { function stopGatewayAccount( ctx: OpenClawStopAccountContext, activeClients: Map, + activeHarnessClients: Map, ) { - const service = activeClients.get(ctx.accountId); - if (service) { + if ( + activeHarnessClients.has(ctx.accountId) || + activeClients.has(ctx.accountId) + ) { ctx.log?.info?.("MoltZap: stopping"); - service.close(); - activeClients.delete(ctx.accountId); } - return Promise.resolve(undefined); + return Effect.runPromise( + stopActiveGatewayAccount( + activeClients, + activeHarnessClients, + ctx.accountId, + ), + ); } function createOutboundSection( activeClients: Map, + activeHarnessClients: Map, ) { return { deliveryMode: "gateway" as const, @@ -1073,7 +1078,9 @@ function createOutboundSection( text: string; accountId?: string | null; }) { - return Effect.runPromise(sendTextEffect(activeClients, ctx)); + return Effect.runPromise( + sendTextEffect(activeClients, activeHarnessClients, ctx), + ); }, }; } @@ -1153,8 +1160,78 @@ function dispatchOutbound( }); } +interface ActiveLegacyOutbound { + readonly _tag: "legacy"; + readonly accountId: string; + readonly service: OpenClawClientService; +} + +interface ActiveHarnessOutbound { + readonly _tag: "harness"; + readonly accountId: string; + readonly client: HarnessClientService; +} + +type ActiveOutbound = ActiveLegacyOutbound | ActiveHarnessOutbound; + +function getActiveOutbound( + activeClients: Map, + activeHarnessClients: Map, + accountId?: string | null, +): ActiveOutbound | undefined { + const requested = accountId?.trim(); + if (requested) { + const harness = activeHarnessClients.get(requested); + if (harness !== undefined) { + return { _tag: "harness", accountId: requested, client: harness.client }; + } + const service = activeClients.get(requested); + return service === undefined + ? undefined + : { _tag: "legacy", accountId: requested, service }; + } + if (activeClients.size + activeHarnessClients.size !== 1) { + return undefined; + } + const harness = activeHarnessClients.entries().next().value; + if (harness !== undefined) { + return { + _tag: "harness", + accountId: harness[0], + client: harness[1].client, + }; + } + const legacy = activeClients.entries().next().value; + return legacy === undefined + ? undefined + : { _tag: "legacy", accountId: legacy[0], service: legacy[1] }; +} + +function dispatchHarnessOutbound( + client: HarnessClientService, + accountId: string, + ctx: { + readonly to: string; + readonly text: string; + }, +): Effect.Effect { + const target = normalizeMoltZapTarget(ctx.to); + if (target === null) { + return Effect.fail(new MoltZapTargetMalformedError({ target: ctx.to })); + } + if (target.kind === "group") { + return Effect.fail( + new MoltZapConversationTargetUnsupportedError({ accountId }), + ); + } + return client + .startConversation([target.display], ctx.text) + .pipe(Effect.asVoid); +} + function sendTextEffect( activeClients: Map, + activeHarnessClients: Map, ctx: { cfg: OpenClawConfig; to: string; @@ -1164,13 +1241,21 @@ function sendTextEffect( ) { const requestedAccountId = ctx.accountId ?? "(unspecified)"; return Effect.gen(function* () { - const active = getActiveService(activeClients, ctx.accountId); + const active = getActiveOutbound( + activeClients, + activeHarnessClients, + ctx.accountId, + ); if (active === undefined) { return yield* new MoltZapClientNotConnectedError({ accountId: requestedAccountId, }); } - yield* dispatchOutbound(active.service, active.accountId, ctx); + if (active._tag === "harness") { + yield* dispatchHarnessOutbound(active.client, active.accountId, ctx); + } else { + yield* dispatchOutbound(active.service, active.accountId, ctx); + } return new OpenClawSendTextSuccess(); }).pipe( Effect.withSpan("createMoltzapChannelPlugin.sendText"), @@ -1198,20 +1283,27 @@ function sendTextEffect( * sequenceDiagram * participant OC as openclaw runtime * participant Plugin as moltzap plugin + * participant Harness as caller-owned HarnessClient * participant Core as MoltZapChannelCore * participant Server as MoltZap server * OC->>Plugin: startAccount(ctx) - * Plugin->>Core: new MoltZapAgentClient → MoltZapChannelCore - * Plugin->>Core: core.connect() — WS auth - * Plugin->>Core: core.onInbound(handler) — register dispatch - * Core->>Plugin: enriched message arrives + * alt HarnessClient is injected + * Plugin->>Harness: drain turns sequentially + * Harness-->>Plugin: originating HarnessTurn + * else legacy profile client + * Plugin->>Core: new MoltZapAgentClient → MoltZapChannelCore + * Plugin->>Core: core.connect() — WS auth + * Plugin->>Core: core.onInbound(handler) — register dispatch + * Core->>Plugin: enriched message arrives + * Plugin->>Plugin: bind HarnessTurn reply authority + * end * Plugin->>OC: dispatchReplyWithBufferedBlockDispatcher * note over OC: agent pipeline → LLM - * OC->>Plugin: deliver(payload, opts) — createReplyDeliver - * Plugin->>Server: core.sendReply(conversationId, text) + * OC->>Plugin: deliver(payload, opts) — createHarnessReplyDeliver + * Plugin->>Plugin: turn.reply(text) + * Plugin->>Server: core ingress bridge sends reply * OC->>Plugin: stopAccount(ctx) - * Plugin->>Core: core.disconnect() - * Plugin->>Plugin: activeClients.delete(account) + * Plugin->>Plugin: stop owned drain or disconnect owned core * ``` * * `deliver` returns `PromiseLike<boolean>` per openclaw contract; @@ -1227,6 +1319,7 @@ export function createMoltzapChannelPlugin( deps: MoltzapChannelPluginDeps = {}, ) { const activeClients = new Map(); + const activeHarnessClients = new Map(); return { id: CHANNEL_ID, @@ -1235,8 +1328,8 @@ export function createMoltzapChannelPlugin( messaging: createMessagingSection(), directory: createDirectorySection(activeClients), config: createConfigSection(), - gateway: createGatewaySection(activeClients, deps), - outbound: createOutboundSection(activeClients), + gateway: createGatewaySection(activeClients, activeHarnessClients, deps), + outbound: createOutboundSection(activeClients, activeHarnessClients), }; } diff --git a/packages/openclaw-channel/src/openclaw-gateway-lifecycle.ts b/packages/openclaw-channel/src/openclaw-gateway-lifecycle.ts new file mode 100644 index 000000000..b1cdd831c --- /dev/null +++ b/packages/openclaw-channel/src/openclaw-gateway-lifecycle.ts @@ -0,0 +1,123 @@ +import type { HarnessClientService } from "@moltzap/client/harness-client"; +import type { MoltZapChannelCore } from "@moltzap/client/channel-base"; +import { Deferred, Effect } from "effect"; + +interface ClosableGatewayService { + readonly close: () => void; +} + +interface LegacyGatewayBinding { + readonly core: MoltZapChannelCore; + readonly service: Service; +} + +/** One caller-owned Harness client with the adapter's private drain signal. */ +export interface ActiveHarnessClient { + readonly client: HarnessClientService; + readonly stopSignal: Deferred.Deferred; +} + +/** + * Stops every adapter binding for an account without closing a Harness client. + * @param activeClients Legacy services owned by the adapter. + * @param activeHarnessClients Harness drains owned by the adapter. + * @param accountId Account whose active binding is stopped. + * @returns A lazy stop operation for the selected account. + */ +export function stopActiveGatewayAccount< + Service extends ClosableGatewayService, +>( + activeClients: Map, + activeHarnessClients: Map, + accountId: string, +): Effect.Effect { + return Effect.gen(function* () { + const harness = activeHarnessClients.get(accountId); + if (harness !== undefined) { + activeHarnessClients.delete(accountId); + yield* Deferred.succeed(harness.stopSignal, undefined); + } + const service = activeClients.get(accountId); + if (service !== undefined) { + activeClients.delete(accountId); + yield* Effect.sync(() => { + service.close(); + }); + } + }).pipe(Effect.withSpan("stopActiveGatewayAccount")); +} + +/** + * Removes a completed drain only when it is still the active generation. + * @param activeHarnessClients Harness drains owned by the adapter. + * @param accountId Account whose drain completed. + * @param active Completed generation. + * @returns A lazy generation-checked removal. + */ +export function finishHarnessClient( + activeHarnessClients: Map, + accountId: string, + active: ActiveHarnessClient, +): Effect.Effect { + return Effect.sync(() => { + if (activeHarnessClients.get(accountId) === active) { + activeHarnessClients.delete(accountId); + } + }); +} + +const removeLegacyClientIfActive = ( + activeClients: Map, + accountId: string, + service: Service, +): Effect.Effect => + Effect.sync(() => { + if (activeClients.get(accountId) === service) { + activeClients.delete(accountId); + } + }); + +/** + * Disconnects one legacy generation without removing a newer replacement. + * @param binding Core and service belonging to one generation. + * @param activeClients Active legacy service by account. + * @param accountId Account owned by the generation. + * @returns A lazy disconnect operation. + */ +export function disconnectLegacyGateway( + binding: LegacyGatewayBinding, + activeClients: Map, + accountId: string, +): Effect.Effect { + return binding.core + .disconnect() + .pipe( + Effect.ensuring( + removeLegacyClientIfActive(activeClients, accountId, binding.service), + ), + ); +} + +/** + * Binds one legacy gateway generation to its account abort signal. + * @param signal Host-owned lifecycle signal. + * @param binding Core and service belonging to one generation. + * @param activeClients Active legacy service by account. + * @param accountId Account owned by the generation. + */ +export function registerLegacyGatewayAbort( + signal: AbortSignal, + binding: LegacyGatewayBinding, + activeClients: Map, + accountId: string, +): void { + signal.addEventListener( + "abort", + () => { + Effect.runFork( + disconnectLegacyGateway(binding, activeClients, accountId), + ); + }, + { once: true }, + ); +} diff --git a/packages/openclaw-channel/src/openclaw-target.ts b/packages/openclaw-channel/src/openclaw-target.ts new file mode 100644 index 000000000..a7895d18e --- /dev/null +++ b/packages/openclaw-channel/src/openclaw-target.ts @@ -0,0 +1,83 @@ +import { Schema } from "effect"; +import { agentName, type AgentName } from "@moltzap/protocol/identity"; + +/** Prefix used for named agent targets. */ +const TARGET_PREFIX_AGENT = "agent:"; + +/** Prefix used for existing conversation targets. */ +export const TARGET_PREFIX_CONVERSATION = "conv:"; + +/** User-facing description of the accepted target forms. */ +export const TARGET_HINT = + 'Use an agent name or "agent:" for DMs or "conv:" for conversations'; + +interface ResolvedAgentTarget { + readonly to: string; + readonly kind: "user"; + readonly display: AgentName; +} + +interface ResolvedConversationTarget { + readonly to: string; + readonly kind: "group"; + readonly display: string; +} + +/** Normalized target consumed by OpenClaw directory and outbound adapters. */ +export type ResolvedMoltZapTarget = + | ResolvedAgentTarget + | ResolvedConversationTarget; + +const isAgentName = Schema.is(agentName); + +function normalizeConversationTarget( + target: string, +): ResolvedConversationTarget | null | undefined { + if (!target.startsWith(TARGET_PREFIX_CONVERSATION)) { + return undefined; + } + const id = target.slice(TARGET_PREFIX_CONVERSATION.length); + return id.length === 0 || id.includes(":") + ? null + : { to: target, kind: "group", display: id }; +} + +function normalizeAgentTarget(target: string): ResolvedAgentTarget | null { + let name: string | null; + if (target.startsWith(TARGET_PREFIX_AGENT)) { + name = target.slice(TARGET_PREFIX_AGENT.length); + } else if (target.includes(":")) { + name = null; + } else { + name = target; + } + return name === null || !isAgentName(name) + ? null + : { to: `${TARGET_PREFIX_AGENT}${name}`, kind: "user", display: name }; +} + +/** + * Normalizes an OpenClaw target into a named agent or existing conversation. + * @param raw User-supplied target. + * @returns A normalized target, or null when the shape is unsupported. + */ +export function normalizeMoltZapTarget( + raw: string, +): ResolvedMoltZapTarget | null { + const target = raw.trim(); + const conversation = normalizeConversationTarget(target); + if (conversation !== undefined) { + return conversation; + } + return normalizeAgentTarget(target); +} + +/** + * Tests whether a target is already in canonical OpenClaw form. + * @param raw User-supplied target. + * @returns Whether the target is canonical and supported. + */ +export function isMoltZapTarget(raw: string): boolean { + const target = raw.trim(); + return normalizeMoltZapTarget(target)?.to === target; +} diff --git a/packages/protocol/scripts/docs/__tests__/typedoc-load.test.ts b/packages/protocol/scripts/docs/__tests__/typedoc-load.test.ts index 29559cd1d..985854ff4 100644 --- a/packages/protocol/scripts/docs/__tests__/typedoc-load.test.ts +++ b/packages/protocol/scripts/docs/__tests__/typedoc-load.test.ts @@ -8,7 +8,7 @@ import { describe, expect, it } from "vitest"; import { loadTypeDoc, normalizeSourcePath } from "../typedoc-load.js"; describe("normalizeSourcePath", () => { - it("retains both workspace source roots", () => { + it("retains workspace source and declaration roots", () => { expect( normalizeSourcePath( "/workspace/v2/moltzap/packages/protocol/src/index.ts", @@ -19,6 +19,11 @@ describe("normalizeSourcePath", () => { "C:\\workspace\\archive-v2\\moltzap\\v2\\identity\\src\\index.ts", ), ).toBe("v2/identity/src/index.ts"); + expect( + normalizeSourcePath( + "/workspace/moltzap-worktree/packages/protocol/dist/socket/agent-client.d.ts", + ), + ).toBe("packages/protocol/dist/socket/agent-client.d.ts"); }); it("recovers a v2 path from a TypeDoc permalink", () => { diff --git a/packages/protocol/scripts/docs/typedoc-load.ts b/packages/protocol/scripts/docs/typedoc-load.ts index 03f1bfd19..43c908880 100644 --- a/packages/protocol/scripts/docs/typedoc-load.ts +++ b/packages/protocol/scripts/docs/typedoc-load.ts @@ -376,7 +376,7 @@ function extractReturnTypeName(node: RawReflection): string | null { } /** - * Normalize a TypeDoc source path to a workspace source root. + * Normalize a TypeDoc source path to a workspace source or declaration root. * @param sourcePath TypeDoc's reported source path. * @param sourceUrl Optional source permalink emitted by TypeDoc. * @returns A workspace-relative source path when one can be recovered. @@ -404,7 +404,7 @@ export function normalizeSourcePath( } function findWorkspacePath(sourcePath: string): string | null { - const match = /(?:^|\/)((?:packages|v2)\/[^/]+\/src(?:\/.*)?$)/.exec( + const match = /(?:^|\/)((?:packages|v2)\/[^/]+\/(?:src|dist)(?:\/.*)?$)/.exec( sourcePath, ); return match?.[1] ?? null;