feat: stream supervisor responses + concurrent tool calls (0.4.0)

Chat-completions now streams progressively: live tool-round progress markers as
the turn runs, then the final answer — instead of blocking until the whole turn
finishes. Tool calls within a round run concurrently; default maxToolRounds 8->12;
round-limit message is now a calm status, not an error. Verified live: SSE chunks
arrive incrementally. Removed the dead post-hoc word-streamer. 110 tests pass.

(Note: build orchestration 'bun run build' has a Windows-only nested-bun quirk;
sub-builds + typechecks all pass; CI builds on ubuntu.)

Author: doeixd

CHANGELOG.md

@@ -4,6 +4,20 @@ All notable changes to this project are documented here.
 
 Format based on [Keep a Changelog](https://keepachangelog.com/).
 
+## [0.4.0] - 2026-06-24
+
+### Added
+
+- **Streaming supervisor responses.** The chat-completions endpoint now streams progressively: as the turn runs its tool rounds (the slow part), live markers (`→ start_loop`, `→ loop_status`, …) appear, then the final answer streams — instead of hanging silently until the whole turn finishes and dumping it at once. Verified live: SSE `chat.completion.chunk`s arrive incrementally.
+
+### Changed
+
+- Supervisor **tool calls within a round run concurrently** now (was sequential), and the default **`maxToolRounds` is raised 8 → 12**. Hitting the round limit no longer reads as an error — it's a calm "I've taken several steps … ask me to continue" status. (Surfaced by the end-to-end smoke loop, where the prior limit/error was easy to hit.)
+
+### Notes
+
+- The end-to-end smoke loop validated the full stack live (auto-start, credential auto-detect, orchestration, the `ralph-worker` agent, and a real worker creating a file until `verify` passed). It also surfaced a footgun to address next: when no worker model is set, the worker can fall back to the `ralph-rlm/supervisor` model — set `worker.providerID`/`worker.modelID` in `ralph-provider.json`.
+
 ## [0.3.10] - 2026-06-24
 
 ### Changed

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@doeixd/opencode-ralph-rlm",
-  "version": "0.3.10",
+  "version": "0.4.0",
   "description": "Ralph RLM v0.2: OpenCode supervisor provider + loop engine + worker plugin",
   "type": "module",
   "workspaces": [

packages/engine/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@doeixd/opencode-ralph-rlm-engine",
   "private": true,
-  "version": "0.3.10",
+  "version": "0.4.0",
   "description": "Ralph RLM loop engine — verify, rollover, OpenCode SDK worker lifecycle",
   "license": "MIT",
   "repository": {

packages/provider/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@doeixd/opencode-ralph-rlm-provider",
   "private": true,
-  "version": "0.3.10",
+  "version": "0.4.0",
   "description": "Ralph RLM OpenAI-compatible supervisor provider (Nitro)",
   "license": "MIT",
   "repository": {

packages/provider/server/lib/openai-compat.ts

@@ -87,55 +87,6 @@ export function encodeSseChunk(payload: Record<string, unknown>): string {
   return `data: ${JSON.stringify(payload)}\n\n`;
 }
 
-export function streamCompletionText(
-  request: OpenAIChatCompletionRequest,
-  text: string
-): ReadableStream<Uint8Array> {
-  const id = makeCompletionId();
-  const model = request.model ?? "ralph-rlm/supervisor";
-  const created = Math.floor(Date.now() / 1000);
-  const encoder = new TextEncoder();
-
-  return new ReadableStream({
-    start(controller) {
-      const words = text.split(/(\s+)/);
-      let index = 0;
-
-      const push = () => {
-        if (index >= words.length) {
-          controller.enqueue(
-            encoder.encode(
-              encodeSseChunk({
-                id,
-                object: "chat.completion.chunk",
-                created,
-                model,
-                choices: [{ index: 0, delta: {}, finish_reason: "stop" }],
-              })
-            )
-          );
-          controller.enqueue(encoder.encode("data: [DONE]\n\n"));
-          controller.close();
-          return;
-        }
-
-        const piece = words[index] ?? "";
-        index += 1;
-        controller.enqueue(
-          encoder.encode(
-            encodeSseChunk({
-              id,
-              object: "chat.completion.chunk",
-              created,
-              model,
-              choices: [{ index: 0, delta: { content: piece }, finish_reason: null }],
-            })
-          )
-        );
-        push();
-      };
-
-      push();
-    },
-  });
-}
+// Streaming is now assembled in the chat-completions route (it interleaves live
+// tool-round progress with the final answer); the old post-hoc word-streamer was
+// removed in favor of that real progressive stream.

packages/provider/server/lib/supervisor-agent.ts

@@ -165,7 +165,8 @@ async function runTestModeTurn(
 
 async function runLlmTurn(
   input: SupervisorTurnInput,
-  ctx: SupervisorToolContext
+  ctx: SupervisorToolContext,
+  onProgress?: (text: string) => void
 ): Promise<SupervisorTurnResult> {
   const config = await loadSupervisorLlmConfig(ctx.worktree);
 
@@ -205,23 +206,31 @@ async function runLlmTurn(
       })),
     });
 
+    // Surface progress to the streaming client as each round's tools run — this
+    // is the slow part of a turn, so live markers beat a silent wait.
     for (const call of result.toolCalls) {
-      const output = await executeSupervisorTool(
-        call.name,
-        parseToolArgs(call.arguments),
-        ctx
-      );
+      onProgress?.(`_→ ${call.name}_\n`);
+    }
+
+    // Tool calls within a round are independent — run them concurrently, then
+    // append results in the model's original order.
+    const outputs = await Promise.all(
+      result.toolCalls.map((call) =>
+        executeSupervisorTool(call.name, parseToolArgs(call.arguments), ctx)
+      )
+    );
+    result.toolCalls.forEach((call, i) => {
       conversation.push({
         role: "tool",
         tool_call_id: call.id,
-        content: output,
+        content: outputs[i] ?? "",
       });
-    }
+    });
   }
 
   const status = await executeSupervisorTool("loop_status", {}, ctx);
   return {
-    content: `Reached tool round limit. Current status:\n${status}`,
+    content: `I've taken several steps but haven't fully wrapped up this turn. Current status:\n${status}\n\nAsk me to continue if you'd like me to keep going.`,
     toolRounds,
     mode: "llm",
   };
