feat(sdk): machine session bus — one wallet per machine, v2 always on (#256)

* feat(sdk): machine session bus — one wallet per machine, v2 always on

- Wrapper sessions on one machine now share a single tiny.place wallet
  safely: a cross-process wallet lock serializes every Signal operation
  (send, destructive inbox read, prekey publish) and a filesystem spool
  routes inbound DMs to their addressed session.
- New tinyplace.harness.control.v1 frame lets an owner address input to
  a specific session by wrapper or harness session id; plaintext DMs
  keep their meaning (injected into the machine's primary session).
- Machine session registry (<config-dir>/harness-bus/sessions) records
  each wrapper session with PID-probed live/inactive state and the
  harness's own session id (e.g. the codex rollout id) once discovered.
- v2 typed-event envelopes are now ALWAYS emitted alongside v1; the
  TINYPLACE_*_V2 opt-in env flag is retired.

Claude-Session: https://claude.ai/code/session_013xjQ7FUumJaRzS9Hv6B5yu

* fix(sdk): wire the machine session bus into the interactive TUI wrapper

The default `tinyplace codex`/`claude` path builds its bridge in
tui.ts (not runHarnessCommand), so it never engaged the bus: no wallet
lock, no spool routing, no session registry entry. startBridge now
creates/registers the bus, threads it into the publisher and receiver,
heartbeats it, and stopBridge marks the session ended.

Claude-Session: https://claude.ai/code/session_013xjQ7FUumJaRzS9Hv6B5yu

* fix(sdk): make FileSessionStore coherent across processes sharing one wallet

load() cached the whole store on first read and never re-read, so two
wrapper sessions on one machine wallet held divergent ratchet copies
and last-writer-wins clobbered each other's chains — the owner then
failed to decrypt every interleaved send. The cache is now validated
against the file's stat (mtime+size) on every load and refreshed after
each flush, so under the machine bus lock every operation starts from
the sibling's latest state.

Claude-Session: https://claude.ai/code/session_013xjQ7FUumJaRzS9Hv6B5yu

Author: senamakel

sdk/typescript/src/cli/harness-control.ts

@@ -0,0 +1,77 @@
+// Owner → machine control frames for the harness session bus.
+//
+// A machine runs ONE tiny.place wallet shared by every wrapped harness session
+// on it (see machine-bus.ts). Plaintext DMs keep their historical meaning —
+// they are injected into the machine's primary session — while a control frame
+// lets the owner address a SPECIFIC session by id. The session id is the one
+// the machine advertises on its envelope scope: the harness's own session id
+// (`scope.harness_session_id`, e.g. the codex rollout id) or the wrapper
+// session id (`scope.wrapper_session_id`) — a frame may name either.
+
+export const HARNESS_CONTROL_VERSION = "tinyplace.harness.control.v1";
+
+export interface HarnessControlFrame {
+  control_version: typeof HARNESS_CONTROL_VERSION;
+  /** `input` types the text into the addressed session's agent as a prompt. */
+  kind: "input";
+  /** Target session (wrapper or harness session id). Absent → primary session. */
+  session_id?: string;
+  text: string;
+}
+
+export interface EncodeHarnessControlOptions {
+  sessionId?: string;
+}
+
+/** Serialize an `input` control frame for the encrypted DM body. */
+export function encodeHarnessControlFrame(
+  text: string,
+  options: EncodeHarnessControlOptions = {},
+): string {
+  const frame: HarnessControlFrame = {
+    control_version: HARNESS_CONTROL_VERSION,
+    kind: "input",
+    ...(options.sessionId ? { session_id: options.sessionId } : {}),
+    text,
+  };
+  return JSON.stringify(frame);
+}
+
+/**
+ * Decode a DM body into a {@link HarnessControlFrame}, or undefined when the
+ * body is not one of ours (plaintext, a session envelope, another protocol, or
+ * a malformed frame). Never throws — inbound Signal DMs are untrusted.
+ */
+export function parseHarnessControlFrame(
+  body: string,
+): HarnessControlFrame | undefined {
+  if (!body.trimStart().startsWith("{")) {
+    return undefined;
+  }
+  let parsed: unknown;
+  try {
+    parsed = JSON.parse(body);
+  } catch {
+    return undefined;
+  }
+  if (parsed === null || typeof parsed !== "object") {
+    return undefined;
+  }
+  const record = parsed as Record<string, unknown>;
+  if (record.control_version !== HARNESS_CONTROL_VERSION) {
+    return undefined;
+  }
+  if (record.kind !== "input" || typeof record.text !== "string") {
+    return undefined;
+  }
+  const sessionId =
+    typeof record.session_id === "string" && record.session_id.length > 0
+      ? record.session_id
+      : undefined;
+  return {
+    control_version: HARNESS_CONTROL_VERSION,
+    kind: "input",
+    ...(sessionId ? { session_id: sessionId } : {}),
+    text: record.text,
+  };
+}

sdk/typescript/src/cli/harness-wrapper.ts

@@ -51,6 +51,8 @@ import {
   type SessionStatusState,
 } from "./harness-status.js";
 import { makeContext } from "./context.js";
+import { parseHarnessControlFrame } from "./harness-control.js";
+import { createMachineBus, type MachineBus } from "./machine-bus.js";
 import { makePathLookup } from "./daemon/providers.js";
 import {
   OpenCodeEventSource,
@@ -83,9 +85,8 @@ export interface HarnessWrapperConfig {
   captureSession: boolean;
   dmRecipient?: string;
   dryRun: boolean;
-  // Opt-in: also emit the v2 typed-event stream (tool_call/tool_result/status/
-  // thinking) alongside v1. Default off so current v1 consumers are untouched
-  // until the OpenHuman receiver understands v2.
+  // The v2 typed-event stream (tool_call/tool_result/status/thinking) is
+  // always emitted alongside v1; the field remains so tests can silence it.
   emitV2: boolean;
   outDir: string;
   provider: HarnessProvider;
@@ -234,10 +235,29 @@ export async function runHarnessCommand(
     stderr: options.stderr ?? process.stderr,
   };
   const writer = new TerminalEnvelopeWriter(config, cwd, stdio.stderr);
-  const publisher = new SessionEnvelopePublisher(config, options, stdio.stderr);
+  // The machine session bus shares this machine's ONE wallet across every
+  // concurrent wrapper session: a cross-process lock serializes Signal I/O and
+  // a spool routes inbound DMs to their addressed session. Only engaged when
+  // the wrapper actually touches the wallet (an owner is configured).
+  const bus =
+    !config.dryRun && Boolean(config.dmRecipient ?? config.receiveFrom)
+      ? createMachineBus({
+          env,
+          wrapperSessionId: config.wrapperSessionId,
+          provider,
+          cwd,
+        })
+      : undefined;
+  const publisher = new SessionEnvelopePublisher(
+    config,
+    options,
+    stdio.stderr,
+    bus,
+  );
   const receiver = config.receiveEnabled
-    ? new InboundMessageReceiver(config, publisher, stdio.stderr)
+    ? new InboundMessageReceiver(config, publisher, stdio.stderr, bus)
     : undefined;
+  bus?.registerSession();
 
   const usePty = config.usePty && options.spawn === undefined;
   // opencode has no per-session transcript files — it is observed over its
@@ -315,6 +335,9 @@ export async function runHarnessCommand(
 
   let exitCode = 1;
   let dmFailures = 0;
+  const busHeartbeat = bus
+    ? setInterval(() => bus.heartbeatSession(), 10_000)
+    : undefined;
   try {
     exitCode = usePty
       ? await runPtyAgent(launch, config, cwd, env, writer, stdio, onInputSink)
@@ -329,6 +352,10 @@ export async function runHarnessCommand(
           onInputSink,
         );
   } finally {
+    if (busHeartbeat) {
+      clearInterval(busHeartbeat);
+    }
+    bus?.endSession();
     // Teardown order: stop the session observer (flushes the publisher) → stop
     // inbound → kill the opencode server. In a `finally` so a spawn failure
     // still tears the server down instead of leaking a `serve` process.
@@ -539,11 +566,9 @@ export function parseHarnessWrapperArgs(
     captureSession: true,
     ...(owner ? { dmRecipient: owner } : {}),
     dryRun: false,
-    emitV2:
-      firstEnv(env, [
-        `TINYPLACE_${provider.toUpperCase()}_V2`,
-        "TINYPLACE_HARNESS_V2",
-      ]) === "1",
+    // v2 is the protocol: the typed-event stream is always emitted (alongside
+    // v1 for older consumers). The former TINYPLACE_*_V2 opt-in is retired.
+    emitV2: true,
     outDir,
     provider,
     statusHeartbeatMs: numberEnvOr(
@@ -1239,8 +1264,14 @@ export class SessionEnvelopePublisher {
     private readonly config: HarnessWrapperConfig,
     private readonly options: TinyPlaceCliOptions,
     private readonly stderr: Writable,
+    private readonly bus?: MachineBus,
   ) {}
 
+  /** Serialize a wallet-touching call across this machine's processes. */
+  private withWalletLock<T>(fn: () => Promise<T>): Promise<T> {
+    return this.bus ? this.bus.withLock(fn) : fn();
+  }
+
   /** Record an owner→agent injection so its echoed-back `user` line is dropped. */
   public markInjected(text: string): void {
     const key = echoKey(text);
@@ -1290,6 +1321,11 @@ export class SessionEnvelopePublisher {
       bridgeLog("publish.echoSkip", { id });
       return;
     }
+    // Record the harness's own session id on the machine registry as soon as
+    // an envelope reveals it — owners address control frames by this id.
+    this.bus?.heartbeatSession({
+      harnessSessionId: envelope.scope.harness_session_id,
+    });
     bridgeLog("publish.enqueue", {
       id,
       role: envelopeRole(envelope),
@@ -1330,24 +1366,24 @@ export class SessionEnvelopePublisher {
       throw new Error("DM forwarding requires a tiny.place signer");
     }
     const recipient = await this.ensureContact(ctx);
+    const signer = ctx.signer;
     try {
-      await sendMessage(
-        ctx.client,
-        ctx.signer,
-        recipient,
-        JSON.stringify(envelope),
+      await this.withWalletLock(() =>
+        sendMessage(ctx.client, signer, recipient, JSON.stringify(envelope)),
       );
     } catch (error) {
       if (!isNotAContactError(error)) {
         throw error;
       }
       this.contactPromise = undefined;
       const refreshedRecipient = await this.ensureContact(ctx);
-      await sendMessage(
-        ctx.client,
-        ctx.signer,
-        refreshedRecipient,
-        JSON.stringify(envelope),
+      await this.withWalletLock(() =>
+        sendMessage(
+          ctx.client,
+          signer,
+          refreshedRecipient,
+          JSON.stringify(envelope),
+        ),
       );
     }
   }
@@ -1496,8 +1532,14 @@ export class InboundMessageReceiver {
     private readonly config: HarnessWrapperConfig,
     private readonly publisher: SessionEnvelopePublisher,
     private readonly stderr: Writable,
+    private readonly bus?: MachineBus,
   ) {}
 
+  /** Serialize a wallet-touching call across this machine's processes. */
+  private withWalletLock<T>(fn: () => Promise<T>): Promise<T> {
+    return this.bus ? this.bus.withLock(fn) : fn();
+  }
+
   /** Register the agent-input writer once the child is spawned. Injection is a
    *  no-op until this is set, so the poll loop can start before the child. */
   public setSink(write: (text: string) => void): void {
@@ -1517,8 +1559,11 @@ export class InboundMessageReceiver {
     }
     // Publish our Signal key bundle so the owner can open a session to us — the
     // wrapper never did this before, which is why it was un-messageable.
+    // Under the machine bus this is wallet-locked: prekey publication mutates
+    // the shared Signal store, and N sessions start concurrently.
+    const signer = ctx.signer;
     try {
-      await publishKeys(ctx.client, ctx.signer);
+      await this.withWalletLock(() => publishKeys(ctx.client, signer));
     } catch (error) {
       this.stderr.write(
         `tinyplace ${this.config.provider}: failed to publish Signal keys: ${describeError(error)}\n`,
@@ -1610,30 +1655,51 @@ export class InboundMessageReceiver {
       if (!ctx.signer) {
         return;
       }
-      const messages = await readMessages(ctx.client, ctx.signer, {
-        limit: 50,
-      });
+      const signer = ctx.signer;
+      this.bus?.heartbeatSession();
+      const messages = await this.withWalletLock(() =>
+        readMessages(ctx.client, signer, { limit: 50 }),
+      );
       let injected = 0;
       let dropped = 0;
       for (const message of messages) {
         const text = this.filterAndParse(message);
-        // `this.sink` is guaranteed defined here — poll() returns early when it
-        // is not. injectOne re-guards it before each write regardless.
-        if (text !== undefined) {
-          injected += 1;
-          // Record before injecting so the tailer's echoed-back `user` line
-          // (the agent logs the injected prompt as its own user turn) is dropped
-          // by the publisher instead of bouncing to the owner.
-          this.publisher.markInjected(text);
-          // Type the body now, then submit with a DISTINCT carriage return after
-          // a short debounce (see injectOne). Enqueued so batched inbound turns
-          // stay serialized (body → Enter → next body) and never interleave.
-          this.enqueueInjection(text);
-        } else {
+        if (text === undefined) {
           dropped += 1;
+          continue;
+        }
+        if (this.bus) {
+          // Machine bus: the inbox is shared by every session on this wallet,
+          // so a message we happened to read may be addressed to ANOTHER
+          // session — spool it; our own spool is drained below.
+          this.bus.routeInbound(message.from, text);
+          continue;
+        }
+        // No bus (single session, no shared wallet): inject directly. A
+        // control frame still resolves to its text so an owner can use one
+        // frame shape everywhere.
+        const frame = parseHarnessControlFrame(text);
+        const body = frame ? frame.text : text;
+        injected += 1;
+        // Record before injecting so the tailer's echoed-back `user` line
+        // (the agent logs the injected prompt as its own user turn) is dropped
+        // by the publisher instead of bouncing to the owner.
+        this.publisher.markInjected(body);
+        // Type the body now, then submit with a DISTINCT carriage return after
+        // a short debounce (see injectOne). Enqueued so batched inbound turns
+        // stay serialized (body → Enter → next body) and never interleave.
+        this.enqueueInjection(body);
+      }
+      if (this.bus) {
+        // Drain what other pollers (or this one) spooled for this session —
+        // plus the default spool when this session is the machine's primary.
+        for (const item of this.bus.consumeInbound()) {
+          injected += 1;
+          this.publisher.markInjected(item.text);
+          this.enqueueInjection(item.text);
         }
       }
-      if (messages.length > 0) {
+      if (messages.length > 0 || injected > 0) {
         bridgeLog("receiver.poll", {
           read: messages.length,
           injected,