Skip to content

feat: adopt streamEvents v2 protocol for enhanced real-time TUI streaming #56

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
avoidwork merged 1 commit into from
May 31, 2026
Merged

feat: adopt streamEvents v2 protocol for enhanced real-time TUI streaming #56

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
82 changes: 79 additions & 3 deletions coverage.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
ℹ ---------------------------------------------------------------------------------------------------------------------
ℹ src | | | |
ℹ agent | | | |
ℹ react.js | 100.00 | 97.62 | 100.00 |
ℹ react.js | 100.00 | 97.87 | 100.00 |
ℹ config | | | |
ℹ schemas.js | 100.00 | 100.00 | 100.00 |
ℹ memory | | | |
Expand Down Expand Up @@ -54,12 +54,88 @@
ℹ web.js | 95.47 | 64.79 | 60.00 | 24-25 39-40 43-45 86-88 123-125 177 189-191 330-331
ℹ tui | | | |
ℹ commandParser.js | 100.00 | 88.00 | 100.00 |
ℹ conversationPanel.js | 95.51 | 96.00 | 80.00 | 242-247 252-257
ℹ conversationPanel.js | 89.08 | 92.59 | 80.00 | 110-121 125-132 268-273 278-283
ℹ inputPanel.js | 85.00 | 71.43 | 60.00 | 54-63 79-80
ℹ markdownText.js | 100.00 | 100.00 | 100.00 |
ℹ messages.js | 100.00 | 94.44 | 100.00 |
ℹ panels.js | 100.00 | 100.00 | 100.00 |
ℹ ---------------------------------------------------------------------------------------------------------------------
ℹ all files | 97.99 | 90.21 | 84.56 |
ℹ all files | 97.68 | 90.10 | 84.56 |
ℹ ---------------------------------------------------------------------------------------------------------------------
ℹ end of coverage report

✖ failing tests:

test at tests/unit/prompts.test.js:13:2
✖ returns the system prompt content (1.248951ms)
AssertionError [ERR_ASSERTION]: Expected values to be strictly equal:
+ actual - expected

+ '\n\nYou are a test assistant.'
- 'You are a test assistant.'

