Skip to content

fix: message.part.delta streaming, .info shape for tokens/finish, parts fallback for empty content #59

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
KochC merged 1 commit into from
Jul 5, 2026
Merged

fix: message.part.delta streaming, .info shape for tokens/finish, parts fallback for empty content #59

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
44 changes: 31 additions & 13 deletions index.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -776,19 +776,26 @@ async function runAgentTurn(client, model, messages, system, callerTools, onChun

try {
for await (const event of stream) {
if (event.type === "message.part.updated") {
if (event.type === "message.part.delta") {
// Real incremental token deltas arrive here, as flat properties (sessionID,
// partID, field, delta) - NOT nested under event.properties.part like
// message.part.updated below. This is the actual live-streaming source; the
// fallback via session.messages() after the loop covers turns where OpenCode
// doesn't emit these (see below).
const props = event.properties
if (
props?.sessionID === sessionID &&
props?.field === "text" &&
typeof props.delta === "string" &&
props.delta.length > 0
) {
content += props.delta
onChunk?.(props.delta)
}
} else if (event.type === "message.part.updated") {
const part = event.properties?.part
const delta = event.properties?.delta

if (
part?.sessionID === sessionID &&
part?.type === "text" &&
typeof delta === "string" &&
delta.length > 0
) {
content += delta
onChunk?.(delta)
} else if (
toolIDSet &&
part?.sessionID === sessionID &&
part?.type === "tool" &&
Expand Down Expand Up @@ -826,15 +833,26 @@ async function runAgentTurn(client, model, messages, system, callerTools, onChun
throw new Error(errorMessage)
}

// Each list item is { info: Message, parts: Part[] } - matching the shape
// client.session.prompt() (the non-tool-calling path) already returns directly.
const messagesResult = await client.session.messages({ path: { id: sessionID } })
const assistantMsg = (messagesResult.data ?? []).filter((m) => m.role === "assistant").at(-1)
const assistantEntry = (messagesResult.data ?? []).filter((m) => m.info?.role === "assistant").at(-1)
const assistantInfo = assistantEntry?.info

// Fallback for turns where message.part.delta never fired (observed for some
// multi-step turns, e.g. continuing a conversation with prior tool calls/results in
// history): use the authoritative final text from the fetched message's parts
// instead of leaving content empty.
if (!content && !toolCall) {
content = extractAssistantText(assistantEntry?.parts ?? [])
}

return {
sessionID,
content,
toolCall,
tokens: assistantMsg?.tokens ?? { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
finish: toolCall ? "tool_calls" : assistantMsg?.finish,
tokens: assistantInfo?.tokens ?? { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
finish: toolCall ? "tool_calls" : assistantInfo?.finish,
}
}

Expand Down
188 changes: 162 additions & 26 deletions index.test.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,9 +78,12 @@ function createStreamingClient(chunks) {
messages: async () => ({
data: [
{
role: "assistant",
tokens: { input: 10, output: 5, reasoning: 0, cache: { read: 0, write: 0 } },
finish: "end_turn",
info: {
role: "assistant",
tokens: { input: 10, output: 5, reasoning: 0, cache: { read: 0, write: 0 } },
finish: "end_turn",
},
parts: [],
},
],
}),
Expand Down Expand Up @@ -337,16 +340,18 @@ test("missing messages field returns 400", async () => {
test("stream: true returns SSE response", async () => {
const events = [
{
type: "message.part.updated",
type: "message.part.delta",
properties: {
part: { sessionID: "sess-123", type: "text" },
sessionID: "sess-123",
field: "text",
delta: "Hello",
},
},
{
type: "message.part.updated",
type: "message.part.delta",
properties: {
part: { sessionID: "sess-123", type: "text" },
sessionID: "sess-123",
field: "text",
delta: " world",
},
},
Expand Down Expand Up @@ -1094,16 +1099,18 @@ test("POST /v1/responses instructions field is incorporated", async () => {
test("POST /v1/responses stream: true returns SSE lifecycle events", async () => {
const events = [
{
type: "message.part.updated",
type: "message.part.delta",
properties: {
part: { sessionID: "sess-123", type: "text" },
sessionID: "sess-123",
field: "text",
delta: "The answer",
},
},
{
type: "message.part.updated",
type: "message.part.delta",
properties: {
part: { sessionID: "sess-123", type: "text" },
sessionID: "sess-123",
field: "text",
delta: " is 42.",
},
},
Expand Down Expand Up @@ -1137,16 +1144,18 @@ test("POST /v1/responses stream: true returns SSE lifecycle events", async () =>
test("POST /v1/responses stream: true emits content_part.done with accumulated text per OpenAI spec", async () => {
const events = [
{
type: "message.part.updated",
type: "message.part.delta",
properties: {
part: { sessionID: "sess-123", type: "text" },
sessionID: "sess-123",
field: "text",
delta: "The answer",
},
},
{
type: "message.part.updated",
type: "message.part.delta",
properties: {
part: { sessionID: "sess-123", type: "text" },
sessionID: "sess-123",
field: "text",
delta: " is 42.",
},
},
Expand Down Expand Up @@ -1635,16 +1644,18 @@ test("POST /v1/messages malformed JSON returns 400", async () => {
test("POST /v1/messages stream: true returns Anthropic SSE events", async () => {
const events = [
{
type: "message.part.updated",
type: "message.part.delta",
properties: {
part: { sessionID: "sess-123", type: "text" },
sessionID: "sess-123",
field: "text",
delta: "Hello",
},
},
{
type: "message.part.updated",
type: "message.part.delta",
properties: {
part: { sessionID: "sess-123", type: "text" },
sessionID: "sess-123",
field: "text",
delta: " world",
},
},
Expand Down Expand Up @@ -1832,16 +1843,18 @@ test("POST /v1beta/models/:model:generateContent malformed JSON returns 400", as
test("POST /v1beta/models/:model:streamGenerateContent returns NDJSON stream", async () => {
const events = [
{
type: "message.part.updated",
type: "message.part.delta",
properties: {
part: { sessionID: "sess-123", type: "text" },
sessionID: "sess-123",
field: "text",
delta: "Gem",
},
},
{
type: "message.part.updated",
type: "message.part.delta",
properties: {
part: { sessionID: "sess-123", type: "text" },
sessionID: "sess-123",
field: "text",
delta: "ini",
},
},
Expand Down Expand Up @@ -2099,9 +2112,12 @@ function createToolCallClient({ toolName, toolArgs, callID = "call_1", finish =
messages: async () => ({
data: [
{
role: "assistant",
tokens: { input: 5, output: 2, reasoning: 0, cache: { read: 0, write: 0 } },
finish,
info: {
role: "assistant",
tokens: { input: 5, output: 2, reasoning: 0, cache: { read: 0, write: 0 } },
finish,
},
parts: [],
},
],
}),
Expand Down Expand Up @@ -2228,6 +2244,126 @@ test("POST /v1/messages stream: true emits a tool_use content block", async () =
assert.ok(text.includes('"stop_reason":"tool_use"'))
})

// Regression test for: OpenCode delivers real incremental streaming text via
// message.part.delta events (flat properties: sessionID, partID, field, delta) - a
// completely separate event type from message.part.updated, which only carries status/
// snapshot updates (used here for tool-call detection). For some turns (observed with a
// multi-message conversation history, e.g. continuing after a prior tool call/result),
// OpenCode never emits message.part.delta at all for the final reply - only
// message.part.updated snapshots - so relying on message.part.delta alone would silently
// produce empty content. The fallback: after the loop, fetch the session's messages
// (client.session.messages()) and extract the final text from the assistant message's
// parts array directly, exactly as the non-tool-calling path already does via
// extractAssistantText().
//
// Each list item from client.session.messages() is `{ info: Message, parts: Part[] }` -
// info.role/info.tokens/info.finish, NOT flat role/tokens/finish directly on the item.
function createToolAwareTextClient({ events, assistantParts, tokens, finish }) {
return {
app: { log: async () => {} },
tool: { ids: async () => ({ data: [] }) },
config: {
providers: async () => ({
data: { providers: [{ id: "openai", models: { "gpt-4o": { id: "gpt-4o", name: "GPT-4o" } } }] },
}),
},
mcp: {
disconnect: async () => {
throw new Error("not connected")
},
add: async () => ({ data: {} }),
},
session: {
create: async () => ({ data: { id: "sess-text-1" } }),
promptAsync: async () => {},
abort: async () => ({ data: true }),
messages: async () => ({
data: [
{
info: { role: "assistant", tokens, finish },
parts: assistantParts,
},
],
}),
},
event: {
subscribe: async () => ({
stream: (async function* () {
for (const event of events) yield event
})(),
}),
},
}
}

test("POST /v1/chat/completions falls back to the final message's parts when message.part.delta never fires", async () => {
const client = createToolAwareTextClient({
events: [{ type: "session.idle", properties: { sessionID: "sess-text-1" } }],
assistantParts: [{ type: "step-start" }, { type: "text", text: "Agentic AI startups raise record funding" }],
tokens: { input: 42, output: 7, reasoning: 0, cache: { read: 0, write: 0 } },
finish: "stop",
})
const handler = createProxyFetchHandler(client)
const request = new Request("http://127.0.0.1:4010/v1/chat/completions", {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify({
model: "gpt-4o",
messages: [
{ role: "user", content: "Search the web then summarize." },
{
role: "assistant",
content: null,
tool_calls: [{ id: "call_1", type: "function", function: { name: "search", arguments: "{}" } }],
},
{ role: "tool", tool_call_id: "call_1", content: "some search result" },
],
tools: [{ type: "function", function: { name: "search" } }],
}),
})

const response = await handler(request)
const body = await response.json()

assert.equal(response.status, 200)
assert.equal(body.choices[0].message.content, "Agentic AI startups raise record funding")
assert.equal(body.choices[0].finish_reason, "stop")
// Also locks in the .info.tokens fix - these were always read as undefined (silently
// falling back to all-zero usage) before, since the flat .tokens field this code used
// to read doesn't exist on the real API's { info, parts } shape.
assert.equal(body.usage.prompt_tokens, 42)
assert.equal(body.usage.completion_tokens, 7)
})

test("POST /v1/chat/completions accumulates content from message.part.delta events", async () => {
const client = createToolAwareTextClient({
events: [
{ type: "message.part.delta", properties: { sessionID: "sess-text-1", field: "text", delta: "Hello" } },
{ type: "message.part.delta", properties: { sessionID: "sess-text-1", field: "text", delta: " world" } },
{ type: "session.idle", properties: { sessionID: "sess-text-1" } },
],
assistantParts: [{ type: "text", text: "Hello world" }],
tokens: { input: 1, output: 1, reasoning: 0, cache: { read: 0, write: 0 } },
finish: "stop",
})
const handler = createProxyFetchHandler(client)
const request = new Request("http://127.0.0.1:4010/v1/chat/completions", {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify({
model: "gpt-4o",
messages: [{ role: "user", content: "Say hello world." }],
tools: [{ type: "function", function: { name: "search" } }],
}),
})

const response = await handler(request)
const body = await response.json()

assert.equal(response.status, 200)
assert.equal(body.choices[0].message.content, "Hello world")
})

// Regression test for: a bridge slot reused by an earlier turn stays connected under its
// old tool schema (OpenCode has no MCP deregistration endpoint), and getDisabledTools()
// only snapshots OpenCode's built-in tool IDs once, before any bridge tools ever exist -
Expand Down