Feat/bus frontend wiring

apps/core/src/agent/loop.ts

@@ -30,7 +30,6 @@ import type {
 import type { SystemBlock } from "../providers/types.js";
 import type { PermissionRequestResult } from "../hooks/PermissionRequest.js";
 import { createInitialSessionState, DEFAULT_LOOP_HEURISTICS } from "./types.js";
-import type { StreamEvent } from "@thisisayande/freecode-shared";
 import { Effect } from "effect";
 import { createToolOrchestrator, getTool } from "../tools/index.js";
 import type { ToolOrchestrator } from "../tools/orchestrator.js";
@@ -98,7 +97,6 @@ export class AgentLoop {
  private orchestrator: ToolOrchestrator;
  private recovery: RecoveryManager;
  private sessionStore: SessionStore | undefined;
- private onToolEvent: ((event: StreamEvent) => void) | undefined;
  private lastThinking: string | undefined;
  private compiler: PromptCompiler;
  // Cancellation: aborted on interrupt(); threaded into provider requests and
@@ -224,7 +222,6 @@ export class AgentLoop {
 
  try {
  this.state = { ...this.state, status: "running" };
- this.onToolEvent = input.onToolEvent;
 
  // Initialize compiler with project info and mode
  this.compiler = new PromptCompiler(
@@ -405,15 +402,15 @@ export class AgentLoop {
  // Emit thinking content if present (for UI to display as streaming reasoning)
  if (providerResult.thinking) {
  this.lastThinking = providerResult.thinking;
- this.onToolEvent?.({
+BusEvents.stream(this.state.sessionId, {
  type: "thinking",
  content: providerResult.thinking,
  });
  }
 
  // Emit text content if present (for UI to display)
  if (providerResult.content) {
- this.onToolEvent?.({
+BusEvents.stream(this.state.sessionId, {
  type: "text",
  content: providerResult.content,
  });
@@ -650,7 +647,7 @@ export class AgentLoop {
  // Prefer streaming when the provider supports it AND we have a listener.
  // If either is missing, fall back to the one-shot execute() path so callers
  // and downstream code paths are unchanged.
- if (aiProvider.stream && this.onToolEvent) {
+ if (aiProvider.stream) {
  let content = "";
  let thinking = "";
  let toolCalls:
@@ -669,11 +666,17 @@ export class AgentLoop {
  switch (chunk.type) {
  case "text_delta":
  content += chunk.delta;
- this.onToolEvent({ type: "text_delta", delta: chunk.delta });
+ BusEvents.stream(this.state.sessionId, {
+ type: "text_delta",
+ delta: chunk.delta,
+ });
  break;
  case "thinking_delta":
  thinking += chunk.delta;
- this.onToolEvent({ type: "thinking_delta", delta: chunk.delta });
+ BusEvents.stream(this.state.sessionId, {
+ type: "thinking_delta",
+ delta: chunk.delta,
+ });
  break;
  case "tool_call":
  (toolCalls ??= []).push({
@@ -831,12 +834,6 @@ export class AgentLoop {
  title: `Tool ${toolCall.tool}`,
  error: `Tool "${toolCall.tool}" is not allowed in plan mode (read-only)`,
  };
- BusEvents.toolCompleted(
- this.state.sessionId,
- toolCall.tool,
- toolCall.id,
- false,
- );
  return blockedResult;
  }
  }
@@ -859,13 +856,6 @@ export class AgentLoop {
  hookContext.toolName ?? toolCall.tool,
  preResult.blockReason ?? "no reason",
  );
- // Emit tool.completed for blocked tool
- BusEvents.toolCompleted(
- this.state.sessionId,
- toolCall.tool,
- toolCall.id,
- false,
- );
  return blockedResult;
  }
 
@@ -899,12 +889,6 @@ export class AgentLoop {
  title: `Tool ${toolCall.tool}`,
  error: `Permission denied: ${permResult.reason ?? "requires approval"}`,
  };
- BusEvents.toolCompleted(
- this.state.sessionId,
- toolCall.tool,
- toolCall.id,
- false,
- );
  return blockedResult;
  }
  }
@@ -921,21 +905,13 @@ export class AgentLoop {
  }
 
  // Emit tool_start event for streaming
- this.onToolEvent?.({
+BusEvents.stream(this.state.sessionId, {
  type: "tool_start",
  toolCallId: toolCall.id,
  toolName: toolCall.tool,
  args: toolCall.args as Record<string, unknown>,
  });
 
- // Emit tool.called event before execution
- BusEvents.toolCalled(
- this.state.sessionId,
- toolCall.tool,
- toolCall.id,
- toolCall.args as Record<string, unknown>,
- );
-
  // Record function.call event
  this.recorder.recordFunctionCall(
  toolCall.tool,
@@ -974,13 +950,6 @@ export class AgentLoop {
  error: String(error),
  duration_ms: Date.now() - startTime,
  };
- BusEvents.toolCompleted(
- this.state.sessionId,
- toolCall.tool,
- toolCall.id,
- false,
- Date.now() - startTime,
- );
  return errorResult;
  }
 
@@ -994,7 +963,7 @@ export class AgentLoop {
  .map((line) =>
  line.length > MAX_LINE_LEN ? line.slice(0, MAX_LINE_LEN) + "..." : line,
  );
- this.onToolEvent?.({
+BusEvents.stream(this.state.sessionId, {
  type: "tool_output",
  toolCallId: toolCall.id,
  content: outputLines.join("\n"),
@@ -1014,19 +983,10 @@ export class AgentLoop {
  Date.now() - startTime,
  );
 
- // Emit tool.completed event with duration
+ // Emit tool_complete event for streaming
  const duration_ms = Date.now() - startTime;
  const success = !result.error;
- BusEvents.toolCompleted(
- this.state.sessionId,
- toolCall.tool,
- toolCall.id,
- success,
- duration_ms,
- );
-
- // Emit tool_complete event for streaming
- this.onToolEvent?.({
+ BusEvents.stream(this.state.sessionId, {
  type: "tool_complete",
  toolCallId: toolCall.id,
  toolName: toolCall.tool,

apps/core/src/agent/types.ts

@@ -267,16 +267,13 @@ export type MessagePart =
 // User Input / Loop Result - Main entry/exit types
 // =============================================================================
 
-import type { StreamEvent } from "@thisisayande/freecode-shared";
-
 export interface UserInput {
  prompt: string;
  sessionId: string;
  provider: string;
  model?: string;
  projectPath: string;
  agentMode?: AgentMode;
- onToolEvent?: (event: StreamEvent) => void;
 }
 
 export interface LoopResult {

apps/core/src/bus/bridge.test.ts

@@ -0,0 +1,61 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+import { busEventToClientEvent } from "./bridge.js";
+
+test("question.asked maps to a question_asked stream event", () => {
+ const out = busEventToClientEvent({
+ type: "question.asked",
+ requestId: "r1",
+ sessionId: "s1",
+ questions: [
+ { question: "Pick", options: [{ label: "A", description: "a" }] },
+ ],
+ } as any);
+ assert.equal(out?.type, "question_asked");
+ assert.equal((out as any).requestId, "r1");
+ assert.equal((out as any).questions.length, 1);
+});
+
+test("internal cache-invalidation events are dropped (return undefined)", () => {
+ assert.equal(
+ busEventToClientEvent({
+ type: "tools.changed",
+ added: [],
+ removed: [],
+ } as any),
+ undefined,
+ );
+ assert.equal(
+ busEventToClientEvent({ type: "mcp.tools.changed", server: "x" } as any),
+ undefined,
+ );
+});
+
+test("a forwarded lifecycle event keeps its type and payload", () => {
+ const out = busEventToClientEvent({
+ type: "subagent.started",
+ subagentId: "a",
+ subagentType: "explore",
+ parentId: "p",
+ task: "t",
+ } as any);
+ assert.equal(out?.type, "subagent.started");
+});
+
+test("a stream relay event is unwrapped to its inner StreamEvent", () => {
+ const inner = { type: "text_delta", delta: "hi" } as const;
+ const out = busEventToClientEvent({
+ type: "stream",
+ sessionId: "s1",
+ event: inner,
+ } as any);
+ assert.deepEqual(out, inner);
+});
+
+test("redundant bus tool.called/tool.completed are dropped", () => {
+ assert.equal(busEventToClientEvent({ type: "tool.called" } as any), undefined);
+ assert.equal(
+ busEventToClientEvent({ type: "tool.completed" } as any),
+ undefined,
+ );
+});

apps/core/src/bus/bridge.ts

@@ -0,0 +1,41 @@
+// =============================================================================
+// Bus → Frontend bridge (pure mapping)
+// Decides how each internal bus event appears on the frontend wire.
+// Returns undefined for internal-only events that must NOT reach frontends.
+// =============================================================================
+
+import type { BusEvent } from "./index.js";
+import type { StreamEvent } from "@thisisayande/freecode-shared";
+
+const INTERNAL_ONLY = new Set([
+ "tools.changed",
+ "mcp.tools.changed",
+ // Redundant with the stream tool_start/tool_complete events (the loop's
+ // authoritative tool lifecycle); dropped so tools are never double-emitted.
+ "tool.called",
+ "tool.completed",
+]);
+
+export function busEventToClientEvent(
+ event: BusEvent,
+): StreamEvent | undefined {
+ if (INTERNAL_ONLY.has(event.type)) return undefined;
+
+ // The bus is only a carrier for stream events — unwrap to the wire language.
+ if (event.type === "stream") return event.event;
+
+ if (event.type === "question.asked") {
+ return {
+ type: "question_asked",
+ requestId: event.requestId,
+ sessionId: event.sessionId,
+ questions: event.questions,
+ };
+ }
+
+ // Lifecycle/progress events (session.*, subagent.*, mcp.server.*, tool.*)
+ // are forwarded verbatim; frontends render what they recognize and ignore
+ // the rest. Cast: these carry richer payloads than the base StreamEvent
+ // union, which frontends read structurally.
+ return event as unknown as StreamEvent;
+}

apps/core/src/bus/index.ts

@@ -6,6 +6,7 @@
 // =============================================================================
 
 import { EventEmitter } from "events";
+import type { StreamEvent } from "@thisisayande/freecode-shared";
 
 // ============================================================================
 // Event Definitions
@@ -106,7 +107,7 @@ export interface ToolCompletedEvent {
 export interface QuestionAskedEvent {
  type: "question.asked";
  requestId: string;
- sessionId: string;
+ sessionId?: string;
  questions: Array<{
  question: string;
  header?: string;
@@ -149,11 +150,25 @@ export interface MCPServerErrorEvent {
  error: string;
 }
 
+// ============================================================================
+// Stream Relay Event
+// Transports a per-session StreamEvent (turn output: text/thinking/tool
+// deltas) over the bus so it shares the single frontend egress. The bus is
+// only the carrier — StreamEvent remains the wire language.
+// ============================================================================
+
+export interface StreamRelayEvent {
+ type: "stream";
+ sessionId: string;
+ event: StreamEvent;
+}
+
 // ============================================================================
 // Union of all Bus Events
 // ============================================================================
 
 export type BusEvent =
+ | StreamRelayEvent
  | SessionCreatedEvent
  | SessionUpdatedEvent
  | SessionErrorEvent
@@ -235,6 +250,7 @@ const pendingQuestions = new Map<
 export async function askQuestion(
  requestId: string,
  questions: QuestionAskedEvent["questions"],
+ sessionId?: string,
 ): Promise<string[]> {
  return new Promise((resolve, reject) => {
  // Store the pending question
@@ -244,11 +260,13 @@ export async function askQuestion(
  bus.publish({
  type: "question.asked",
  requestId,
+ sessionId,
  questions,
  } as QuestionAskedEvent);
 
- // Timeout after 5 minutes
- setTimeout(
+ // Timeout after 5 minutes. unref() so a pending question never keeps the
+ // process alive on its own (it also lets tests exit once resolved).
+ const timer = setTimeout(
  () => {
  if (pendingQuestions.has(requestId)) {
  pendingQuestions.delete(requestId);
@@ -257,6 +275,7 @@ export async function askQuestion(
  },
  5 * 60 * 1000,
  );
+ timer.unref?.();
  });
 }
 
@@ -287,6 +306,9 @@ export function rejectQuestion(requestId: string): void {
 // ============================================================================
 
 export const BusEvents = {
+ stream: (sessionId: string, event: StreamEvent) =>
+ bus.publish({ type: "stream", sessionId, event } as StreamRelayEvent),
+
  sessionCreated: (sessionId: string, projectPath: string) =>
  bus.publish({
  type: "session.created",