at TestContext.<anonymous> (file:///home/jason/Projects/madz/tests/unit/prompts.test.js:18:10)
at Test.runInAsyncScope (node:async_hooks:226:14)
at Test.run (node:internal/test_runner/test:1201:25)
at Test.start (node:internal/test_runner/test:1096:17)
at node:internal/test_runner/test:1617:71
at node:internal/per_context/primordials:466:82
at new Promise (<anonymous>)
at new SafePromise (node:internal/per_context/primordials:435:3)
at node:internal/per_context/primordials:466:9
at Array.map (<anonymous>) {
generatedMessage: true,
code: 'ERR_ASSERTION',
actual: '\n\nYou are a test assistant.',
expected: 'You are a test assistant.',
operator: 'strictEqual',
diff: 'simple'
}

test at tests/unit/prompts.test.js:21:2
✖ strips frontmatter from system prompt (0.274557ms)
AssertionError [ERR_ASSERTION]: Expected values to be strictly equal:
+ actual - expected

+ '\n\nHello world.'
- 'Hello world.'

at TestContext.<anonymous> (file:///home/jason/Projects/madz/tests/unit/prompts.test.js:26:10)
at Test.runInAsyncScope (node:async_hooks:226:14)
at Test.run (node:internal/test_runner/test:1201:25)
at Suite.processPendingSubtests (node:internal/test_runner/test:831:18)
at Test.postRun (node:internal/test_runner/test:1330:19)
at Test.run (node:internal/test_runner/test:1258:12)
at async Promise.all (index 0)
at async Suite.run (node:internal/test_runner/test:1619:7)
at async startSubtestAfterBootstrap (node:internal/test_runner/harness:385:3) {
generatedMessage: true,
code: 'ERR_ASSERTION',
actual: '\n\nHello world.',
expected: 'Hello world.',
operator: 'strictEqual',
diff: 'simple'
}

test at tests/unit/prompts.test.js:50:2
✖ handles truncated frontmatter (0.205637ms)
AssertionError [ERR_ASSERTION]: Expected values to be strictly equal:
+ actual - expected

+ '\njust broken frontmatter'
- '---\njust broken frontmatter'

at TestContext.<anonymous> (file:///home/jason/Projects/madz/tests/unit/prompts.test.js:55:10)
at Test.runInAsyncScope (node:async_hooks:226:14)
at Test.run (node:internal/test_runner/test:1201:25)
at Suite.processPendingSubtests (node:internal/test_runner/test:831:18)
at Test.postRun (node:internal/test_runner/test:1330:19)
at Test.run (node:internal/test_runner/test:1258:12)
at async Suite.processPendingSubtests (node:internal/test_runner/test:831:7) {
generatedMessage: true,
code: 'ERR_ASSERTION',
actual: '\njust broken frontmatter',
expected: '---\njust broken frontmatter',
operator: 'strictEqual',
diff: 'simple'
}
126 changes: 91 additions & 35 deletions src/agent/react.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,8 +72,8 @@ function extractContent(result, fallback) {
}

/**
* 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.
* Run the agent in streaming mode using the `streamEvents` API with v2 protocol.
* Yields granular events for text streaming, reasoning content, and tool execution.
* @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)
Expand All @@ -82,60 +82,116 @@ function extractContent(result, fallback) {
* @returns {{ content: string }} The agent's final text response
*/
async function callReactAgentStreaming(agent, initMessages, originalMessage, config, callback) {
const stream = await agent.stream(
const streamOptions = {
configurable: config?.configurable,
};

const stream = await agent.streamEvents(
{ messages: initMessages },
{
streamMode: "values",
...(config?.configurable && { configurable: config.configurable }),
},
{ version: "v2", ...streamOptions },
);

let toolCallSet = new Set();
let lastText = "";

for await (const chunk of stream) {
const msgs = chunk?.messages;
if (!Array.isArray(msgs)) continue;
for await (const event of stream) {
// Chat model text/reasoning streaming events
if (event.event === "on_chat_model_stream") {
const chunk = event.data?.chunk;
if (!chunk) continue;

// Track final text content from chat model stream
let textContent = "";
if (typeof chunk.content === "string") {
textContent = chunk.content;
} else if (
typeof chunk.content === "object" &&
chunk.content !== null &&
!Array.isArray(chunk.content) &&
chunk.content.text
) {
textContent = chunk.content.text;
}

for (const msg of msgs) {
if (!(msg instanceof AIMessage || msg instanceof AIMessageChunk)) continue;
// Emit text content deltas
if (Array.isArray(chunk.content)) {
for (const block of chunk.content) {
if (block.type === "text" && block.text && block.text.length > 0) {
textContent = block.text;
}
}
}
if (textContent.length > 0) {
// For tool-invoking LLM calls, the text might be empty or tool-call-related
callback({ type: "text", text: textContent });
// Note: the TUI accumulates text in committedContent for the final response,
// so we don't need to track it here.
}

// Check for tool calls
const toolCalls = msg.tool_calls || [];
// Emit reasoning/thinking content
if (chunk.reasoning) {
callback({ type: "reasoning", text: chunk.reasoning });
}
}

// Tool execution start
if (event.event === "on_tool_start" && event.name === "tool") {
const input = event.data?.input || {};
const toolCalls = Array.isArray(input.tool_calls) ? input.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 });
callback({
type: "tool_start",
toolName: tc.name || input.name || "unknown",
toolCallId: tc.id,
});
}
}
}

// 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 });
}
// Tool execution end with result
if (event.event === "on_tool_end" && event.name === "tool") {
const output = event.data?.output || {};
const input = event.data?.input || {};
const toolCalls = Array.isArray(input.tool_calls) ? input.tool_calls : [];
const toolName = input.name || toolCalls[0]?.name || output.tool_calls?.[0]?.name || "tool";
const toolCallId = toolCalls[0]?.id || "";
const resultData =
output.content || toolCalls[0]?.output || output.tool_calls?.[0]?.output || "";

callback({
type: "tool_end",
toolName,
toolCallId,
data: typeof resultData === "string" ? resultData.slice(0, 500) : resultData,
});
}

// Tool execution error
if (event.event === "on_tool_error" && event.name === "tool") {
const input = event.data?.input || {};
const toolCalls = Array.isArray(input.tool_calls) ? input.tool_calls : [];
const toolName = input.name || toolCalls[0]?.name || "unknown";
const toolCallId = toolCalls[0]?.id || "";
callback({
type: "tool_error",
toolName,
toolCallId,
error: event.data?.error,
});
}
}

// Emit tool_end for any pending tool calls
// Emit tool_end for any tool_start that didn't get a corresponding tool_end
// (e.g. if the stream was interrupted)
for (const key of toolCallSet) {
const [name] = key.split("|");
callback({ type: "tool_end", toolName: name });
}

if (lastText) {
return { content: lastText };
}
// Return originalMessage as fallback — the streaming callback
// accumulates the actual text in committedContent which is
// preferred by the TUI over this fallback value.
return { content: originalMessage };
}
2 changes: 2 additions & 0 deletions src/provider/openai.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import { ChatOpenAI } from "@langchain/openai";
* @property {string} credentials.apiKey - The API key for authentication
* @property {number} [temperature] - Sampling temperature (0-2)
* @property {number} [maxTokens] - Maximum output tokens
* @property {boolean} [streaming] - Enable streaming token output
*/

/**
Expand All @@ -23,6 +24,7 @@ export function createChatModel(config) {
temperature: config.temperature,
maxTokens: config.maxTokens,
apiKey: config.credentials.apiKey,
streaming: config.streaming !== false,
configuration: {
baseURL: config.base_url,
},
Expand Down
42 changes: 39 additions & 3 deletions src/tui/app.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -117,10 +117,12 @@ export default function App({ config, registry, sessionState, dispatchProvider,

let _currentToolCallCount = 0;
let committedContent = "";
let committedReasoning = "";
let lastToolCallDisplay = "";
let _activeToolCall = null;

try {
const response = await dispatchProvider(
const _response = await dispatchProvider(
text,
sessionState ? sessionState.getProvider() : null,
(event) => {
Expand All @@ -139,7 +141,7 @@ export default function App({ config, registry, sessionState, dispatchProvider,
}
try {
if (event.type === "text") {
committedContent = event.text;
committedContent = (committedContent || "") + event.text;
setMessages((prev) => {
const cloned = [...prev];
const last = cloned[cloned.length - 1];
Expand All @@ -148,8 +150,34 @@ export default function App({ config, registry, sessionState, dispatchProvider,
}
return cloned;
});
} else if (event.type === "reasoning") {
committedReasoning = (committedReasoning || "") + event.text;
setMessages((prev) => {
const cloned = [...prev];
const last = cloned[cloned.length - 1];
if (last.role === "assistant" && last.streaming) {
last.reasoningContent = (committedReasoning || "") + "\u2588";
}
return cloned;
});
} else if (event.type === "tool_start") {
activeToolCall = {
name: event.toolName,
toolCallId: event.toolCallId,
startedAt: Date.now(),
};
setMessages((prev) => {
const cloned = [...prev];
const last = cloned[cloned.length - 1];
if (last.role === "assistant" && last.streaming) {
last.activeToolCall = { name: event.toolName };
last.toolCallDisplay = lastToolCallDisplay;
}
return cloned;
});
} else if (event.type === "tool_end") {
_currentToolCallCount++;
activeToolCall = null;
const resultLine = event.data
? ` Result: ${JSON.stringify(event.data).slice(0, 200)}`
: "";
Expand All @@ -162,11 +190,13 @@ export default function App({ config, registry, sessionState, dispatchProvider,
const cloned = [...prev];
const last = cloned[cloned.length - 1];
if (last.role === "assistant" && last.streaming) {
last.activeToolCall = null;
last.toolCallDisplay = lastToolCallDisplay;
}
return cloned;
});
} else if (event.type === "tool_error") {
activeToolCall = null;
const errorLine = event.toolName
? `- Tool: ${event.toolName} (error: ${event.error})`
: `- Tool call failed (${event.toolCallId || "unknown"})`;
Expand All @@ -176,6 +206,7 @@ export default function App({ config, registry, sessionState, dispatchProvider,
const cloned = [...prev];
const last = cloned[cloned.length - 1];
if (last.role === "assistant" && last.streaming) {
last.activeToolCall = null;
last.toolCallDisplay = lastToolCallDisplay;
}
return cloned;
Expand All @@ -189,14 +220,19 @@ export default function App({ config, registry, sessionState, dispatchProvider,
},
);

const responseContent = response.content || committedContent || "";
// committedContent is accumulated from streaming text events —
// this is the actual AI response. response.content is only the
// originalMessage fallback from callReactAgentStreaming.
const responseContent = committedContent;

setMessages((prev) => {
const cloned = [...prev];
const last = cloned[cloned.length - 1];
if (last.role === "assistant" && last.streaming) {
last.content = responseContent;
last.reasoningContent = committedReasoning || undefined;
last.streaming = false;
last.activeToolCall = null;
if (lastToolCallDisplay) {
last.toolCallDisplay = lastToolCallDisplay;
}
Expand Down
Loading