@@ -261,3 +270,23 @@ export async function supervisorTurn(
     isTestMode() ? runTestModeTurn(input, ctx) : runLlmTurn(input, ctx)
   );
 }
+
+/**
+ * Streaming variant: runs the same serialized turn but calls `onProgress` with
+ * live markers as tool rounds execute, and returns the final result (whose
+ * `content` the caller streams to the client). Test mode has no tool rounds, so
+ * it emits no progress — just the scripted result.
+ */
+export async function supervisorTurnStreaming(
+  input: SupervisorTurnInput,
+  onProgress: (text: string) => void
+): Promise<SupervisorTurnResult> {
+  const ctx: SupervisorToolContext = {
+    sessionKey: input.sessionKey,
+    worktree: input.worktree,
+  };
+
+  return withSessionTurnLock(input.sessionKey, () =>
+    isTestMode() ? runTestModeTurn(input, ctx) : runLlmTurn(input, ctx, onProgress)
+  );
+}

packages/provider/server/lib/supervisor-config.ts

@@ -66,7 +66,7 @@ export async function loadSupervisorLlmConfig(
   const maxToolRounds = toBoundedInt(
     fileConfig.supervisor?.maxToolRounds ??
       Number(process.env.RALPH_SUPERVISOR_MAX_TOOL_ROUNDS),
-    8,
+    12,
     1,
     24
   );

packages/provider/server/routes/api/health.get.ts

@@ -16,7 +16,7 @@ export default defineHandler(async () => {
   return {
     healthy: true,
     provider: "@doeixd/opencode-ralph-rlm",
-    version: "0.3.10",
+    version: "0.4.0",
     opencode: {
       baseUrl: runtime.baseUrl,
       ...opencode,

packages/provider/server/routes/v1/chat/completions.post.ts

@@ -1,7 +1,8 @@
 import { defineHandler, readBody } from "nitro/h3";
 import {
   buildCompletionResponse,
-  streamCompletionText,
+  encodeSseChunk,
+  makeCompletionId,
   type OpenAIChatCompletionRequest,
 } from "../../../lib/openai-compat.js";
 import {
@@ -14,7 +15,7 @@ import {
   logSessionDebug,
 } from "../../../lib/session-debug.js";
 import { resolveWorktree } from "../../../lib/worktree.js";
-import { supervisorTurn } from "../../../lib/supervisor-agent.js";
+import { supervisorTurn, supervisorTurnStreaming } from "../../../lib/supervisor-agent.js";
 
 export default defineHandler(async (event): Promise<Response> => {
   const body = await readBody<OpenAIChatCompletionRequest>(event);
@@ -53,46 +54,96 @@ export default defineHandler(async (event): Promise<Response> => {
     );
   }
 
-  let turn;
-  try {
-    turn = await supervisorTurn({
-      sessionKey: session.sessionKey,
-      worktree,
-      messages: body.messages,
-      ...(body.model ? { model: body.model } : {}),
-    });
-  } catch (err) {
-    const message = err instanceof Error ? err.message : String(err);
-    return new Response(
-      JSON.stringify({
-        error: { message, type: "api_error" },
-      }),
-      { status: 500, headers: { "content-type": "application/json" } }
-    );
-  }
-
-  const headers = {
+  const baseHeaders = {
     "x-ralph-session-key": session.sessionKey,
     "x-ralph-session-source": session.source,
-    "x-ralph-supervisor-mode": turn.mode,
   };
 
+  const turnInput = {
+    sessionKey: session.sessionKey,
+    worktree,
+    messages: body.messages,
+    ...(body.model ? { model: body.model } : {}),
+  };
+
+  // Streaming: emit live progress markers as the turn's tool rounds run (the
+  // slow part), then stream the final answer — instead of blocking until the
+  // whole turn finishes and dumping it at once.
   if (body.stream) {
-    const stream = streamCompletionText(body, turn.content);
+    const id = makeCompletionId();
+    const model = body.model ?? "ralph-rlm/supervisor";
+    const created = Math.floor(Date.now() / 1000);
+    const encoder = new TextEncoder();
+    const delta = (content: string) =>
+      encoder.encode(
+        encodeSseChunk({
+          id,
+          object: "chat.completion.chunk",
+          created,
+          model,
+          choices: [{ index: 0, delta: { content }, finish_reason: null }],
+        })
+      );
+
+    const stream = new ReadableStream<Uint8Array>({
+      async start(controller) {
+        // An immediate keep-alive delta so the client shows activity at once.
+        controller.enqueue(delta(""));
+        try {
+          const turn = await supervisorTurnStreaming(turnInput, (text) =>
+            controller.enqueue(delta(text))
+          );
+          for (const piece of (turn.content || "Done.").split(/(\s+)/)) {
+            if (piece) controller.enqueue(delta(piece));
+          }
+        } catch (err) {
+          const message = err instanceof Error ? err.message : String(err);
+          controller.enqueue(delta(`\n\n[error] ${message}`));
+        }
+        controller.enqueue(
+          encoder.encode(
+            encodeSseChunk({
+              id,
+              object: "chat.completion.chunk",
+              created,
+              model,
+              choices: [{ index: 0, delta: {}, finish_reason: "stop" }],
+            })
+          )
+        );
+        controller.enqueue(encoder.encode("data: [DONE]\n\n"));
+        controller.close();
+      },
+    });
+
     return new Response(stream, {
       headers: {
-        ...headers,
+        ...baseHeaders,
         "content-type": "text/event-stream; charset=utf-8",
         "cache-control": "no-cache",
         connection: "keep-alive",
       },
     });
   }
 
+  let turn;
+  try {
+    turn = await supervisorTurn(turnInput);
+  } catch (err) {
+    const message = err instanceof Error ? err.message : String(err);
+    return new Response(
+      JSON.stringify({
+        error: { message, type: "api_error" },
+      }),
+      { status: 500, headers: { "content-type": "application/json" } }
+    );
+  }
+
   const completion = buildCompletionResponse(body, turn.content);
   return new Response(JSON.stringify(completion), {
     headers: {
-      ...headers,
+      ...baseHeaders,
+      "x-ralph-supervisor-mode": turn.mode,
       "content-type": "application/json",
     },
   });

packages/provider/server/test/supervisor-agent.test.ts

@@ -2,7 +2,7 @@ import path from "node:path";
 import { mkdtemp, cp, rm } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { describe, expect, test, beforeEach, afterEach } from "bun:test";
-import { supervisorTurn } from "../lib/supervisor-agent.js";
+import { supervisorTurn, supervisorTurnStreaming } from "../lib/supervisor-agent.js";
 import { loopRegistry } from "../lib/loop-registry.js";
 import { getOpencodeRuntime } from "../lib/runtime.js";
 import { createProviderMockRuntime, mockSubscribe } from "./mock-runtime.js";
@@ -60,6 +60,22 @@ describe("supervisorTurn (RALPH_TEST_MODE)", () => {
     expect(engine?.state.attempt).toBe(1);
   });
 
+  test("supervisorTurnStreaming returns the same result and starts the loop", async () => {
+    const progress: string[] = [];
+    const turn = await supervisorTurnStreaming(
+      {
+        sessionKey: "sess-1",
+        worktree,
+        messages: [{ role: "user", content: "Implement marker file; tests must pass" }],
+      },
+      (text) => progress.push(text)
+    );
+
+    expect(turn.mode).toBe("test");
+    expect(turn.content.toLowerCase()).toContain("attempt 1");
+    expect(loopRegistry.get("sess-1")?.state.started).toBe(true);
+  });
+
   test("returns status on status request", async () => {
     await supervisorTurn({
       sessionKey: "sess-2",