fix(runtime): start the reader when the supervisor owns the Core (#99)

readerEnabled was only honored on the attach path. The normal first launch
spawns its own Core and returned from start() without calling startReader(),
so the read-only Reader never ran: no reader_port in the receipt and every
read fell back to Core.

Start the Reader after every successful owned start (first launch, race
recovery and relaunch). Clear readerChild when it exits so a later restart
can spawn a fresh one, and stop the old Reader before a relaunch. Add a
public-start regression using fake child processes.

Co-authored-by: qiye.sjh <shijiahao_0802@youzan.com>
Co-authored-by: Claude Code <noreply@anthropic.com>

memhub/packages/runtime-control/src/core-supervisor-reader-start.test.ts

@@ -0,0 +1,66 @@
+import type { ChildProcess, SpawnOptions } from "node:child_process";
+import { EventEmitter } from "node:events";
+import { describe, expect, it, vi } from "vitest";
+
+import { CoreSupervisor } from "./core-supervisor.js";
+
+function fakeChild(): ChildProcess {
+  const child = new EventEmitter() as ChildProcess;
+  child.stderr = new EventEmitter() as ChildProcess["stderr"];
+  child.stdout = new EventEmitter() as ChildProcess["stdout"];
+  child.kill = vi.fn(() => {
+    setImmediate(() => child.emit("exit", 0, null));
+    return true;
+  });
+  return child;
+}
+
+function fakeReaderChild(): ChildProcess {
+  return fakeChild();
+}
+
+describe("CoreSupervisor reader startup", () => {
+  it("starts the reader after spawning its own Core", async () => {
+    const core = fakeChild();
+    const reader = fakeReaderChild();
+    const spawned: string[][] = [];
+    const spawnProcess = vi.fn(
+      (_command: string, args: readonly string[], _options: SpawnOptions) => {
+        spawned.push([...args]);
+        if (args.includes("--reader")) {
+          // The Reader attaches its stderr listener synchronously after this
+          // call returns, so publish the port on the next tick.
+          setImmediate(() =>
+            reader.stderr?.emit("data", Buffer.from("reader_port:43210\n")),
+          );
+          return reader;
+        }
+        return core;
+      },
+    );
+    const supervisor = new CoreSupervisor({
+      launch: { command: "one-memory", args: [] },
+      dataDir: "/tmp/one-memory-reader-start-test",
+      instanceId: "owned",
+      port: 9600,
+      attempts: 1,
+      delayMs: 0,
+      readerEnabled: true,
+      spawnProcess,
+      probeHealth: async () => ({
+        status: "ok",
+        service: "one-memory-core",
+        version: "0.0.0",
+        instance_id: "owned",
+      }),
+    });
+
+    await supervisor.start();
+
+    expect(spawnProcess).toHaveBeenCalledTimes(2);
+    expect(spawned[1]).toContain("--reader");
+    expect(supervisor.readerPort).toBe(43210);
+
+    await supervisor.stop();
+  });
+});

memhub/packages/runtime-control/src/core-supervisor.ts

@@ -339,7 +339,11 @@ export class CoreSupervisor {
     }
     this.useOwnedIdentity();
     try {
-      return await this.attempt();
+      const health = await this.attempt();
+      if (this.options.readerEnabled) {
+        await this.startReader();
+      }
+      return health;
     } catch (error) {
       // A standalone Runtime may have won the race after our initial
       // discovery read but before our child acquired the data lock. Verify it
@@ -347,7 +351,12 @@ export class CoreSupervisor {
       let raced: HealthResponse | null = null;
       try {
         raced = await this.attachExisting();
-        if (raced !== null) return raced;
+        if (raced !== null) {
+          if (this.options.readerEnabled) {
+            await this.startReader();
+          }
+          return raced;
+        }
       } catch (attachError) {
         // A stale receipt can fail identity validation after the owned Core
         // exited. Treat that as a cleanup candidate, not as a new startup
@@ -387,7 +396,11 @@ export class CoreSupervisor {
       this.exitSignal = null;
       this.spawnError = null;
       try {
-        return await this.attempt();
+        const health = await this.attempt();
+        if (this.options.readerEnabled) {
+          await this.startReader();
+        }
+        return health;
       } catch (retryError) {
         await this.logStartFailure(retryError);
         throw retryError;
@@ -696,6 +709,9 @@ export class CoreSupervisor {
           await delay(delayMs);
           if (this.stopping) return;
           try {
+            // The old Reader pointed at the Core that just died. Stop it so
+            // the relaunched Core gets a Reader with the current token/port.
+            await this.stopReader();
             // Keep the configured port stable across relaunches. The packaged
             // renderer and discovery receipts point at this port, so moving to
             // an ephemeral port would make a recovered Core unreachable.
@@ -706,6 +722,9 @@ export class CoreSupervisor {
             this.exited = false;
             this.spawnError = null;
             const health = await this.attempt();
+            if (this.options.readerEnabled) {
+              await this.startReader();
+            }
             await this.options.diagnostics?.log({
               level: "info",
               component: "runtime.core",
@@ -824,6 +843,14 @@ export class CoreSupervisor {
       cwd: this.options.launch.cwd,
     });
     this.readerChild = child;
+    // A Reader exits when its Core goes away. Clear the handle so the restart
+    // path can spawn a fresh one instead of returning early forever.
+    child.on("exit", () => {
+      if (this.readerChild === child) {
+        this.readerChild = null;
+        this.activeReaderPort = null;
+      }
+    });
     // Wait briefly for the reader to publish its port on stderr.
     this.activeReaderPort = await new Promise<number | null>((resolve) => {
       let settled = false;