fix(mastracode): coordinate signals over Unix socket PubSub (#16669)

This adds a Unix socket PubSub implementation and wires Mastra Code to
use it for local cross-process signal/stream coordination on macOS/Linux
so that you never see the "thread is locked" message when starting
multiple mc instances in the same dir.

Used MC to test this but I need unix pubsub to make it easier to test
drucker agents.

Mastra Code now defaults to a local socket PubSub per project resource
id, skips file thread locks when that cross-process PubSub is active,
and closes the socket PubSub during CLI cleanup. Windows keeps the
existing in-process PubSub/thread-lock behavior for now.

After:

```ts
const signalsPubSub =
  configuredPubSub ?? (process.platform === 'win32' ? undefined : createSignalsPubSub(project.resourceId));
```

I tested this with core signal/runtime tests, Unix socket PubSub tests,
and the Mastra Code build/typecheck.


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
ELI5

This PR makes Mastra Code on macOS/Linux use a per-project local
Unix-domain socket so multiple local mastracode processes can share
signals and live-stream updates without getting blocked by file-based
thread locks; Windows behavior is unchanged. It also ensures the socket
PubSub is closed cleanly on shutdown and preserves configured pubsub
behavior unless cross-process mode is enabled.

---

Overview

Adds a Unix-domain socket PubSub transport and wires Mastra Code to use
a deterministic per-resource Unix socket (derived from
project.resourceId) for local cross-process signal and stream
coordination on macOS/Linux. The Mastra Code TUI enables the Unix-socket
PubSub by default when not on Windows (createMastraCode still honors an
explicit configured pubsub), cross-process mode disables file
thread-locks only when active, and shutdown paths close the socket
PubSub. The UnixSocketPubSub implementation handles broker/client roles,
reconnection, stale-socket reclamation, resubscription on reconnect,
rejecting pending subscribe waiters on broker failure, and robust
close/unlink behavior.

---

Key changes

- New IPC transport
  - packages/core/src/events/unix-socket-pubsub.ts
- Implements UnixSocketPubSub: newline-delimited JSON over Unix domain
sockets with broker/client semantics, write batching/flush,
resubscribe-on-reconnect, subscribe waiters, and safe close/unlink.
- Public API: socketPath, supportedModes, isBroker, publish, subscribe,
unsubscribe, flush, close.

- Mastra Code integration
  - mastracode/src/index.ts
    - MastraCodeConfig.unixSocketPubSub?: boolean added.
- createMastraCode derives signalsPubSub from config.pubsub or
(non-Windows) a per-resource Unix-socket PubSub when enabled;
crossProcessPubSub defaults to the Unix-socket choice only if no
configured pubsub is provided. threadLock remains conditional on
crossProcessPubSub.
  - mastracode/src/utils/signals-pubsub.ts
- getSignalsPubSubSocketPath(resourceId): deterministic SHA-256–derived
short socket path under app data signals dir.
- createSignalsPubSub(resourceId) factory that returns a
UnixSocketPubSub bound to that path.
  - mastracode/src/main.ts and mastracode/src/headless.ts
- TUI entry initializes createMastraCode({ unixSocketPubSub: true }),
stores returned signalsPubSub, and includes signalsPubSub.close() in
async cleanup; headless cleanup also closes optional signalsPubSub.
  - mastracode/src/onboarding/settings.ts
- Introduces SignalSettings and adds
GlobalSettings.signals.unixSocketPubSub (default false) with
migration/load handling.

- Tests
  - packages/core/src/events/unix-socket-pubsub.test.ts
- Expanded Vitest suite: fan-out, subscriber exception isolation, broker
promotion after close, stale-socket reclamation, and subscribe rejection
when broker disconnects.
  - packages/core/src/agent/__tests__/agent-signals.test.ts
- Non-Windows integration test: two AgentThreadStreamRuntime instances
using UnixSocketPubSub on the same socket both receive streamed text
deltas.
  - mastracode/src/__tests__/index.test.ts
- Tests cover createMastraCode behavior for provided pubsub vs
crossProcessPubSub and expected threadLock settings.
  - Unix-socket tests are skipped on Windows.

- Exports and release notes
  - packages/core/src/events/index.ts re-exports unix-socket-pubsub.
- .changeset/unix-socket-signals-pubsub.md added (patch for
`@mastra/core` and mastracode).

---

Platform behavior

- macOS/Linux: TUI enables per-resource Unix socket PubSub by default
(unless overridden) to allow cross-process coordination and avoid
file-based thread locks.
- Windows: unchanged — continues using in-process PubSub and file-based
thread-lock behavior; Unix-socket tests are skipped.

---

Testing performed

- Unit tests for UnixSocketPubSub covering normal and edge cases (broker
disconnect, stale socket, broker promotion, subscriber failures).
- Integration test validating cross-runtime signal/stream delivery
between runtimes using the same socket.
- Mastra Code build/typecheck and core signal/runtime tests run
successfully with the changeset.

<!-- review_stack_entry_start -->

[![Review Change
Stack](https://storage.googleapis.com/coderabbit_public_assets/review-stack-in-coderabbit-ui.svg)](https://app.coderabbit.ai/change-stack/mastra-ai/mastra/pull/16669?utm_source=github_walkthrough&utm_medium=github&utm_campaign=change_stack)

<!-- review_stack_entry_end -->
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: Mastra Code (openai/gpt-5.5) <noreply@mastra.ai>

Author: TylerBarnes

.changeset/unix-socket-signals-pubsub.md

@@ -0,0 +1,10 @@
+---
+'@mastra/core': patch
+'mastracode': patch
+---
+
+Added a Unix socket PubSub transport and wired the Mastra Code TUI through a per-resource socket so local sessions can coordinate thread streams across processes. Programmatic `createMastraCode` usage remains opt-in:
+
+```ts
+await createMastraCode({ unixSocketPubSub: true });
+```

mastracode/src/__tests__/index.test.ts

@@ -324,6 +324,32 @@ describe('createMastraCode', () => {
     );
   });
 
+  it('keeps thread locks enabled for configured PubSub unless cross-process mode is explicit', async () => {
+    const pubsub = {} as any;
+    const { createMastraCode } = await import('../index.js');
+
+    await createMastraCode({ pubsub, unixSocketPubSub: true });
+
+    const harnessConfig = harnessConstructorMock.mock.calls.at(-1)?.[0] as
+      | { pubsub?: unknown; threadLock?: unknown }
+      | undefined;
+    expect(harnessConfig?.pubsub).toBe(pubsub);
+    expect(harnessConfig?.threadLock).toBeDefined();
+  });
+
+  it('skips thread locks for configured PubSub when cross-process mode is explicit', async () => {
+    const pubsub = {} as any;
+    const { createMastraCode } = await import('../index.js');
+
+    await createMastraCode({ pubsub, crossProcessPubSub: true });
+
+    const harnessConfig = harnessConstructorMock.mock.calls.at(-1)?.[0] as
+      | { pubsub?: unknown; threadLock?: unknown }
+      | undefined;
+    expect(harnessConfig?.pubsub).toBe(pubsub);
+    expect(harnessConfig?.threadLock).toBeUndefined();
+  });
+
   it('restores the current thread caveman observation setting at startup', async () => {
     harnessGetCurrentThreadIdMock.mockReturnValue('thread-1');
     harnessListThreadsMock.mockResolvedValue([{ id: 'thread-1', metadata: { cavemanObservations: true } }]);

mastracode/src/headless.ts

@@ -628,7 +628,8 @@ export async function headlessMain(predrainedInput?: string | null): Promise<nev
 
   // Cleanup
   releaseAllThreadLocks();
-  await Promise.allSettled([mcpManager?.disconnect(), harness?.stopHeartbeats()]);
+  const closeSignalsPubSub = (result.signalsPubSub as { close?: () => Promise<void> | void } | undefined)?.close;
+  await Promise.allSettled([mcpManager?.disconnect(), harness?.stopHeartbeats(), closeSignalsPubSub?.()]);
 
   process.exit(exitCode);
 }

mastracode/src/index.ts

@@ -73,6 +73,7 @@ import {
   getResourceIdOverride,
 } from './utils/project.js';
 import type { StorageConfig } from './utils/project.js';
+import { createSignalsPubSub } from './utils/signals-pubsub.js';
 import { createStorage, createVectorStore } from './utils/storage-factory.js';
 import { acquireThreadLock, releaseThreadLock } from './utils/thread-lock.js';
 
@@ -135,6 +136,8 @@ export interface MastraCodeConfig {
   browser?: HarnessConfig['browser'];
   /** PubSub for signal routing. When crossProcessPubSub is true, thread locks are disabled. */
   pubsub?: PubSub;
+  /** Use Mastra Code's built-in Unix socket PubSub for local cross-process signal routing. */
+  unixSocketPubSub?: boolean;
   /** Marks the configured PubSub as cross-process-safe, allowing Mastra Code to skip file thread locks. */
   crossProcessPubSub?: boolean;
 }
@@ -232,8 +235,11 @@ export async function createMastraCode(config?: MastraCodeConfig) {
     project.resourceIdOverride = true;
   }
 
-  const signalsPubSub = config?.pubsub;
-  const crossProcessPubSub = config?.crossProcessPubSub ?? false;
+  const configuredPubSub = config?.pubsub;
+  const useUnixSocketPubSub =
+    (config?.unixSocketPubSub ?? globalSettings.signals?.unixSocketPubSub ?? false) && process.platform !== 'win32';
+  const signalsPubSub = configuredPubSub ?? (useUnixSocketPubSub ? createSignalsPubSub(project.resourceId) : undefined);
+  const crossProcessPubSub = config?.crossProcessPubSub ?? (!configuredPubSub && useUnixSocketPubSub);
   if (crossProcessPubSub && !signalsPubSub) {
     throw new Error('crossProcessPubSub requires a pubsub instance');
   }
@@ -675,11 +681,11 @@ export async function createMastraCode(config?: MastraCodeConfig) {
     harness,
     mcpManager,
     hookManager,
+    signalsPubSub,
     authStorage,
     resolveModel,
     storageWarning,
     observabilityWarning,
-    signalsPubSub,
     builtinPacks,
     builtinOmPacks,
     effectiveDefaults,

mastracode/src/main.ts

@@ -21,6 +21,7 @@ let harness: Awaited<ReturnType<typeof createMastraCode>>['harness'];
 let mcpManager: Awaited<ReturnType<typeof createMastraCode>>['mcpManager'];
 let hookManager: Awaited<ReturnType<typeof createMastraCode>>['hookManager'];
 let authStorage: Awaited<ReturnType<typeof createMastraCode>>['authStorage'];
+let signalsPubSub: Awaited<ReturnType<typeof createMastraCode>>['signalsPubSub'];
 let analytics: ReturnType<typeof createMastraCodeAnalytics> | undefined;
 
 // Global safety nets — catch any uncaught errors from storage init, etc.
@@ -43,11 +44,12 @@ async function tuiMain(pipedInput?: string | null) {
     return browserPromise;
   };
 
-  const result = await createMastraCode();
+  const result = await createMastraCode({ unixSocketPubSub: true });
   harness = result.harness;
   mcpManager = result.mcpManager;
   hookManager = result.hookManager;
   authStorage = result.authStorage;
+  signalsPubSub = result.signalsPubSub;
 
   if (result.storageWarning) {
     console.info(`⚠ ${result.storageWarning}`);
@@ -119,7 +121,13 @@ async function tuiMain(pipedInput?: string | null) {
 
 const asyncCleanup = async () => {
   releaseAllThreadLocks();
-  await Promise.allSettled([mcpManager?.disconnect(), harness?.stopHeartbeats(), analytics?.shutdown()]);
+  const closeSignalsPubSub = (signalsPubSub as { close?: () => Promise<void> | void } | undefined)?.close;
+  await Promise.allSettled([
+    mcpManager?.disconnect(),
+    harness?.stopHeartbeats(),
+    closeSignalsPubSub?.(),
+    analytics?.shutdown(),
+  ]);
 };
 
 process.on('beforeExit', () => {

mastracode/src/onboarding/settings.ts

@@ -207,10 +207,17 @@ export interface GlobalSettings {
   lsp?: LSPConfig;
   // Browser automation configuration
   browser: BrowserSettings;
+  // Signal routing configuration
+  signals: SignalSettings;
   // Cloud observability configuration (per-resource project IDs; tokens stored in auth.json)
   observability: ObservabilitySettings;
 }
 
+export interface SignalSettings {
+  /** Opt into local Unix socket PubSub for cross-process signal routing. */
+  unixSocketPubSub: boolean;
+}
+
 export interface ObservabilityResourceConfig {
   /** Cloud project ID for this resource */
   projectId: string;
@@ -279,6 +286,7 @@ const DEFAULTS: GlobalSettings = {
     viewport: { width: 1280, height: 720 },
     stagehand: { env: 'LOCAL' },
   },
+  signals: { unixSocketPubSub: false },
   observability: { resources: {}, localTracing: false },
 };
 
@@ -513,6 +521,12 @@ function migrateFromAuth(settingsPath: string): boolean {
         memoryGateway: raw.memoryGateway && typeof raw.memoryGateway === 'object' ? raw.memoryGateway : {},
         lsp: raw.lsp && typeof raw.lsp === 'object' ? (raw.lsp as LSPConfig) : undefined,
         browser: parseBrowserSettings(raw.browser),
+        signals: {
+          unixSocketPubSub:
+            raw.signals && typeof raw.signals === 'object' && typeof raw.signals.unixSocketPubSub === 'boolean'
+              ? raw.signals.unixSocketPubSub
+              : DEFAULTS.signals.unixSocketPubSub,
+        },
         observability: parseObservabilitySettings(raw.observability),
       };
       applyQuietModePreferenceRollout(settings, raw.onboarding);
@@ -633,6 +647,12 @@ export function loadSettings(filePath: string = getSettingsPath()): GlobalSettin
       memoryGateway: raw.memoryGateway && typeof raw.memoryGateway === 'object' ? raw.memoryGateway : {},
       lsp: raw.lsp && typeof raw.lsp === 'object' ? (raw.lsp as LSPConfig) : undefined,
       browser: parseBrowserSettings(raw.browser),
+      signals: {
+        unixSocketPubSub:
+          raw.signals && typeof raw.signals === 'object' && typeof raw.signals.unixSocketPubSub === 'boolean'
+            ? raw.signals.unixSocketPubSub
+            : DEFAULTS.signals.unixSocketPubSub,
+      },
       observability: parseObservabilitySettings(raw.observability),
     };
 

mastracode/src/utils/signals-pubsub.ts

@@ -0,0 +1,27 @@
+import { createHash } from 'node:crypto';
+import * as fs from 'node:fs';
+import * as path from 'node:path';
+
+import { UnixSocketPubSub } from '@mastra/core/events';
+
+import { getAppDataDir } from './project.js';
+
+function getSignalsDir(): string {
+  const dir = path.join(getAppDataDir(), 'signals');
+  if (!fs.existsSync(dir)) {
+    fs.mkdirSync(dir, { recursive: true });
+  }
+  return dir;
+}
+
+function shortHash(value: string): string {
+  return createHash('sha256').update(value).digest('hex').slice(0, 16);
+}
+
+export function getSignalsPubSubSocketPath(resourceId: string): string {
+  return path.join(getSignalsDir(), `${shortHash(resourceId)}.sock`);
+}
+
+export function createSignalsPubSub(resourceId: string): UnixSocketPubSub {
+  return new UnixSocketPubSub(getSignalsPubSubSocketPath(resourceId));
+}

packages/core/src/agent/__tests__/agent-signals.test.ts

@@ -1,7 +1,12 @@
+import { mkdtemp, rm } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+
 import { MockLanguageModelV2, convertArrayToReadableStream } from '@internal/ai-sdk-v5/test';
 import { beforeEach, describe, expect, it, vi } from 'vitest';
 
 import { EventEmitterPubSub } from '../../events/event-emitter';
+import { UnixSocketPubSub } from '../../events/unix-socket-pubsub';
 import { Mastra } from '../../mastra';
 import { MockMemory } from '../../memory/mock';
 import { Agent } from '../agent';
@@ -884,6 +889,70 @@ describe('Agent signals', () => {
     senderSubscription.unsubscribe();
   });
 
+  it.runIf(process.platform !== 'win32')(
+    'broadcasts subscribed thread stream parts across UnixSocketPubSub runtime instances',
+    async () => {
+      const tempDir = await mkdtemp(join(tmpdir(), 'mastra-agent-signals-'));
+      const ownerPubSub = new UnixSocketPubSub(join(tempDir, 'signals.sock'));
+      const followerPubSub = new UnixSocketPubSub(join(tempDir, 'signals.sock'));
+      const ownerRuntime = new AgentThreadStreamRuntime();
+      const followerRuntime = new AgentThreadStreamRuntime();
+      const owner = new Agent({
+        id: 'unix-stream-agent',
+        name: 'Unix Stream Owner Agent',
+        instructions: 'Test',
+        model: createTextStreamModel('owner response'),
+      });
+      const follower = new Agent({
+        id: 'unix-stream-agent',
+        name: 'Unix Stream Follower Agent',
+        instructions: 'Test',
+        model: createTextStreamModel('follower response'),
+      });
+      let finishRun!: () => void;
+      const output = {
+        runId: 'unix-run-1',
+        status: 'running',
+        fullStream: (async function* () {
+          yield { type: 'text-delta', runId: 'unix-run-1', payload: { text: 'hello over uds' } };
+          yield { type: 'finish', runId: 'unix-run-1', payload: {} };
+        })(),
+        _waitUntilFinished: () => new Promise<void>(resolve => (finishRun = resolve)),
+      } as any;
+
+      try {
+        const ownerSubscription = await ownerRuntime.subscribeToThread(
+          owner,
+          { resourceId: 'unix-resource', threadId: 'unix-thread' },
+          ownerPubSub,
+        );
+        const followerSubscription = await followerRuntime.subscribeToThread(
+          follower,
+          { resourceId: 'unix-resource', threadId: 'unix-thread' },
+          followerPubSub,
+        );
+        const ownerRun = readNextRunWithParts(ownerSubscription.stream[Symbol.asyncIterator]());
+        const followerRun = readNextRunWithParts(followerSubscription.stream[Symbol.asyncIterator]());
+
+        ownerRuntime.registerRun(
+          owner,
+          output,
+          { runId: 'unix-run-1', memory: { resource: 'unix-resource', thread: 'unix-thread' } } as any,
+          ownerPubSub,
+        );
+
+        await expect(ownerRun).resolves.toMatchObject({ value: { text: 'hello over uds' }, done: false });
+        await expect(followerRun).resolves.toMatchObject({ value: { text: 'hello over uds' }, done: false });
+        finishRun();
+        ownerSubscription.unsubscribe();
+        followerSubscription.unsubscribe();
+      } finally {
+        await Promise.allSettled([ownerPubSub.close(), followerPubSub.close()]);
+        await rm(tempDir, { recursive: true, force: true });
+      }
+    },
+  );
+
   it('supports cross-instance thread subscriptions through an injected PubSub without Mastra', async () => {
     const pubsub = new EventEmitterPubSub();
     const runner = new Agent({

packages/core/src/events/index.ts

@@ -2,3 +2,4 @@ export * from './types';
 export * from './pubsub';
 export * from './event-emitter';
 export { CachingPubSub, withCaching, type CachingPubSubOptions } from './caching-pubsub';
+export * from './unix-socket-pubsub';