Skip to content

feat: migrate to scoped bus API (forService/forSession) #145

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
four-bytes-robby merged 1 commit into from
Jun 14, 2026
Merged

feat: migrate to scoped bus API (forService/forSession) #145

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions bun.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
"four-bytes"
],
"dependencies": {
"@four-bytes/opencode-plugin-lib": "github:four-bytes/four-opencode-plugin-lib#v0.5.1",
"@four-bytes/opencode-plugin-lib": "github:four-bytes/four-opencode-plugin-lib#v0.6.0",
"@opencode-ai/plugin": "1.16.2",
"@opentui/core": "0.3.2",
"@opentui/solid": "0.3.2",
Expand Down
30 changes: 25 additions & 5 deletions src/status.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ import { writeFileSync, mkdirSync, existsSync } from "fs";
import { createHash } from "crypto";
import { homedir } from "os";
import { join } from "path";
import { AsyncLocalStorage } from "node:async_hooks";
import type { PluginInput } from "@opencode-ai/plugin";
import { BusClient } from "@four-bytes/opencode-plugin-lib";
import type { BrainStatusEvent } from "./event-bus";
Expand All @@ -25,11 +26,24 @@ const _state = { current: {} as Record<string, unknown> };
_state.current = { status: "init", statusText: "", version: "" };
let _version = "";
let _sessionId = "";
let _channel = "brain/status";

// ALS stores the session ID for the duration of each tool execute.
// write() reads from here first, so concurrent sessions never overwrite each other's channel.
const _sessionAls = new AsyncLocalStorage<string>();

/**
* Run fn in an ALS context bound to sessionId.
* All updateStatus calls inside fn publish to brain/{sessionId}.
* Outside any withSessionId context (e.g. startup, auto-ingest), they publish to brain/status.
*/
export function withSessionId<T>(id: string, fn: () => Promise<T>): Promise<T> {
return _sessionAls.run(id, fn);
}

let _client: PluginInput["client"] | null = null;
let _server: ReturnType<typeof Bun.serve> | null = null;
let _port = 0;
let _brainBus: BusClient | null = null;
let _busPromise: Promise<BusClient> | null = null;

/** Initialize with client for toast support */
Expand All @@ -41,7 +55,6 @@ export function initVersion(v: string): void {
export function setSessionId(id: string): void {
if (id === _sessionId) return;
_sessionId = id;
_channel = `brain/${id}`;

if (_port > 0) {
try {
Expand Down Expand Up @@ -116,11 +129,18 @@ export function stopStatusServer(): void {

function write(data: Record<string, unknown>): void {
_state.current = { ..._state.current, ...data };
const payload = { ..._state.current, version: _version, sessionId: _sessionId || undefined } as BrainStatusEvent;
// 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.
const sid = _sessionAls.getStore() ?? "";
const payload = { ..._state.current, version: _version, sessionId: sid || undefined } as BrainStatusEvent;

@coderabbitai coderabbitai Bot Jun 14, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Type safety violation: sessionId is not in BrainStatusEvent interface.

Line 135 adds sessionId to the payload and casts it to BrainStatusEvent, but the interface (defined in src/event-bus.ts:5-12) does not include a sessionId field. This bypasses TypeScript's type checking and could cause runtime issues if consumers expect only the declared fields.

Proposed fix: Add sessionId to BrainStatusEvent interface

Update src/event-bus.ts to include the sessionId field:

 export interface BrainStatusEvent {
   status?: "init" | "busy" | "ready" | "error";
   statusText?: string;
   current?: number;
   total?: number;
   version?: string;
   error?: string;
+  sessionId?: string;
 }

Then the cast on line 135 will be type-safe.

Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/status.ts` at line 135, The `BrainStatusEvent` interface does not include
a `sessionId` field, but the payload construction in the status module is
attempting to add `sessionId` and casting it to `BrainStatusEvent`, which
violates type safety. Update the `BrainStatusEvent` interface definition to
include the `sessionId` field with an appropriate type (likely `string |
undefined` based on the usage pattern), so that the `sessionId` property is
properly declared and the type cast on line 135 becomes type-safe without
needing to bypass TypeScript's type checking.


// Real-time push via plugin bus (HTTP fallback still serves status endpoint)
// Real-time push via scoped plugin bus (HTTP fallback still serves status endpoint)
getBus()
.then((bus) => bus.publish(_channel, payload))
.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);
});
Expand Down
139 changes: 56 additions & 83 deletions src/tui.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,161 +1,134 @@
/** @jsxImportSource @opentui/solid */

import { createSignal, createEffect, onMount, onCleanup } from "solid-js";
import { createSignal, onMount, onCleanup, Show } from "solid-js";
import type { TuiPlugin, TuiPluginApi } from "@opencode-ai/plugin/tui";
import type { RGBA } from "@opentui/core";
import { BusTui } from "@four-bytes/opencode-plugin-lib/tui";
import { ProgressBar } from "@four-bytes/opencode-plugin-lib/tui-components";
import type { BrainStatusEvent } from "./event-bus";
import { Spinner } from "./spinner";
function BrainStatusBar(props: { centered?: boolean; api: TuiPluginApi; sessionId?: string }) {
const [indicator, setIndicator] = createSignal("•");
const [status, setStatus] = createSignal("connecting…");

function BrainStatusBar(props: { variant: "sidebar" | "home"; api: TuiPluginApi; sessionId?: string }) {
const [statusText, setStatusText] = createSignal("connecting…");
const [version, setVersion] = createSignal("");
const [current, setCurrent] = createSignal(0);
const [total, setTotal] = createSignal(0);
const [pct, setPct] = createSignal(0);
const [fg, setFg] = createSignal<string | RGBA>("");
const [busy, setBusy] = createSignal(false);
let pulse = 0;
const [hasError, setHasError] = createSignal(false);
let lastPoll = Date.now();

const theme = () => props.api.theme.current;
const connecting = () => (!version() || Date.now() - lastPoll > 2000) && !busy();
const showProgress = () => busy() && current() > 0 && total() > 0;

const handleStatus = (data: BrainStatusEvent) => {
try {
lastPoll = Date.now();
setVersion(data.version ?? "");
pulse++;
setHasError(false);

if (data.status === "error") {
setBusy(false);
setIndicator("•");
setStatus(data.error || data.statusText || "error occurred");
setStatusText(data.error || data.statusText || "error occurred");
setFg(theme().error);
setHasError(true);
} else if (data.status === "init") {
setBusy(true);
setStatus(data.statusText ?? "initializing…");
setStatusText(data.statusText ?? "initializing…");
setCurrent(0);
setTotal(0);
setFg(theme().warning);
} else if (data.status === "busy") {
setBusy(true);
setCurrent(data.current ?? 0);
setTotal(data.total ?? 0);
setStatus(data.statusText ?? "working…");
setFg(pulse % 2 === 0 ? theme().warning : theme().accent);
setStatusText(data.statusText ?? "working…");
setFg(theme().warning);
} else {
setCurrent(0);
setTotal(0);
setBusy(false);
setIndicator("•");
setStatus("ready");
setStatusText("ready");
setFg(theme().success);
}
} catch {
setBusy(false);
setIndicator("•");
setStatus("error occurred");
setStatusText("error occurred");
setFg(theme().error);
setHasError(true);
}
};

onMount(() => {
// Bus is a signal so createEffect below can react when BusTui resolves.
const [bus, setBus] = createSignal<BusTui | null>(null);
let unsub: (() => void) | null = null;
let sessionUnsub: (() => void) | null = null;
let unmounted = false;

onCleanup(() => {
unmounted = true;
unsub?.();
sessionUnsub?.();
bus()?.close();
});

// Real-time WebSocket subscription via plugin bus.
// HTTP fallback removed — bus is the only transport; status server stays
// for any external probes (e.g. /status endpoint) but TUI never polls it.
BusTui.connect()
.then((b) => {
if (unmounted) { b.close(); return; }
setBus(b);
// Always subscribe to brain/status — server publishes here during ingest,
// before any chat message creates a session. Once sessionId is known,
// also subscribe to the per-session channel (server switches to it
// after first chat.message). Prevents missing pre-session status updates.
unsub = b.subscribe("brain/status", (envelope) => {
const data = envelope.payload as BrainStatusEvent & { sessionId?: string };
// Only process if no session ID OR matches current session
if (data.sessionId && props.sessionId && data.sessionId !== props.sessionId) return;
handleStatus(data);
// Scoped subscription: forService("brain") + forSession(sid) replaces
// the old brain/{sid} channel. No sessionId filter needed — the bus
// only delivers events for the scoped session (or unscoped when sid missing).
const scoped = b.forService("brain");
const brainBus = props.sessionId ? scoped.forSession(props.sessionId) : scoped;

@cubic-dev-ai cubic-dev-ai Bot Jun 14, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1: Subscription scope is fixed at mount and does not track later sessionId updates. Sidebar can miss status updates after session creation because publisher switches to session-scoped channel.

Prompt for AI agents 
Check if this issue is valid — if so, understand the root cause and fix it. At src/tui.tsx, line 82:

<comment>Subscription scope is fixed at mount and does not track later `sessionId` updates. Sidebar can miss status updates after session creation because publisher switches to session-scoped channel.</comment>

<file context>
@@ -1,161 +1,134 @@
+        // the old brain/{sid} channel. No sessionId filter needed — the bus
+        // only delivers events for the scoped session (or unscoped when sid missing).
+        const scoped = b.forService("brain");
+        const brainBus = props.sessionId ? scoped.forSession(props.sessionId) : scoped;
+        unsub = brainBus.subscribe("status", (envelope) => {
+          handleStatus(envelope.payload as BrainStatusEvent);
</file context>

unsub = brainBus.subscribe("status", (envelope) => {
handleStatus(envelope.payload as BrainStatusEvent);
});
})
.catch((err) => {
console.warn("[brain TUI] BusTui connect failed:", (err as Error).message);
});

// React to props.sessionId changes — sessionId is set later by the host
// (after first chat.message), so it is often undefined at mount time.
// Tracks bus() so the effect re-runs once BusTui.connect() resolves.
createEffect(() => {
const sid = props.sessionId;
const b = bus();
sessionUnsub?.();
if (!sid || !b) return;
sessionUnsub = b.subscribe(`brain/${sid}`, (envelope) => {
const data = envelope.payload as BrainStatusEvent & { sessionId?: string };
if (data.sessionId && data.sessionId !== sid) return;
handleStatus(data);
});
});
});

const StatusRow = () => (
<box width="100%" flexDirection="row">
<text fg={theme().textMuted}>🧠 {version()} </text>
{busy() ? <Spinner fg={fg()} /> : <text fg={connecting() ? theme().error : fg()}>{indicator()}</text>}
<text fg={connecting() ? theme().error : theme().textMuted}> {connecting() ? "connecting…" : status()}</text>
{busy() && current() > 0 && total() > 0 && (
<text> </text>
)}
{busy() && current() > 0 && total() > 0 && (
<ProgressBar current={current()} total={total()} showLabel={true} fillBg="#aaa" fillFg="#000" width="auto" />
)}
</box>
);
const indicatorColor = () => connecting() ? theme().error : (hasError() ? theme().error : fg());
const textColor = () => connecting() ? theme().error : theme().textMuted;

return (
<box width="100%">
{props.centered ? (
<box>
<text> </text>
<box width="100%" flexDirection="row" justifyContent="center">
<text fg={theme().text}><b>Brain</b></text>
<text> </text>
<StatusRow />
</box>
</box>
) : (
<box width="100%">
<Show when={props.variant === "sidebar"}>
{/* Sidebar: header row + status row (two-column: text left, bar right) */}
<box width="100%" flexDirection="column">
<box flexDirection="row">
<text fg={theme().text}><b>Brain</b></text>
<text fg={theme().textMuted}> 🧠 {version()}</text>
</box>
<box width="100%" flexDirection="row">
{busy() ? <Spinner fg={fg()} /> : <text fg={connecting() ? theme().error : fg()}>{indicator()}</text>}
<text fg={connecting() ? theme().error : theme().textMuted}> {connecting() ? "connecting…" : status()}</text>
{busy() && current() > 0 && total() > 0 && (
<text> </text>
)}
{busy() && current() > 0 && total() > 0 && (
<ProgressBar current={current()} total={total()} showLabel={true} fillBg="#aaa" fillFg="#000" width="auto" />
)}
<box width="100%" flexDirection="row" justifyContent="space-between">
<box flexDirection="row" flexShrink={1}>
{busy() ? <Spinner fg={fg()} /> : <text fg={indicatorColor()}>•</text>}
<text fg={textColor()}> {connecting() ? "connecting…" : statusText()}</text>
<Show when={showProgress()}>
<text fg={theme().textMuted}> {current()}/{total()}</text>
</Show>
</box>
<Show when={showProgress()}>
<box flexShrink={0}>
<ProgressBar current={current()} total={total()} showLabel={false} width="auto" barWidth={12} fillBg="#aaa" fillFg="#000" />
</box>
</Show>
</box>
</box>
)}
</Show>
<Show when={props.variant === "home"}>
{/* Home: single compact row, centered */}
<box width="100%" flexDirection="row" justifyContent="center">
<text fg={theme().textMuted}>🧠 {version()} </text>
{busy() ? <Spinner fg={fg()} /> : <text fg={indicatorColor()}>•</text>}
<text fg={textColor()}> {connecting() ? "connecting…" : statusText()}</text>
<Show when={showProgress()}>
<text fg={theme().textMuted}> </text>
<ProgressBar current={current()} total={total()} showLabel={false} width="auto" barWidth={10} fillBg="#aaa" fillFg="#000" />
</Show>
</box>
</Show>
</box>
);
}
Expand All @@ -164,10 +137,10 @@ export { BrainStatusBar };

const tui: TuiPlugin = (api) => {
api.slots.register({
order: 60, // below deepseek-meter (55)
order: 60,
slots: {
sidebar_content: (_ctx: any, props: any) => <BrainStatusBar api={api} sessionId={props.session_id} />,
home_bottom: () => <BrainStatusBar api={api} centered />,
sidebar_content: (_ctx: any, props: any) => <BrainStatusBar api={api} variant="sidebar" sessionId={props.session_id} />,
home_bottom: () => <BrainStatusBar api={api} variant="home" />,
},
});
return Promise.resolve();
Expand Down
Loading