feat(sdk): make the published event source configurable (#131)

## Problem

The SDK transports hardcoded the `source` field when publishing user
prompts to the Axon API:

- `sdk/src/claude/transport.ts` → `"claude-sdk-client"`
- `sdk/src/acp/axon-stream.ts` → `"acp-sdk-client"`

Callers had no way to override it, so events from different integrations
writing to the same channel were indistinguishable.

## Approach

Rather than threading `source` through every `send()` / `prompt()` call
(the ACP `prompt()` path delegates to the ACP SDK's
`ClientSideConnection` and has no clean per-message hook), the source is
now a value the **client holds and can change at any time**. The
transport/stream resolves it **lazily at each publish**, so:

```ts
conn.setSource("my-integration");
await conn.send("hello");      // published with source "my-integration"
conn.setSource("other");
await conn.prompt({ ... });    // published with source "other"
```

This works uniformly for both Claude and ACP without touching the SDK
send methods.

## Changes

- **`BaseConnectionOptions.source`** — optional initial `source` for
both connection types.
- **`ClaudeAxonConnection` / `ACPAxonConnection`** — new
`setSource(source)` method and `source` getter; hold a mutable
`currentSource` and pass a resolver (`() => this.currentSource`) to the
transport/stream.
- **`AxonTransportOptions.source` / `AxonStreamOptions.source`** —
accept `string | (() => string | undefined)`, resolved at publish time.
- Defaults unchanged: omit `source` (or return `undefined`) and the
built-in identifiers are still used.

## Tests

Added coverage for: custom static source, lazy per-message resolution,
`undefined` → default fallback (transport + stream), and the
`setSource()` / `source` accessors on both connections.

- `npm run typecheck`, `npm run check` (biome), and `vitest` (512 tests)
all pass.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Reflex <reflex@runloop.ai>
Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: 3 people

sdk/src/acp/axon-stream.test.ts

@@ -764,6 +764,50 @@ describe("axonStream", () => {
       expect(published[0].source).toBe("acp-sdk-client");
     });
 
+    it("publishes with a custom source when provided via options", async () => {
+      const { writable } = axonStream({ axon: axon as never, source: "my-custom-client" });
+
+      await writeMessage(writable, {
+        jsonrpc: "2.0",
+        method: "session/cancel",
+        params: { sessionId: "s1" },
+      });
+
+      expect(published[0].source).toBe("my-custom-client");
+    });
+
+    it("resolves the source lazily on each publish when given a resolver", async () => {
+      let current = "first";
+      const { writable } = axonStream({ axon: axon as never, source: () => current });
+
+      await writeMessage(writable, {
+        jsonrpc: "2.0",
+        method: "session/cancel",
+        params: { sessionId: "s1" },
+      });
+      current = "second";
+      await writeMessage(writable, {
+        jsonrpc: "2.0",
+        method: "session/cancel",
+        params: { sessionId: "s2" },
+      });
+
+      expect(published[0].source).toBe("first");
+      expect(published[1].source).toBe("second");
+    });
+
+    it("falls back to the default when the resolver returns undefined", async () => {
+      const { writable } = axonStream({ axon: axon as never, source: () => undefined });
+
+      await writeMessage(writable, {
+        jsonrpc: "2.0",
+        method: "session/cancel",
+        params: { sessionId: "s1" },
+      });
+
+      expect(published[0].source).toBe("acp-sdk-client");
+    });
+
     it("publishes JSON-RPC notifications with method as event_type", async () => {
       const { writable } = axonStream({ axon: axon as never });
 

sdk/src/acp/axon-stream.ts

@@ -21,6 +21,9 @@ const NOTIFICATION_TYPES = new Set<string>([
   CLIENT_METHODS.session_elicitation_complete,
 ]);
 
+/** Default `source` used when publishing events from the ACP SDK stream. */
+const DEFAULT_SOURCE = "acp-sdk-client";
+
 /**
  * Creates an ACP-compatible `Stream` backed by an Axon channel from
  * `@runloop/api-client`.
@@ -42,6 +45,13 @@ const NOTIFICATION_TYPES = new Set<string>([
 export function axonStream(options: AxonStreamOptions): Stream {
   const { axon, signal, onAxonEvent, log, afterSequence, replayTargetSequence } = options;
   const onError = options.onError ?? makeDefaultOnError("axonStream");
+  const optionSource = options.source;
+  // Resolve lazily on each publish so callers can change the source between
+  // messages (e.g. via a resolver closing over mutable state).
+  const resolveSource = (): string => {
+    const value = typeof optionSource === "function" ? optionSource() : optionSource;
+    return value ?? DEFAULT_SOURCE;
+  };
 
   // Maps outbound JSON-RPC request method -> id so we can correlate
   // the broker's response (which only carries event_type, not an id).
@@ -66,7 +76,14 @@ export function axonStream(options: AxonStreamOptions): Stream {
     replayTargetSequence,
   );
 
-  const writable = createWritable(axon, pendingRequests, pendingClientRequests, onError, log);
+  const writable = createWritable(
+    axon,
+    pendingRequests,
+    pendingClientRequests,
+    onError,
+    log,
+    resolveSource,
+  );
 
   return { readable, writable };
 }
@@ -447,6 +464,7 @@ function axonEventToJsonRpc(
  * @param axon                 - Axon channel to publish to.
  * @param pendingRequests      - Shared map tracking outbound request method → JSON-RPC ID.
  * @param pendingClientRequests - Shared map tracking agent-to-client request ID → method.
+ * @param resolveSource         - Resolver invoked per publish to obtain the `source`.
  * @returns A `WritableStream` that accepts JSON-RPC messages.
  */
 function createWritable(
@@ -455,6 +473,7 @@ function createWritable(
   pendingClientRequests: Map<string | number, string>,
   onError: (error: unknown) => void,
   log: LogFn | undefined,
+  resolveSource: () => string,
 ): WritableStream<AnyMessage> {
   return new WritableStream<AnyMessage>({
     async write(message) {
@@ -465,7 +484,7 @@ function createWritable(
           event_type: eventType,
           origin: "USER_EVENT",
           payload,
-          source: "acp-sdk-client",
+          source: resolveSource(),
         });
       } catch (err) {
         onError(err);

sdk/src/acp/connection.test.ts

@@ -81,6 +81,40 @@ describe("ACPAxonConnection", () => {
     });
   });
 
+  describe("source / setSource()", () => {
+    it("defaults to undefined (stream supplies the built-in default)", () => {
+      const ctrl = createControllableStream();
+      const { axon } = createMockAxon(ctrl);
+
+      const conn = new ACPAxonConnection(axon as never, { id: "dbx-test" } as never);
+      expect(conn.source).toBeUndefined();
+      conn.disconnect();
+    });
+
+    it("reflects the value passed via options", () => {
+      const ctrl = createControllableStream();
+      const { axon } = createMockAxon(ctrl);
+
+      const conn = new ACPAxonConnection(axon as never, { id: "dbx-test" } as never, {
+        source: "from-options",
+      });
+      expect(conn.source).toBe("from-options");
+      conn.disconnect();
+    });
+
+    it("updates the current source via setSource()", () => {
+      const ctrl = createControllableStream();
+      const { axon } = createMockAxon(ctrl);
+
+      const conn = new ACPAxonConnection(axon as never, { id: "dbx-test" } as never);
+      conn.setSource("my-client");
+      expect(conn.source).toBe("my-client");
+      conn.setSource(undefined);
+      expect(conn.source).toBeUndefined();
+      conn.disconnect();
+    });
+  });
+
   describe("default permission handler", () => {
     it("prefers allow_always when available", async () => {
       const ctrl = createControllableStream();

sdk/src/acp/connection.ts

@@ -136,6 +136,14 @@ export class ACPAxonConnection {
   /** Optional teardown callback invoked by {@link disconnect}. */
   private disconnectFn: (() => void | Promise<void>) | undefined;
 
+  /**
+   * The `source` attached to outbound events. Read by the stream at publish
+   * time, so mutating it via {@link setSource} affects subsequent messages
+   * without reconnecting. `undefined` means the stream's built-in default
+   * is used.
+   */
+  private currentSource: string | undefined;
+
   private log: LogFn;
 
   /**
@@ -159,6 +167,7 @@ export class ACPAxonConnection {
     this.handleError = options?.onError ?? makeDefaultOnError("ACPAxonConnection");
     this.handlePermission = options?.requestPermission;
     this.disconnectFn = options?.onDisconnect;
+    this.currentSource = options?.source;
     this.axonEventListeners = new ListenerSet<AxonEventListener>(this.handleError);
     this.timelineEventListeners = new ListenerSet<TimelineEventListener<ACPTimelineEvent>>(
       this.handleError,
@@ -200,6 +209,7 @@ export class ACPAxonConnection {
       log: verbose ? (tag, ...args) => this.log(tag, ...args) : undefined,
       afterSequence: this.options.afterSequence,
       replayTargetSequence,
+      source: () => this.currentSource,
     });
 
     const customCreateClient = this.options.createClient;
@@ -291,6 +301,34 @@ export class ACPAxonConnection {
     return this.protocol.prompt(params);
   }
 
+  /**
+   * The `source` string currently attached to events published by this
+   * connection (e.g. prompts via {@link prompt}). `undefined` means the
+   * built-in default (`"acp-sdk-client"`) is used.
+   */
+  get source(): string | undefined {
+    return this.currentSource;
+  }
+
+  /**
+   * Sets the `source` string attached to subsequently published events.
+   *
+   * Takes effect immediately and applies to every later message published
+   * through the stream (such as {@link prompt}) until changed again — set it
+   * before a message to control that message's `source`. Pass `undefined` to
+   * restore the built-in default.
+   *
+   * Because prompts share one underlying stream, set the source before
+   * issuing a prompt and avoid overlapping prompts that need different
+   * sources — concurrent in-flight messages observe whichever value is
+   * current when each is published.
+   *
+   * @param source - The source identifier, or `undefined` for the default.
+   */
+  setSource(source: string | undefined): void {
+    this.currentSource = source;
+  }
+
   /**
    * Cancels an ongoing prompt turn for a session.
    *

sdk/src/acp/types.ts

@@ -54,6 +54,17 @@ export interface AxonStreamOptions {
    * enqueued. After replay ends, only unresolved requests are enqueued.
    */
   replayTargetSequence?: number;
+
+  /**
+   * The `source` string attached to every published Axon event, or a
+   * resolver invoked at publish time to obtain it. Use the resolver form to
+   * change the `source` between messages without recreating the stream
+   * (e.g. `() => this.currentSource`). When omitted, or when the resolver
+   * returns `undefined`, the default is used.
+   *
+   * @defaultValue `"acp-sdk-client"`
+   */
+  source?: string | (() => string | undefined);
 }
 
 /**

sdk/src/claude/connection.test.ts

@@ -260,6 +260,30 @@ describe("ClaudeAxonConnection", () => {
     });
   });
 
+  describe("source / setSource()", () => {
+    it("defaults to undefined (transport supplies the built-in default)", async () => {
+      const conn = await createConnectedClient(transport);
+      expect(conn.source).toBeUndefined();
+    });
+
+    it("reflects the value passed via options", async () => {
+      const axon = createMockAxon();
+      const conn = new ClaudeAxonConnection(axon as never, { id: "dbx-test" } as never, {
+        source: "from-options",
+        replay: false,
+      });
+      expect(conn.source).toBe("from-options");
+    });
+
+    it("updates the current source via setSource()", async () => {
+      const conn = await createConnectedClient(transport);
+      conn.setSource("my-client");
+      expect(conn.source).toBe("my-client");
+      conn.setSource(undefined);
+      expect(conn.source).toBeUndefined();
+    });
+  });
+
   describe("publish()", () => {
     it("delegates to axon.publish() with the provided params", async () => {
       const conn = await createConnectedClient(transport);

sdk/src/claude/connection.ts

@@ -205,6 +205,14 @@ export class ClaudeAxonConnection {
   /** Optional teardown callback invoked by {@link disconnect}. */
   private disconnectFn: (() => void | Promise<void>) | undefined;
 
+  /**
+   * The `source` attached to outbound events. Read by the transport at
+   * publish time, so mutating it via {@link setSource} affects subsequent
+   * messages without recreating the transport. `undefined` means the
+   * transport's built-in default is used.
+   */
+  private currentSource: string | undefined;
+
   private static readonly MESSAGE_QUEUE_HIGH_WATER_MARK = 1000;
 
   /** Buffer of SDK messages that arrived before a consumer called {@link nextMessage}. */
@@ -282,6 +290,7 @@ export class ClaudeAxonConnection {
     this.options = options ?? {};
     this.handleError = options?.onError ?? makeDefaultOnError("ClaudeAxonConnection");
     this.disconnectFn = options?.onDisconnect;
+    this.currentSource = options?.source;
     this.log = makeLogger("claude-sdk", options?.verbose ?? false);
     this.axonEventListeners = new ListenerSet<AxonEventListener>(this.handleError);
     this.timelineEventListeners = new ListenerSet<TimelineEventListener<ClaudeTimelineEvent>>(
@@ -335,6 +344,7 @@ export class ClaudeAxonConnection {
         },
         afterSequence: this.options.afterSequence,
         replayTargetSequence,
+        source: () => this.currentSource,
       });
     }
 
@@ -888,6 +898,30 @@ export class ClaudeAxonConnection {
    * await conn.send("What files are in this directory?");
    * ```
    */
+  /**
+   * The `source` string currently attached to events published by this
+   * connection (e.g. user prompts via {@link send}). `undefined` means the
+   * built-in default (`"claude-sdk-client"`) is used.
+   */
+  get source(): string | undefined {
+    return this.currentSource;
+  }
+
+  /**
+   * Sets the `source` string attached to subsequently published events.
+   *
+   * Takes effect immediately and applies to every later message published
+   * through the transport (such as {@link send}) until changed again — set
+   * it before a message to control that message's `source`. Pass `undefined`
+   * to restore the built-in default. (The lower-level {@link publish} method
+   * is unaffected; it sets `source` directly from its params.)
+   *
+   * @param source - The source identifier, or `undefined` for the default.
+   */
+  setSource(source: string | undefined): void {
+    this.currentSource = source;
+  }
+
   async send(prompt: string | SDKUserMessage): Promise<void> {
     if (this.fatal) {
       throw new ConnectionStateError(

sdk/src/claude/transport.test.ts

@@ -117,6 +117,35 @@ describe("AxonTransport", () => {
       });
     });
 
+    it("uses a custom source when provided via options", async () => {
+      const custom = new AxonTransport(axon as never, { source: "my-custom-client" });
+      await custom.connect();
+      await custom.write(JSON.stringify({ type: "user" }));
+
+      expect(axon.publish.mock.calls[0][0].source).toBe("my-custom-client");
+    });
+
+    it("resolves the source lazily on each write when given a resolver", async () => {
+      let current = "first";
+      const custom = new AxonTransport(axon as never, { source: () => current });
+      await custom.connect();
+
+      await custom.write(JSON.stringify({ type: "user" }));
+      current = "second";
+      await custom.write(JSON.stringify({ type: "user" }));
+
+      expect(axon.publish.mock.calls[0][0].source).toBe("first");
+      expect(axon.publish.mock.calls[1][0].source).toBe("second");
+    });
+
+    it("falls back to the default when the resolver returns undefined", async () => {
+      const custom = new AxonTransport(axon as never, { source: () => undefined });
+      await custom.connect();
+      await custom.write(JSON.stringify({ type: "user" }));
+
+      expect(axon.publish.mock.calls[0][0].source).toBe("claude-sdk-client");
+    });
+
     it("uses the type value itself when not in mapping table", async () => {
       await transport.connect();
       await transport.write(JSON.stringify({ type: "custom_type" }));