fix: use stream(values) for reliable agent streaming

coverage.txt

@@ -4,7 +4,7 @@
 ℹ ---------------------------------------------------------------------------------------------------------------------
 ℹ src                    |        |          |         | 
 ℹ  agent                 |        |          |         | 
-ℹ   react.js             | 100.00 |    64.29 |  100.00 | 
+ℹ   react.js             | 100.00 |    97.62 |  100.00 | 
 ℹ  config                |        |          |         | 
 ℹ   schemas.js           | 100.00 |   100.00 |  100.00 | 
 ℹ  memory                |        |          |         | 
@@ -40,6 +40,7 @@
 ℹ   common.js            | 100.00 |    93.33 |   83.33 | 
 ℹ   cron.js              | 100.00 |    98.65 |   90.00 | 
 ℹ   filesystem.js        |  94.50 |    86.79 |   79.17 | 44-45 107-110 170-177 187-188 196-202 397-398 415-419 422-423
+ℹ   hello.js             | 100.00 |   100.00 |  100.00 | 
 ℹ   image.js             |  97.90 |    95.83 |   50.00 | 92-94
 ℹ   index.js             | 100.00 |   100.00 |  100.00 | 
 ℹ   memory.js            |  97.48 |    92.31 |   85.71 | 37-40
@@ -58,6 +59,6 @@
 ℹ   messages.js          | 100.00 |    94.44 |  100.00 | 
 ℹ   panels.js            | 100.00 |   100.00 |  100.00 | 
 ℹ ---------------------------------------------------------------------------------------------------------------------
-ℹ all files              |  98.17 |    88.82 |   84.93 | 
+ℹ all files              |  98.17 |    90.45 |   84.98 | 
 ℹ ---------------------------------------------------------------------------------------------------------------------
 ℹ end of coverage report

src/agent/react.js

@@ -1,5 +1,5 @@
 import { createReactAgent as createReactAgentGraph } from "@langchain/langgraph/prebuilt";
