You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by wy...@apache.org on 2026/09/14 16:09:48 UTC
(maka) 16/20: fix(model): settle at the provider finish boundary
This is an automated email from the ASF dual-hosted git repository.
M4n5ter pushed a commit to branch feat/runtime-host-rust
in repository https://gitbox.apache.org/repos/asf/maka.git
commit d42650e71b4b3ca792c3ce0db0efb403ecd96e46
Author: Wang <m4...@gmail.com>
AuthorDate: Mon Sep 14 22:42:07 2026 +0800
fix(model): settle at the provider finish boundary
Follow upstream main request settlement without waiting for transport EOF or admitting late errors. Keep synthetic incomplete finishes rejected and release the SDK stream.
Generated-by: OpenAI Codex
---
crates/js-runtime/trusted/provider-errors.js | 9 +++----
crates/model/src/events.rs | 2 +-
crates/model/tests/fixtures/provider-errors.mjs | 29 ++++++++++++++++++++++-
crates/model/tests/integration/provider_stream.rs | 16 ++++++++++++-
docs/rust-runtime.md | 2 ++
docs/rust-runtime.zh-CN.md | 1 +
6 files changed, 52 insertions(+), 7 deletions(-)
diff --git a/crates/js-runtime/trusted/provider-errors.js b/crates/js-runtime/trusted/provider-errors.js
index 564ac72693..b7172c7c24 100644
--- a/crates/js-runtime/trusted/provider-errors.js
+++ b/crates/js-runtime/trusted/provider-errors.js
@@ -84,7 +84,6 @@ export function observesOutput(part) {
export async function forwardProviderStream(open, normalize, emit, kind) {
let observedOutput = false;
let replaySafe = true;
- let finished = false;
const truncated = () =>
emit({
type: 'error',
@@ -129,7 +128,7 @@ export async function forwardProviderStream(open, normalize, emit, kind) {
return await failed(error);
}
if (next.done) {
- if (!finished) await truncated();
+ await truncated();
return;
}
const raw = next.value;
@@ -146,9 +145,8 @@ export async function forwardProviderStream(open, normalize, emit, kind) {
}
// Capture before normalization, filtering or asynchronous delivery. A
// provider tool may already be running without a completed tool-call.
- if (raw.type === 'finish') finished = true;
if (
- finished ||
+ raw.type === 'finish' ||
raw.type === 'tool-result' ||
(['tool-input-start', 'tool-call'].includes(raw.type) &&
raw.providerExecuted !== undefined &&
@@ -165,6 +163,9 @@ export async function forwardProviderStream(open, normalize, emit, kind) {
part.timestamp = part.timestamp.toISOString();
}
await emit(part);
+ // The provider's terminal fact settles the response; transport EOF or a
+ // late connection error cannot revise it. finally releases the stream.
+ if (raw.type === 'finish') return;
}
} finally {
// Match for-await cleanup without replacing a known failure with cancel noise.
diff --git a/crates/model/src/events.rs b/crates/model/src/events.rs
index f0a3a5f496..cda3e9abef 100644
--- a/crates/model/src/events.rs
+++ b/crates/model/src/events.rs
@@ -317,7 +317,7 @@ mod tests {
String::from_utf8_lossy(&output.stderr)
);
let events: Vec<Value> = serde_json::from_slice(&output.stdout).unwrap();
- assert_eq!(events.len(), 8);
+ assert_eq!(events.len(), 7);
for (index, event) in events.iter().enumerate() {
let Err(ModelError::Provider(failure)) = Normalizer::default().push(event.clone())
else {
diff --git a/crates/model/tests/fixtures/provider-errors.mjs b/crates/model/tests/fixtures/provider-errors.mjs
index aed098fdd9..5607a0b9e9 100644
--- a/crates/model/tests/fixtures/provider-errors.mjs
+++ b/crates/model/tests/fixtures/provider-errors.mjs
@@ -35,7 +35,6 @@ const cases = [
[[{ type: 'tool-call', toolCallId: 'tool', providerExecuted: true }], false],
[[{ type: 'tool-result', toolCallId: 'tool' }], false],
[[{ ...text, providerMetadata: { anthropic: { signature: 'signed' } } }], false],
- [[{ type: 'finish' }], false],
];
const result = [];
for (const [parts, replaySafe] of cases) {
@@ -55,6 +54,34 @@ for (const [parts, replaySafe] of cases) {
assert.equal(failure.error.replaySafe, replaySafe);
result.push(failure);
}
+// A real finish is terminal even when the transport would later fail; cleanup
+// must still run. Synthetic Chat EOF is not a provider finish.
+for (const finish of [
+ { type: 'finish', finishReason: { unified: 'stop', raw: 'stop' } },
+ { type: 'finish', finishReason: { unified: 'other' } },
+]) {
+ const emitted = [];
+ let closed = false;
+ await forwardProviderStream(
+ async () => ({
+ stream: (async function* () {
+ try {
+ yield text;
+ yield finish;
+ throw new Error('must not read past the terminal boundary');
+ } finally {
+ closed = true;
+ }
+ })(),
+ }),
+ (part) => part,
+ async (part) => emitted.push(part),
+ 'openai_chat',
+ );
+ assert(closed);
+ if (finish.finishReason.raw) assert.deepEqual(emitted, [text, finish]);
+ else assert.equal(emitted.at(-1).error.reason, 'stream_truncated');
+}
const local = new Error('local emit failed');
await assert.rejects(
forwardProviderStream(
diff --git a/crates/model/tests/integration/provider_stream.rs b/crates/model/tests/integration/provider_stream.rs
index 0363f90c41..552a4d3152 100644
--- a/crates/model/tests/integration/provider_stream.rs
+++ b/crates/model/tests/integration/provider_stream.rs
@@ -132,15 +132,29 @@ async fn sdk_text_and_tool_streams_cross_real_http_before_response_finishes() {
let release = gate.clone();
let expected_text = "hello😀".repeat(2048);
let (first, last) = fixtures(kind.clone(), &expected_text);
+ let keep_open = matches!(kind, ProviderKind::Anthropic);
let server = tokio::spawn(async move {
let (mut socket, _) = listener.accept().await.unwrap();
let request = read_request(&mut socket).await;
- let header = format!("HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: {}\r\nConnection: close\r\nX-Latin-1: ÿ\r\n\r\n", first.len() + last.len());
+ let length = if keep_open {
+ String::new()
+ } else {
+ format!("Content-Length: {}\r\n", first.len() + last.len())
+ };
+ let header = format!("HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\n{length}Connection: close\r\nX-Latin-1: ÿ\r\n\r\n");
let header: Vec<u8> = header.chars().map(|ch| u8::try_from(ch).unwrap()).collect();
socket.write_all(&header).await.unwrap();
socket.write_all(first.as_bytes()).await.unwrap();
release.notified().await;
socket.write_all(last.as_bytes()).await.unwrap();
+ if keep_open {
+ let mut byte = [0];
+ let closed = tokio::time::timeout(Duration::from_secs(5), socket.read(&mut byte))
+ .await
+ .expect("provider finish must release HTTP without waiting for idle timeout")
+ .unwrap();
+ assert_eq!(closed, 0);
+ }
request
});
let mut customized = request(kind.clone(), base);
diff --git a/docs/rust-runtime.md b/docs/rust-runtime.md
index 99880676ca..a54a886842 100644
--- a/docs/rust-runtime.md
+++ b/docs/rust-runtime.md
@@ -115,6 +115,8 @@ Do not run untrusted code or point test instances at existing user data.
and unclassified network failures or idle timeouts are not retried.
Model activity refreshes the 120-second idle budget; active streams have no fixed
two-minute duration limit. User cancellation still closes and drains the request.
+ A real provider finish releases the stream without waiting for transport EOF;
+ synthetic finishes from truncated streams are not successful completions.
## Code layout
diff --git a/docs/rust-runtime.zh-CN.md b/docs/rust-runtime.zh-CN.md
index e8bca86dec..6c89b03733 100644
--- a/docs/rust-runtime.zh-CN.md
+++ b/docs/rust-runtime.zh-CN.md
@@ -93,6 +93,7 @@ Windows 使用私有 named pipe。可选 `--websocket 127.0.0.1:0` 监听需要
10 次，使用冻结输入与可取消退避。Provider 工具活动或重放元数据阻止重试；未知/本地
错误、尚未分类的网络故障与空闲超时不重试。
模型活动刷新 120 秒空闲预算，持续输出不受两分钟总时限限制；用户取消仍关闭请求并等待收尾。
+ 真实 provider finish 到达后释放流，不等待传输 EOF；残缺流合成的 finish 不算成功完成。
## 代码组织
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org