fix: sidebar always subscribes to brain/status for ingest progress

ISSUES.md

@@ -0,0 +1,53 @@
+**Status:** Last reviewed 2026-06-14. 2/2 fixed (brain), 3/3 fixed (plugin-lib), 2/2 fixed (local-bus), 2/2 fixed (context-curator).
+
+# Known Issues
+
+## #1 — Bus reconnect never triggered after initial connection
+
+**Symptom:** Once brain's `_busPromise` resolves successfully, a subsequent bus death (idle
+shutdown, crash) causes all `write()` calls to fail silently. No reconnect is ever attempted
+for the lifetime of the plugin process. The TUI stops receiving status updates; the bus
+never restarts from the brain side.
+
+**Root cause:** `getBus()` in `status.ts` caches `_busPromise` and only nulls it on
+`BusClient.connect()` rejection. `BusClient.connect()` never rejects (falls back to
+`MemoryBusClient`), so `_busPromise` is never null after the first call.
+
+When the bus dies post-connect, `write()` swallows the publish error in `.catch()` but does
+NOT reset `_busPromise`. Every subsequent `write()` calls the same dead `BusClient`.
+
+**Location:** `src/status.ts` — `write()` (line ~139) and `getBus()` (line ~75).
+
+**Fix:** Reset `_busPromise = null` when a publish fails, so the next `write()` triggers
+`BusClient.connect()` which re-discovers the new bus (or spawns one):
+
+```typescript
+getBus()
+  .then(async (bus) => {
+    const scoped = bus.forService("brain");
+    const target = sid ? scoped.forSession(sid) : scoped;
+    await target.publish("status", payload);
+  })
+  .catch((err) => {
+    console.warn("[brain] Bus publish failed:", (err as Error).message);
+    _busPromise = null; // reset — next write() will reconnect
+  });
+```
+
+**Dependency:** This fix is only fully effective once `@four-bytes/opencode-plugin-lib` has
+the spawn lock (plugin-lib ISSUES #2) to prevent simultaneous re-spawn races.
+
+---
+
+✅ FIXED — commit cb15dc8 (reset `_busPromise = null` in write catch)
+
+## #2 — ALS-based session scoping is a footgun (future work)
+
+`withSessionId` / `AsyncLocalStorage` wraps every tool execute to scope status publishes to
+the right channel. This is fragile: any await that crosses an async boundary without the ALS
+context will silently publish to the wrong (or no) session channel.
+
+The ROADMAP (Wave 2, Task 2.3) replaces this with an explicit `createSessionStatus(sessionId)`
+that holds a `SessionPublisher`. No ALS needed; session ID is passed explicitly at call sites.
+
+⚠️ ROADMAP Wave 2 — not yet fixed.

package.json

@@ -15,7 +15,7 @@
     "four-bytes"
   ],
   "dependencies": {
-    "@four-bytes/opencode-plugin-lib": "github:four-bytes/four-opencode-plugin-lib#v0.6.0",
+    "@four-bytes/opencode-plugin-lib": "github:four-bytes/four-opencode-plugin-lib#v0.6.1",
     "@opencode-ai/plugin": "1.16.2",
     "@opentui/core": "0.3.2",
     "@opentui/solid": "0.3.2",
@@ -30,6 +30,9 @@
   "devDependencies": {
     "bun-types": "1.3.14"
   },
+  "bin": {
+    "brain-cli": "src/cli.ts"
+  },
   "exports": {
     "./server": {
       "types": "./dist/four-opencode-brain.d.ts",

src/four-opencode-brain.ts

@@ -53,7 +53,7 @@ function calculateIngestTimeout(fileCount: number): number {
 }
 
 /** Unified status updates — see src/status.ts */
-import { updateStatus, initStatus, initVersion, setSessionId, stopStatusServer, toast } from "./status";
+import { updateStatus, initStatus, initVersion, setSessionId, stopStatusServer, toast, withSessionId } from "./status";
 
 
 
@@ -91,18 +91,25 @@ const _serverPlugin = async (input: PluginInput) => {
   try { hasGit = statSync(join(normDir, ".git")).isDirectory(); } catch {}
   const shouldSkip = !hasGit || isSystemDir;
 
-  if (autoIngest && directory && !shouldSkip) {
-    log("info", "auto-ingest", "Auto-ingest starting", { directory });
-
-    // Fire-and-forget — don't block plugin readiness
-    (async () => {
+  // Auto-ingest is deferred until session.created — we need a session ID to publish
+  // status updates on a scoped bus channel. The actual ingest is triggered from the
+  // "event" hook below. _autoIngestDone ensures it runs exactly once per plugin lifetime.
+  let _autoIngestDone = false;
+
+  /**
+   * Run the auto-ingest inside a withSessionId(...) context so all updateStatus calls
+   * publish on brain/{sessionId} — matching the TUI's forSession(sessionId) subscription.
+   */
+  const runAutoIngest = async (sessionID: string) => {
+    log("info", "auto-ingest", "Auto-ingest starting (deferred)", { directory, sessionID });
+    return withSessionId(sessionID, async () => {
       // Signal TUI we're scanning the directory tree
       updateStatus("busy", { text: "scanning files…", total: 0 });
 
       // Quick preliminary file count for toast + timeout calculation
       let fileCount = 0;
       try {
-        const walked = await resolveFiles(directory, true);
+        const walked = await resolveFiles(directory!, true);
         fileCount = walked.files.length;
         updateStatus("busy", { text: `scanning files… ${fileCount}`, total: fileCount });
         const timeoutS = (calculateIngestTimeout(fileCount) / 1000).toFixed(0);
@@ -118,10 +125,10 @@ const _serverPlugin = async (input: PluginInput) => {
       let lastUpdate = 0;
       try {
         const result = await withTimeout(
-          ingestPath(ingestDb, directory, {
+          ingestPath(ingestDb, directory!, {
             recursive: true,
             reIndex: false,
-            project: directory,
+            project: directory!,
             progressCallback: ({ current, total }) => {
               const now = Date.now();
               if (now - lastUpdate < 500 && current !== total) return; // throttle to 0.5s (but always emit final update)
@@ -133,7 +140,7 @@ const _serverPlugin = async (input: PluginInput) => {
           `auto-ingest ${directory}`,
         );
         if (result.filesFound === 0) {
-          const dirname = directory.split("/").filter(Boolean).pop() ?? directory;
+          const dirname = directory!.split("/").filter(Boolean).pop() ?? directory!;
           const msg = `🧠 Found 0 files in ${dirname} — check path`;
           updateStatus("warning", { text: msg.replace("🧠 ", ""), toast: msg.replace("🧠 ", "") });
           toast( msg.replace("🧠 ", ""), "warning", "Brain 🧠");
@@ -173,10 +180,10 @@ const _serverPlugin = async (input: PluginInput) => {
       } finally {
         ingestDb.close();
       }
-    })();
-  }
+    });
+  };
 
-  else if (autoIngest && shouldSkip) {
+  if (autoIngest && shouldSkip) {
     log("warn", "auto-ingest", "Skipped — not a git repo or system dir: " + normDir);
     updateStatus("warning", { text: "ingest excluded" });
   }
@@ -229,6 +236,7 @@ const _serverPlugin = async (input: PluginInput) => {
       reIndex: s.boolean().optional().describe("Force re-index even if unchanged (default: false)"),
     },
     execute: async (args, toolCtx) => {
+      return withSessionId(toolCtx.sessionID, async () => {
       const db = initBrainDatabase();
       const resolvedPath = resolve(toolCtx.directory, args.path);
       try {
@@ -288,6 +296,7 @@ const _serverPlugin = async (input: PluginInput) => {
       } finally {
         db.close();
       }
+      }); // withSessionId
     },
   });
 
@@ -301,6 +310,7 @@ const _serverPlugin = async (input: PluginInput) => {
       project: s.string().optional().describe("Project name or hash to scope search"),
     },
     execute: async (args, toolCtx) => {
+      return withSessionId(toolCtx.sessionID, async () => {
       updateStatus("busy", { text: "searching…" });
       const db = initBrainDatabase();
       try {
@@ -339,13 +349,15 @@ const _serverPlugin = async (input: PluginInput) => {
       } finally {
         db.close();
       }
+      }); // withSessionId
     },
   });
 
   const brain_reindex = tool({
     description: "Rebuild vec0 vector index from chunks.",
     args: {},
-    execute: async () => {
+    execute: async (_args, toolCtx) => {
+      return withSessionId(toolCtx.sessionID, async () => {
       updateStatus("busy", { text: "Rebuilding vector index…" });
       const db = initBrainDatabase();
       try {
@@ -384,6 +396,7 @@ const _serverPlugin = async (input: PluginInput) => {
       } finally {
         db.close();
       }
+      }); // withSessionId
     },
   });
 
@@ -407,7 +420,8 @@ const _serverPlugin = async (input: PluginInput) => {
       diaryContent: s.string().optional().describe("Diary entry content (for add)"),
       diaryDate: s.string().optional().describe("Diary date YYYY-MM-DD (defaults today)"),
     },
-    execute: async (args) => {
+    execute: async (args, toolCtx) => {
+      return withSessionId(toolCtx.sessionID, async () => {
       const db = initBrainDatabase();
       try {
         switch (args.mode) {
@@ -481,6 +495,7 @@ const _serverPlugin = async (input: PluginInput) => {
       } finally {
         db.close();
       }
+      }); // withSessionId
     },
   });
 
@@ -519,7 +534,8 @@ const _serverPlugin = async (input: PluginInput) => {
       confidence: s.number().optional(),
       review_state: s.string().optional(),
     },
-    execute: async (args) => {
+    execute: async (args, toolCtx) => {
+      return withSessionId(toolCtx.sessionID, async () => {
       const db = initBrainDatabase();
       try {
                 updateStatus("busy", { text: "Saving knowledge entry…" });
@@ -545,6 +561,7 @@ const _serverPlugin = async (input: PluginInput) => {
       } finally {
         db.close();
       }
+      }); // withSessionId
     },
   });
 
@@ -560,7 +577,8 @@ const _serverPlugin = async (input: PluginInput) => {
       commit_ref: s.string().optional(),
       observed_symptoms: s.string().optional(),
     },
-    execute: async (args) => {
+    execute: async (args, toolCtx) => {
+      return withSessionId(toolCtx.sessionID, async () => {
       const db = initBrainDatabase();
       try {
                 updateStatus("busy", { text: "Recording occurrence…" });
@@ -584,6 +602,7 @@ const _serverPlugin = async (input: PluginInput) => {
       } finally {
         db.close();
       }
+      }); // withSessionId
     },
   });
 
@@ -595,7 +614,8 @@ const _serverPlugin = async (input: PluginInput) => {
       review_state: s.string().describe("draft|reviewed|accepted|rejected|superseded"),
       confidence: s.number().optional(),
     },
-    execute: async (args) => {
+    execute: async (args, toolCtx) => {
+      return withSessionId(toolCtx.sessionID, async () => {
       const db = initBrainDatabase();
       try {
                 updateStatus("busy", { text: "Updating review…" });
@@ -615,6 +635,7 @@ const _serverPlugin = async (input: PluginInput) => {
       } finally {
         db.close();
       }
+      }); // withSessionId
     },
   });
 
@@ -629,7 +650,8 @@ const _serverPlugin = async (input: PluginInput) => {
       limit: s.number().optional().describe("Max results (default 20)"),
       offset: s.number().optional().describe("Result offset"),
     },
-    execute: async (args) => {
+    execute: async (args, toolCtx) => {
+      return withSessionId(toolCtx.sessionID, async () => {
       const db = initBrainDatabase();
       try {
         const results = kbSearch(db, {
@@ -650,6 +672,7 @@ const _serverPlugin = async (input: PluginInput) => {
       } finally {
         db.close();
       }
+      }); // withSessionId
     },
   });
 
@@ -695,6 +718,7 @@ const _serverPlugin = async (input: PluginInput) => {
   return {
     "experimental.chat.system.transform": async (_hookInput, output) => {
       output.system.push(brainSystemPrompt());
+      if (_hookInput?.sessionID) setSessionId(_hookInput.sessionID);
     },
     "chat.message": async (_hookInput, output) => {
       if (_hookInput?.sessionID) setSessionId(_hookInput.sessionID);
@@ -713,6 +737,23 @@ const _serverPlugin = async (input: PluginInput) => {
       }
     },
     "event": async (eventInput) => {
+      // Capture session ID as soon as a session exists — this enables forSession(sid)
+      // publishes from updateStatus(). Also triggers deferred auto-ingest exactly once,
+      // wrapped in withSessionId(sid) so its status updates land on the right channel.
+      if (eventInput.event.type === "session.created") {
+        const { sessionID } = eventInput.event.properties as { sessionID?: string };
+        if (sessionID) {
+          setSessionId(sessionID);
+          if (!_autoIngestDone && autoIngest && directory && !shouldSkip) {
+            _autoIngestDone = true;
+            // Fire-and-forget — don't block the event hook
+            runAutoIngest(sessionID).catch((err) => {
+              log("error", "auto-ingest", `Deferred auto-ingest failed: ${String(err)}`);
+            });
+          }
+        }
+        return;
+      }
       if (eventInput.event.type === "session.idle") {
         const { sessionID } = eventInput.event.properties;
         let text = "";

src/status.ts

@@ -65,6 +65,11 @@ export function setSessionId(id: string): void {
       writeFileSync(portFile, JSON.stringify({ port: _port }));
     } catch { /* ignore */ }
   }
+
+  // Re-publish current state so TUI receives it even when session.created never fired (continue mode)
+  void withSessionId(id, async () => {
+    write({});
+  });
 }
 
 export function initStatus(client: PluginInput["client"], directory: string): void {
@@ -130,15 +135,18 @@ export function stopStatusServer(): void {
 function write(data: Record<string, unknown>): void {
   _state.current = { ..._state.current, ...data };
   // ALS-stored session ID wins over global (prevents cross-session channel overwrite).
-  // If no ALS context (startup, auto-ingest fire-and-forget), use the unscoped "brain" service.
+  // If no session ID is available, skip the bus publish entirely — the TUI only
+  // subscribes on forSession(sid), so there is no unscoped consumer to receive on.
+  // The HTTP /status endpoint still serves polling clients.
   const sid = _sessionAls.getStore() ?? "";
   const payload = { ..._state.current, version: _version, sessionId: sid || undefined } as BrainStatusEvent;
 
+  if (!sid) return;
+
   // Real-time push via scoped plugin bus (HTTP fallback still serves status endpoint)
   getBus()
     .then(async (bus) => {
-      const scoped = bus.forService("brain");
-      const target = sid ? scoped.forSession(sid) : scoped;
+      const target = bus.forService("brain").forSession(sid);
       await target.publish("status", payload);
     })
     .catch((err) => {