-import { HumanMessage, SystemMessage, AIMessage } from "@langchain/core/messages";
+import { HumanMessage, SystemMessage, AIMessage, AIMessageChunk } from "@langchain/core/messages";
 
 /**
  * Create a ReAct agent from a chat model and optional tools and checkpointer.
@@ -56,70 +56,24 @@ export function callReactAgent(agent, message, config, systemPrompt, callback) {
 function extractContent(result, fallback) {
 	const msgsArray = Array.isArray(result.messages) ? result.messages : [];
 
-	const lastAI = [...msgsArray].reverse().find((msg) => msg instanceof AIMessage);
+	const lastAI = [...msgsArray]
+		.reverse()
+		.find((msg) => msg instanceof AIMessage || msg instanceof AIMessageChunk);
 	if (lastAI && lastAI.content) {
 		const content =
 			typeof lastAI.content === "string" ? lastAI.content : JSON.stringify(lastAI.content);
-		if (content.trim()) {
-			return { content: content.trim() };
+		const trimmed = content.trim();
+		if (trimmed && trimmed !== "[]" && trimmed !== "{}") {
+			return { content: trimmed };
 		}
 	}
 
 	return { content: fallback };
 }
 
 /**
- * Emits a tool event from a tools channel ProtocolEvent.
- * @param {ProtocolEvent} event
- * @param {(event: StreamEvent) => void} callback
- */
-function emitToolEvent(event, callback) {
-	const { data } = event.params;
-	if (!data || typeof data !== "object") return;
-
-	// Normalize tool event names for different LangGraph versions
-	const eventName = data.event || data.langgraph_event || "";
-
-	if (eventName === "on_tool_start" || eventName === "tool_called") {
-		callback({
-			type: "tool_start",
-			toolName: data.name || data.tool_name || data.tool || "",
-			toolCallId: data.toolCallId || data.tool_call_id,
-		});
-	} else if (
-		eventName === "on_tool_event" ||
-		eventName === "partial_result" ||
-		eventName === "tool_output"
-	) {
-		callback({
-			type: "tool_event",
-			toolCallId: data.toolCallId || data.tool_call_id,
-			data: data.data ?? data.output,
-		});
-	} else if (eventName === "on_tool_end" || eventName === "tool_finished") {
-		callback({
-			type: "tool_end",
-			toolName: data.name || data.tool_name || data.tool || "",
-			toolCallId: data.toolCallId || data.tool_call_id,
-			data: data.output ?? data.data,
-		});
-	} else if (
-		eventName === "on_tool_error" ||
-		eventName === "tool_error" ||
-		eventName === "partial_error"
-	) {
-		const errMsg = data.error || data.message || "Unknown error";
-		callback({
-			type: "tool_error",
-			toolName: data.name || data.tool_name || data.tool || "",
-			toolCallId: data.toolCallId || data.tool_call_id,
-			error: String(errMsg),
-		});
-	}
-}
-
-/**
- * Run the agent in streaming mode via LangGraph event streaming v3.
+ * Run the agent in streaming mode using state updates. Yields state snapshots
+ * after each step, extracting tool calls and final text from the messages array.
  * @param {ReturnType<typeof createReactAgentGraph>} agent - A compiled ReAct agent
  * @param {import("@langchain/core/messages").BaseMessage[]} initMessages - Initial messages
  * @param {string} originalMessage - Original user message (fallback)
@@ -128,48 +82,60 @@ function emitToolEvent(event, callback) {
  * @returns {{ content: string }} The agent's final text response
  */
 async function callReactAgentStreaming(agent, initMessages, originalMessage, config, callback) {
-	const streamOptions = {
-		version: "v3",
-		...(config?.configurable && { configurable: config.configurable }),
-	};
-	const stream = await agent.streamEvents({ messages: initMessages }, streamOptions);
-
-	let fullContent = "";
-	let hasContent = false;
-
-	// Consume text: iterate ChatModelStream instances and collect incremental text deltas
-	for await (const chatMessage of stream.messages) {
-		try {
-			let accumulated = "";
-			for await (const delta of chatMessage.text) {
-				accumulated += delta;
-				const trimmed = accumulated.trim();
-				if (trimmed) {
-					hasContent = true;
-					fullContent = trimmed;
-					callback({ type: "text", text: trimmed });
+	const stream = await agent.stream(
+		{ messages: initMessages },
+		{
+			streamMode: "values",
+			...(config?.configurable && { configurable: config.configurable }),
+		},
+	);
+
+	let toolCallSet = new Set();
+	let lastText = "";
+
+	for await (const chunk of stream) {
+		const msgs = chunk?.messages;
+		if (!Array.isArray(msgs)) continue;
+
+		for (const msg of msgs) {
+			if (!(msg instanceof AIMessage || msg instanceof AIMessageChunk)) continue;
+
+			// Check for tool calls
+			const toolCalls = msg.tool_calls || [];
+			for (const tc of toolCalls) {
+				const key = tc.name + "|" + tc.id;
+				if (!toolCallSet.has(key)) {
+					toolCallSet.add(key);
+					callback({ type: "tool_start", toolName: tc.name, toolCallId: tc.id });
 				}
 			}
-		} catch (_err) {
-			// Text projection may throw if the message had no text blocks; skip
-		}
-	}
 
-	// Consume tool and lifecycle events from the raw ProtocolEvent stream
-	for await (const event of stream) {
-		if (!event || !event.params) continue;
-		if (event.method === "tools") {
-			try {
-				emitToolEvent(event, callback);
-			} catch (_emitErr) {
-				// Callback error — don't break the streaming loop
+			// Extract text content from the message
+			let text = "";
+			if (typeof msg.content === "string") {
+				text = msg.content;
+			} else if (
+				typeof msg.content === "object" &&
+				msg.content !== null &&
+				typeof msg.content.text === "string"
+			) {
+				text = msg.content.text;
+			}
+			if (text.trim()) {
+				lastText = text.trim();
+				callback({ type: "text", text: lastText });
 			}
 		}
 	}
 
-	if (hasContent) {
-		return { content: fullContent };
+	// Emit tool_end for any pending tool calls
+	for (const key of toolCallSet) {
+		const [name] = key.split("|");
+		callback({ type: "tool_end", toolName: name });
 	}
 
+	if (lastText) {
+		return { content: lastText };
+	}
 	return { content: originalMessage };
 }

src/tools/hello.js

@@ -0,0 +1,39 @@
+import { tool } from "@langchain/core/tools";
+import { z } from "zod";
+
+/**
+ * Simple debug tool that always returns hello_world!
+ * @param {object} input - Tool input
+ * @param {object} _options - Runtime options
+ * @returns {Promise<string>} JSON string result
+ */
+export async function helloWorldImpl(_input, _options) {
+	return JSON.stringify({ result: "hello_world!" });
+}
+
+/**
+ * @param {z.infer<typeof HelloSchema>} input - Tool input
+ * @param {object} _options - Runtime options
+ * @returns {string} JSON result string
+ */
+export const hello_world = tool(helloWorldImpl, {
+	name: "hello_world",
+	description: "A simple debug tool that always returns 'hello_world!'",
+	schema: z.object({
+		name: z.string().optional().describe("Optional greeting target — e.g. 'world'"),
+	}),
+});
+
+/**
+ * Create a hello_world tool with runtime options
+ * @returns {object} LangChain Tool
+ */
+export function createHelloWorldTool() {
+	return tool(helloWorldImpl, {
+		name: "hello_world",
+		description: "A simple debug tool that returns 'hello_world!'",
+		schema: z.object({
+			name: z.string().optional().describe("Optional greeting target"),
+		}),
+	});
+}

src/tools/index.js

@@ -17,6 +17,7 @@ import { createCodeTool } from "./code.js";
 import { createCronTool } from "./cron.js";
 import { createTtsTool } from "./tts.js";
 import { createMoaTool } from "./moa.js";
+import { createHelloWorldTool } from "./hello.js";
 
 /**
  * Maps tool names to required permission scopes.
@@ -45,6 +46,7 @@ export const TOOL_PERMISSIONS = {
 	cronjob: ["network:outbound"],
 	text_to_speech: [], // requires OPENAI_API_KEY
 	mixture_of_agents: [], // requires OPENROUTER_API_KEY
+	hello_world: [], // debug tool, always available
 };
 
 // Factory functions keyed by tool name
@@ -69,6 +71,7 @@ const TOOL_FACTORIES = {
 	cronjob: createCronTool,
 	text_to_speech: createTtsTool,
 	mixture_of_agents: createMoaTool,
+	hello_world: createHelloWorldTool,
 };
 
 /**

src/tui/app.js

@@ -8,6 +8,7 @@ import { InputPanel } from "./inputPanel.js";
 import { isStreamingMessage } from "./messages.js";
 import { Banner } from "./banner.js";
 import { setConfigValue } from "../config/loader.js";
+import fs from "node:fs";
 
 const EXIT_MESSAGE = "\n";
 
@@ -123,6 +124,19 @@ export default function App({ config, registry, sessionState, dispatchProvider,
 				text,
 				sessionState ? sessionState.getProvider() : null,
 				(event) => {
+					try {
+						const cbData = {
+							type: event.type,
+							text: (event.text || "").slice(0, 80),
+							toolName: event.toolName || "",
+							toolCallId: event.toolCallId || "",
+							data: event.data,
+							error: event.error || "",
+						};
+						fs.appendFileSync("/tmp/madz_tui.log", JSON.stringify(cbData) + "\n");
+					} catch {
+						/* */
+					}
 					try {
 						if (event.type === "text") {
 							committedContent = event.text;

tests/unit/hello.test.js

@@ -0,0 +1,53 @@
+import { describe, it } from "node:test";
+import assert from "node:assert";
+import { helloWorldImpl, hello_world, createHelloWorldTool } from "../../src/tools/hello.js";
+import { buildToolConfig, TOOL_PERMISSIONS } from "../../src/tools/index.js";
+
+describe("hello_world tool", () => {
+	it("returns hello_world! as JSON", async () => {
+		const result = await helloWorldImpl({}, {});
+		const parsed = JSON.parse(result);
+		assert.strictEqual(parsed.result, "hello_world!");
+	});
+
+	it("returns consistent result regardless of input", async () => {
+		const result = await helloWorldImpl({ name: "everyone" }, {});
+		const parsed = JSON.parse(result);
+		assert.strictEqual(parsed.result, "hello_world!");
+	});
+
+	it("has correct tool name and description", async () => {
+		assert.strictEqual(hello_world.name, "hello_world");
+		assert.ok(hello_world.description.includes("hello_world"));
+	});
+
+	it("can be instantiated via factory", async () => {
+		const tool = createHelloWorldTool();
+		assert.strictEqual(tool.name, "hello_world");
+		assert.ok(typeof tool.call !== "undefined" || typeof tool.invoke !== "undefined");
+	});
+
+	it("has empty permission requirements", async () => {
+		assert.deepStrictEqual(TOOL_PERMISSIONS.hello_world, []);
+	});
+
+	it("is included in buildToolConfig with empty permissions", async () => {
+		const tools = await buildToolConfig({
+			permissions: [], // no permissions enabled
+			allowedPaths: [],
+			maxReadSize: "1mb",
+		});
+		const toolNames = tools.map((t) => t.name);
+		assert.ok(toolNames.includes("hello_world"));
+	});
+
+	it("is included in buildToolConfig with all permissions", async () => {
+		const tools = await buildToolConfig({
+			permissions: ["filesystem:read", "filesystem:write", "network:outbound"],
+			allowedPaths: [],
+			maxReadSize: "1mb",
+		});
+		const toolNames = tools.map((t) => t.name);
+		assert.ok(toolNames.includes("hello_world"));
+	});
+});