From 2007a12a48484d3b9e4366d63591ed41875f32ad Mon Sep 17 00:00:00 2001 From: lloydzhou Date: Thu, 10 Sep 2026 19:23:55 +0800 Subject: [PATCH 01/10] =?UTF-8?q?feat(bash):=20=E5=9C=A8=E7=BB=88=E7=AB=AF?= =?UTF-8?q?=E6=A0=87=E9=A2=98=E6=A0=8F=E5=B1=95=E7=A4=BA=E6=9C=80=E8=BF=91?= =?UTF-8?q?=E4=B8=80=E6=AC=A1=20LLM=20call=20=E8=BE=93=E5=87=BA=E9=80=9F?= =?UTF-8?q?=E5=BA=A6=20(tok/s)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - claude_sse.awk 在 transport 层记录 start_ms/end_ms - USAGE 事件下沉时间字段，上层用 output_tokens/duration 计算速度 - stats.json 新增 last_call_speed_tok_per_sec - term_title.awk 标题栏增加 S:tok/s - 删除未使用的 util_date_ms - 更新 tests/test.sh async bash 标题栏 golden Tested: make test-bash (221 passed) --- src/agent.sh | 8 +++----- src/awk/claude_sse.awk | 14 +++++++++++++- src/awk/stats.awk | 5 +++-- src/awk/term_title.awk | 5 +++-- tests/test.sh | 2 +- 5 files changed, 23 insertions(+), 11 deletions(-) diff --git a/src/agent.sh b/src/agent.sh index f26d7ceb..8ee48311 100755 --- a/src/agent.sh +++ b/src/agent.sh @@ -1425,7 +1425,9 @@ agent_loop_stream() { ;; STOP) stop="${REPLY_MESSAGE[1]}" ;; ERROR) loop_error="${REPLY_MESSAGE[1]}"; stop="error"; break ;; - USAGE) _ctx_tokens=$(agent_record_usage "agent" agent_request_count false) ;; + USAGE) _ctx_tokens=$(agent_record_usage "agent" agent_request_count false) + local _dur _speed; _dur=$((${REPLY_MESSAGE[6]:-0}-${REPLY_MESSAGE[5]:-0})); _speed=$((_dur>0?${REPLY_MESSAGE[2]:-0}*1000/_dur:0)) + store_stats_update last_call_speed_tok_per_sec="$_speed" current_context_tokens="${_ctx_tokens:-0}" ;; esac done exec 8<&- @@ -1443,10 +1445,6 @@ agent_loop_stream() { if [[ -n "$tool_conv_results" ]]; then store_conv_add_tool_results "$tool_conv_results" fi - # Update context tokens from USAGE (used by next turn's compact check) - if [[ -n "$_ctx_tokens" && "$_ctx_tokens" -gt 0 ]]; then - store_stats_update current_context_tokens=${_ctx_tokens} - fi # tool_use/tool_calls → loop continues; otherwise exit unless a sub-agent result arrived meanwhile if [[ "$stop" != "tool_use" && "$stop" != "tool_calls" ]]; then agent_drain_notify_buf && continue diff --git a/src/awk/claude_sse.awk b/src/awk/claude_sse.awk index 79210b63..1f0cbf86 100644 --- a/src/awk/claude_sse.awk +++ b/src/awk/claude_sse.awk @@ -3,6 +3,16 @@ # Output: Unified protocol via emit1/emit/emit_flush # Requires: awk -v verbose=true/false -f json.awk -f protocol.awk -f todo_protocol.awk -f claude_sse.awk +function date_ms( cmd, ms) { + cmd = "perl -MTime::HiRes=time -e \047printf \042%d\\n\042, time * 1000\047" + if ((cmd | getline ms) > 0) { close(cmd); return ms + 0 } + close(cmd) + cmd = "date +%s%3N" + if ((cmd | getline ms) > 0) { close(cmd); return ms + 0 } + close(cmd) + return systime() * 1000 +} + BEGIN { event = "" block_type = "" @@ -15,6 +25,7 @@ BEGIN { cache_read_input_tokens = 0 cache_creation_input_tokens = 0 pending_stop_reason = "" + start_ms = date_ms() } /^:/ { next } @@ -130,7 +141,8 @@ BEGIN { END { if (pending_stop_reason != "") { - emit1("USAGE"); emit(pending_input_tokens + 0); emit(pending_output_tokens + 0); emit(pending_cache_read_tokens + 0); emit(pending_cache_creation_tokens + 0); emit_flush() + end_ms = date_ms() + emit1("USAGE"); emit(pending_input_tokens + 0); emit(pending_output_tokens + 0); emit(pending_cache_read_tokens + 0); emit(pending_cache_creation_tokens + 0); emit(start_ms + 0); emit(end_ms); emit_flush() emit1("STOP"); emit(pending_stop_reason); emit_flush() } else { # No message_stop received — emit an error STOP so the caller loop diff --git a/src/awk/stats.awk b/src/awk/stats.awk index 550e3051..f378bafd 100644 --- a/src/awk/stats.awk +++ b/src/awk/stats.awk @@ -65,8 +65,9 @@ function _init_fields() { _field_keys[7] = "total_cache_creation_tokens" _field_keys[8] = "current_context_tokens" _field_keys[9] = "sub_agent_request_count" - _field_keys[10] = "last_updated" - _field_count = 10 + _field_keys[10] = "last_call_speed_tok_per_sec" + _field_keys[11] = "last_updated" + _field_count = 11 for (i = 1; i <= _field_count; i++) { _vals[_field_keys[i]] = (_field_keys[i] == "last_updated") ? "" : 0 diff --git a/src/awk/term_title.awk b/src/awk/term_title.awk index 532c9555..8c2b1ca2 100644 --- a/src/awk/term_title.awk +++ b/src/awk/term_title.awk @@ -26,12 +26,13 @@ NR == 1 { o = jnum($0, "total_output_tokens") c = jnum($0, "current_context_tokens") cr = jnum($0, "total_cache_read_tokens") + spd = jnum($0, "last_call_speed_tok_per_sec") } END { prefix = (status == "idle") ? "" : "⏳ " progress = (status == "idle") ? 0 : 3 esc = sprintf("%c", 27) bel = sprintf("%c", 7) - printf "%s]0;%s%s T:%s R:%s I:%s(%s) O:%s C:%s%s%s]9;4;%d%s", \ - esc, prefix, model, fmt(t), fmt(r), fmt(i+cr), pct(cr, cr+i), fmt(o), fmt(c), bel, esc, progress, bel > "/dev/stderr" + printf "%s]0;%s%s T:%s R:%s I:%s(%s) O:%s C:%s S:%stok/s%s%s]9;4;%d%s", \ + esc, prefix, model, fmt(t), fmt(r), fmt(i+cr), pct(cr, cr+i), fmt(o), fmt(c), fmt(spd), bel, esc, progress, bel > "/dev/stderr" } diff --git a/tests/test.sh b/tests/test.sh index b5b0afd2..bf0b4c3f 100755 --- a/tests/test.sh +++ b/tests/test.sh @@ -718,7 +718,7 @@ test_agent_async_bash() { else red "Agent async bash"; echo " Output: $output"; ((FAIL++)) || true fi - if printf '%s' "$output" | grep -q $'\033]0;⏳ test T:[0-9,]* R:[0-9,]* I:[0-9,]*(0%) O:[0-9,]* C:[0-9,]*\a\033]9;4;3\a'; then + if printf '%s' "$output" | grep -q $'\033]0;⏳ test T:[0-9,]* R:[0-9,]* I:[0-9,]*(0%) O:[0-9,]* C:[0-9,]* S:[0-9,]*tok/s\a\033]9;4;3\a'; then green "Agent async bash title keeps loading while pending"; ((PASS++)) || true else red "Agent async bash title keeps loading while pending"; echo " Output: $output"; ((FAIL++)) || true From 386e2c23e3543c6df8cfc7a65922df767bb08e9e Mon Sep 17 00:00:00 2001 From: lloydzhou Date: Thu, 10 Sep 2026 20:15:24 +0800 Subject: [PATCH 02/10] =?UTF-8?q?feat(go,rust,c):=20=E5=90=8C=E6=AD=A5=20l?= =?UTF-8?q?ast=5Fcall=5Fspeed=5Ftok=5Fper=5Fsec=20=E5=88=B0=E4=B8=89?= =?UTF-8?q?=E7=89=88=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - transport 层记录 SSE 流 start/end 毫秒时间戳（读流开始→流结束，不含建连/TTFB） - agent 层 speed = output_tokens / duration_ms * 1000，duration>0 除零守卫 - stats.json 写入 last_call_speed_tok_per_sec，终端标题栏展示 S:xx tok/s - compact/sub_agent 的 usage 不更新 speed，口径与 bash 版一致 - tests/test.sh Test 40 增加 speed 字段检查 - 三版本 e2e 均 222 passed --- c/agent.c | 18 ++++++++++++++---- c/agent.h | 2 ++ c/store.c | 12 ++++++++---- c/transport.c | 31 ++++++++++++++++++++++++++++--- c/transport.h | 9 ++++++++- go/agent_test.go | 10 +++++----- go/store.go | 10 +++++++++- go/transport.go | 38 +++++++++++++++++++++++++++++++------- go/types.go | 27 +++++++++++++++------------ rust/src/agent.rs | 25 +++++++++++++++++++++++-- rust/src/lib.rs | 6 +++++- rust/src/sse.rs | 25 ++++++++++++++++++++++++- rust/src/store.rs | 7 ++++++- tests/test.sh | 9 ++++++++- 14 files changed, 186 insertions(+), 43 deletions(-) diff --git a/c/agent.c b/c/agent.c index 036e4374..3eeae244 100644 --- a/c/agent.c +++ b/c/agent.c @@ -1001,6 +1001,8 @@ int agent_loop(Agent *agent, const char *user_input, const char *turn_kind) { agent->last_cache_creation_tokens = accum->cache_creation_tokens; agent->last_context_tokens = accum->in_tokens + accum->out_tokens + accum->cache_read_tokens + accum->cache_creation_tokens; + agent->last_start_ms = accum->start_ms; + agent->last_end_ms = accum->end_ms; if (accum->in_tokens > 0 || accum->out_tokens > 0) { DisplayMessage *dm = malloc(sizeof(DisplayMessage)); @@ -1027,6 +1029,12 @@ int agent_loop(Agent *agent, const char *user_input, const char *turn_kind) { store_stats_set_int_file(agent->paths.stats, "current_context_tokens", agent->last_context_tokens); } + /* 对齐 bash 版：speed = output_tokens / (end_ms - start_ms) * 1000，duration > 0 */ + { + long long dur = agent->last_end_ms - agent->last_start_ms; + int speed = (dur > 0) ? (int)((long long)agent->last_output_tokens * 1000LL / dur) : 0; + store_stats_set_int_file(agent->paths.stats, "last_call_speed_tok_per_sec", speed); + } /* 对齐 bash 版 store_stats_update 末尾的 display_term_title */ agent_update_title(agent); } @@ -3019,27 +3027,29 @@ void agent_update_title_status(Agent *agent, const char *status) { int ao = json_get_int(jp.val, "total_output_tokens"); long long ll_cr = json_get_ll(jp.val, "total_cache_read_tokens"); int ct = json_get_int(jp.val, "current_context_tokens"); + int speed = json_get_int(jp.val, "last_call_speed_tok_per_sec"); - /* 对齐 bash 版 term_title.awk: model T:turn R:req I:in+cr(pct) O:out C:ctx */ + /* 对齐 bash 版 term_title.awk: model T:turn R:req I:in+cr(pct) O:out C:ctx S:speedtok/s */ long long total_i = ll_ai + ll_cr; int pct = (total_i > 0) ? (int)((ll_cr * 100.0 / total_i) + 0.5) : 0; char pct_s[32]; if (total_i > 0) snprintf(pct_s, sizeof(pct_s), "%d%%", pct); else snprintf(pct_s, sizeof(pct_s), "—"); - char tc_s[32], ar_s[32], total_i_s[32], ao_s[32], ct_s[32]; + char tc_s[32], ar_s[32], total_i_s[32], ao_s[32], ct_s[32], speed_s[32]; format_int_commas(tc, tc_s, sizeof(tc_s)); format_int_commas(ar, ar_s, sizeof(ar_s)); format_ll_commas(total_i, total_i_s, sizeof(total_i_s)); format_int_commas(ao, ao_s, sizeof(ao_s)); format_int_commas(ct, ct_s, sizeof(ct_s)); + format_int_commas(speed, speed_s, sizeof(speed_s)); int idle = (status && strcmp(status, "idle") == 0 && agent->active_task_count <= 0); const char *prefix = idle ? "" : "\xe2\x8f\xb3 "; int progress = idle ? 0 : 3; FILE *err = agent->err ? agent->err : stderr; - fprintf(err, "\x1b]0;%s%s T:%s R:%s I:%s(%s) O:%s C:%s\x07\x1b]9;4;%d\x07", - prefix, agent->model, tc_s, ar_s, total_i_s, pct_s, ao_s, ct_s, progress); + fprintf(err, "\x1b]0;%s%s T:%s R:%s I:%s(%s) O:%s C:%s S:%stok/s\x07\x1b]9;4;%d\x07", + prefix, agent->model, tc_s, ar_s, total_i_s, pct_s, ao_s, ct_s, speed_s, progress); fflush(err); free(stats_content); diff --git a/c/agent.h b/c/agent.h index b22e2507..38b89023 100644 --- a/c/agent.h +++ b/c/agent.h @@ -77,6 +77,8 @@ typedef struct { int last_output_tokens; int last_cache_read_tokens; int last_cache_creation_tokens; + long long last_start_ms; + long long last_end_ms; int active_task_count; int sub_agent_depth; /* 主代理为 0，第一层子代理为 1 */ diff --git a/c/store.c b/c/store.c index ddef6fc3..5a3fd933 100644 --- a/c/store.c +++ b/c/store.c @@ -152,7 +152,7 @@ int store_session_init(const SessionPaths *p, int is_new) { "\"total_input_tokens\":0," "\"total_output_tokens\":0,\"total_cache_read_tokens\":0," "\"total_cache_creation_tokens\":0,\"current_context_tokens\":0," - "\"last_updated\":\"\"}\n"); + "\"last_call_speed_tok_per_sec\":0,\"last_updated\":\"\"}\n"); fclose(f); /* 写入 session_start 事件（与 bash 版对齐） */ @@ -543,6 +543,7 @@ static void stats_write_canonical(const char *path, int total_cache_read_tokens, int total_cache_creation_tokens, int current_context_tokens, + int last_call_speed_tok_per_sec, const char *last_updated) { StrBuf buf; sb_init(&buf); @@ -550,11 +551,11 @@ static void stats_write_canonical(const char *path, "\"compact_request_count\":%d,\"sub_agent_request_count\":%d," "\"total_input_tokens\":%d,\"total_output_tokens\":%d," "\"total_cache_read_tokens\":%d,\"total_cache_creation_tokens\":%d," - "\"current_context_tokens\":%d,\"last_updated\":", + "\"current_context_tokens\":%d,\"last_call_speed_tok_per_sec\":%d,\"last_updated\":", current_turn_count, agent_request_count, compact_request_count, sub_agent_request_count, total_input_tokens, total_output_tokens, total_cache_read_tokens, total_cache_creation_tokens, - current_context_tokens); + current_context_tokens, last_call_speed_tok_per_sec); sb_append_json_string(&buf, last_updated ? last_updated : ""); sb_append(&buf, "}\n"); util_write_file(path, buf.data); @@ -617,6 +618,7 @@ void store_stats_set_int_file(const char *path, const char *key, int value) { int total_input_tokens = 0, total_output_tokens = 0; int total_cache_read_tokens = 0, total_cache_creation_tokens = 0; int current_context_tokens = 0; + int last_call_speed_tok_per_sec = 0; char *content = store_stats_read(path); if (content && content[0]) { @@ -631,6 +633,7 @@ void store_stats_set_int_file(const char *path, const char *key, int value) { total_cache_read_tokens = json_get_int(jp.val, "total_cache_read_tokens"); total_cache_creation_tokens = json_get_int(jp.val, "total_cache_creation_tokens"); current_context_tokens = json_get_int(jp.val, "current_context_tokens"); + last_call_speed_tok_per_sec = json_get_int(jp.val, "last_call_speed_tok_per_sec"); } } @@ -643,6 +646,7 @@ void store_stats_set_int_file(const char *path, const char *key, int value) { else if (strcmp(key, "total_cache_read_tokens") == 0) total_cache_read_tokens = value; else if (strcmp(key, "total_cache_creation_tokens") == 0) total_cache_creation_tokens = value; else if (strcmp(key, "current_context_tokens") == 0) current_context_tokens = value; + else if (strcmp(key, "last_call_speed_tok_per_sec") == 0) last_call_speed_tok_per_sec = value; time_t now = time(NULL); struct tm tm_buf; @@ -653,7 +657,7 @@ void store_stats_set_int_file(const char *path, const char *key, int value) { compact_request_count, sub_agent_request_count, total_input_tokens, total_output_tokens, total_cache_read_tokens, total_cache_creation_tokens, - current_context_tokens, ts); + current_context_tokens, last_call_speed_tok_per_sec, ts); if (content) { free(content); } diff --git a/c/transport.c b/c/transport.c index c1879d80..a902a1c6 100644 --- a/c/transport.c +++ b/c/transport.c @@ -5,8 +5,15 @@ #include #include #include +#include #include +static long long now_ms(void) { + struct timeval tv; + gettimeofday(&tv, NULL); + return (long long)tv.tv_sec * 1000LL + tv.tv_usec / 1000; +} + /* ============================================================ * libcurl 回调 * ============================================================ */ @@ -59,6 +66,7 @@ typedef struct { int *responses_item_indexes; int responses_item_count; int responses_item_cap; + long long start_ms; /* 流开始时间戳（毫秒），对齐 bash 版 */ } StreamCtx; static void emit_simple_event(sse_callback_fn callback, void *ctx, @@ -207,6 +215,8 @@ static void parse_openai_sse_event(StreamCtx *sctx, const char *data, size_t dat memset(&evt, 0, sizeof(evt)); evt.type = SSE_USAGE; fill_openai_usage_event(&evt, usage); + evt.start_ms = sctx->start_ms; + evt.end_ms = now_ms(); sctx->callback(sctx->ctx, &evt); } } @@ -261,6 +271,8 @@ static void responses_emit_usage(StreamCtx *sctx) { evt.in_tokens = sctx->responses_input_tokens; evt.out_tokens = sctx->responses_output_tokens; evt.cache_read_tokens = sctx->responses_cache_read_tokens; + evt.start_ms = sctx->start_ms; + evt.end_ms = now_ms(); sctx->callback(sctx->ctx, &evt); } @@ -361,13 +373,13 @@ static size_t stream_cb(char *ptr, size_t size, size_t nmemb, void *userdata) { const char *data = line + 6; if (strcmp(sctx->provider, "openai") == 0) parse_openai_sse_event(sctx, data, strlen(data)); else if (strcmp(sctx->provider, "responses") == 0) parse_responses_sse_event(sctx, sctx->event ? sctx->event : "", data, strlen(data)); - else sse_parse_event(sctx->provider, data, strlen(data), sctx->callback, sctx->ctx); + else sse_parse_event(sctx->provider, data, strlen(data), sctx->callback, sctx->ctx, sctx->start_ms); } else if (strncmp(line, "data:", 5) == 0) { const char *data = line + 5; while (*data == ' ') data++; if (strcmp(sctx->provider, "openai") == 0) parse_openai_sse_event(sctx, data, strlen(data)); else if (strcmp(sctx->provider, "responses") == 0) parse_responses_sse_event(sctx, sctx->event ? sctx->event : "", data, strlen(data)); - else sse_parse_event(sctx->provider, data, strlen(data), sctx->callback, sctx->ctx); + else sse_parse_event(sctx->provider, data, strlen(data), sctx->callback, sctx->ctx, sctx->start_ms); } /* 重置行缓冲 */ sb_truncate(&sctx->line_buf, 0); @@ -510,6 +522,8 @@ static void process_residual_json(StreamCtx *sctx, const char *provider, evt.out_tokens = json_get_int(usage, "output_tokens"); evt.cache_read_tokens = json_get_int(usage, "cache_read_input_tokens"); evt.cache_creation_tokens = json_get_int(usage, "cache_creation_input_tokens"); + evt.start_ms = sctx->start_ms; + evt.end_ms = now_ms(); callback(ctx, &evt); } } else { @@ -563,6 +577,8 @@ static void process_residual_json(StreamCtx *sctx, const char *provider, memset(&evt, 0, sizeof(evt)); evt.type = SSE_USAGE; fill_openai_usage_event(&evt, usage); + evt.start_ms = sctx->start_ms; + evt.end_ms = now_ms(); callback(ctx, &evt); } } @@ -580,6 +596,7 @@ int http_post_sse(const char *url, const char **headers, int header_count, struct timespec start_ts; clock_gettime(CLOCK_MONOTONIC, &start_ts); + long long start_ms = now_ms(); struct curl_slist *hdrs = NULL; for (int i = 0; i < header_count; i++) { @@ -600,6 +617,7 @@ int http_post_sse(const char *url, const char **headers, int header_count, sb_init(&sctx.line_buf); sctx.cancelled = cancelled; sctx.provider = (char *)provider; + sctx.start_ms = start_ms; curl_easy_setopt(curl, CURLOPT_URL, url); curl_easy_setopt(curl, CURLOPT_POST, 1L); @@ -716,7 +734,8 @@ static char *dup_and_free(char *s) { #endif int sse_parse_event(const char *provider, const char *data, size_t data_len, - sse_callback_fn callback, void *ctx) { + sse_callback_fn callback, void *ctx, + long long start_ms) { if (data_len == 0) return 0; if (strcmp(data, "[DONE]") == 0) { if (strcmp(provider, "claude") == 0) emit_simple_event(callback, ctx, SSE_STOP, "end_turn"); @@ -796,6 +815,8 @@ int sse_parse_event(const char *provider, const char *data, size_t data_len, if (it > 0) evt.in_tokens = it; if (cr > 0) evt.cache_read_tokens = cr; if (cc > 0) evt.cache_creation_tokens = cc; + evt.start_ms = start_ms; + evt.end_ms = now_ms(); callback(ctx, &evt); } } else if (strcmp(type, "message_start") == 0) { @@ -808,6 +829,8 @@ int sse_parse_event(const char *provider, const char *data, size_t data_len, evt.in_tokens = json_get_int(usage, "input_tokens"); evt.cache_read_tokens = json_get_int(usage, "cache_read_input_tokens"); evt.cache_creation_tokens = json_get_int(usage, "cache_creation_input_tokens"); + evt.start_ms = start_ms; + evt.end_ms = now_ms(); callback(ctx, &evt); } } else if (strcmp(type, "error") == 0) { @@ -979,6 +1002,8 @@ void sse_accum_callback(void *ctx, const SseEvent *evt) { if (evt->out_tokens > 0) acc->out_tokens = evt->out_tokens; if (evt->cache_read_tokens > 0) acc->cache_read_tokens = evt->cache_read_tokens; if (evt->cache_creation_tokens > 0) acc->cache_creation_tokens = evt->cache_creation_tokens; + acc->start_ms = evt->start_ms; + acc->end_ms = evt->end_ms; break; case SSE_STOP: diff --git a/c/transport.h b/c/transport.h index 3cbcdc80..6bcff7bd 100644 --- a/c/transport.h +++ b/c/transport.h @@ -34,6 +34,8 @@ typedef struct { int out_tokens; /* USAGE: 输出 token */ int cache_read_tokens; /* USAGE: 缓存读取 token */ int cache_creation_tokens; /* USAGE: 缓存创建 token */ + long long start_ms; /* USAGE: 流开始时间戳（毫秒） */ + long long end_ms; /* USAGE: 流结束时间戳（毫秒） */ } SseEvent; typedef void (*sse_callback_fn)(void *ctx, const SseEvent *evt); @@ -60,7 +62,8 @@ int http_post_sse(const char *url, const char **headers, int header_count, /* 解析 SSE 事件行（从 HTTP 响应体的 "data: ..." 行解析） */ int sse_parse_event(const char *provider, const char *data, size_t data_len, - sse_callback_fn callback, void *ctx); + sse_callback_fn callback, void *ctx, + long long start_ms); /* ============================================================ * SSE 累积器 — 用于 agent_loop 中收集流式事件 @@ -95,6 +98,10 @@ typedef struct { int cache_read_tokens; int cache_creation_tokens; + /* 流时间戳（毫秒），用于计算 tok/s */ + long long start_ms; + long long end_ms; + /* 停止原因 */ char *stop_reason; diff --git a/go/agent_test.go b/go/agent_test.go index f6d3c7dc..b014f271 100644 --- a/go/agent_test.go +++ b/go/agent_test.go @@ -873,14 +873,14 @@ func TestResponsesTransportBodyAndSSE(t *testing.T) { ch := make(chan Event, 8) pending := map[int]*responsesPendingCall{} indexes := map[string]int{} - if tr.handleResponsesEvent("response.output_item.added", `{"output_index":2,"item":{"id":"item_1","type":"function_call","call_id":"call_1","name":"Read"}}`, ch, pending, indexes, new(bool), new(int), new(int), new(int)) { + if tr.handleResponsesEvent("response.output_item.added", `{"output_index":2,"item":{"id":"item_1","type":"function_call","call_id":"call_1","name":"Read"}}`, ch, pending, indexes, new(bool), new(int), new(int), new(int), 0) { t.Fatal("tool item must not terminate the stream") } - if tr.handleResponsesEvent("response.function_call_arguments.delta", `{"item_id":"item_1","delta":"{\"path\":\"/tmp/a\"}"}`, ch, pending, indexes, new(bool), new(int), new(int), new(int)) { + if tr.handleResponsesEvent("response.function_call_arguments.delta", `{"item_id":"item_1","delta":"{\"path\":\"/tmp/a\"}"}`, ch, pending, indexes, new(bool), new(int), new(int), new(int), 0) { t.Fatal("argument delta must not terminate the stream") } input, output, cached := 0, 0, 0 - if !tr.handleResponsesEvent("response.completed", `{"response":{"usage":{"input_tokens":15,"output_tokens":8,"cached_tokens":4,"input_tokens_details":{"cached_tokens":6}}}}`, ch, pending, indexes, new(bool), &input, &output, &cached) { + if !tr.handleResponsesEvent("response.completed", `{"response":{"usage":{"input_tokens":15,"output_tokens":8,"cached_tokens":4,"input_tokens_details":{"cached_tokens":6}}}}`, ch, pending, indexes, new(bool), &input, &output, &cached, 0) { t.Fatal("completed must terminate the stream") } close(ch) @@ -894,7 +894,7 @@ func TestResponsesTransportBodyAndSSE(t *testing.T) { } failure := make(chan Event, 3) - if !tr.handleResponsesEvent("error", `{"reason":"upstream failed"}`, failure, map[int]*responsesPendingCall{}, map[string]int{}, new(bool), new(int), new(int), new(int)) { + if !tr.handleResponsesEvent("error", `{"reason":"upstream failed"}`, failure, map[int]*responsesPendingCall{}, map[string]int{}, new(bool), new(int), new(int), new(int), 0) { t.Fatal("error must terminate the stream") } close(failure) @@ -907,7 +907,7 @@ func TestResponsesTransportBodyAndSSE(t *testing.T) { } bare := make(chan Event, 3) - if !tr.handleResponsesEvent("error", `{}`, bare, map[int]*responsesPendingCall{}, map[string]int{}, new(bool), new(int), new(int), new(int)) { + if !tr.handleResponsesEvent("error", `{}`, bare, map[int]*responsesPendingCall{}, map[string]int{}, new(bool), new(int), new(int), new(int), 0) { t.Fatal("bare error must terminate the stream") } close(bare) diff --git a/go/store.go b/go/store.go index b28bfece..b5121b92 100644 --- a/go/store.go +++ b/go/store.go @@ -55,6 +55,12 @@ func (s *FileStore) UpdateStats(usage Usage, model string) error { s.stats.OutputTokens += usage.OutputTokens s.stats.CacheWrite += usage.CacheWrite s.stats.CacheRead += usage.CacheRead + // 对齐 bash 版：speed = output_tokens / (end_ms - start_ms) * 1000，duration > 0 + if usage.EndMs > usage.StartMs { + s.stats.LastCallSpeedTokPerSec = usage.OutputTokens * 1000 / int(usage.EndMs-usage.StartMs) + } else { + s.stats.LastCallSpeedTokPerSec = 0 + } return s.flushStats() } @@ -799,7 +805,8 @@ func (s *FileStore) FormatTitle(model, status string) string { if status == "idle" { progress = 0 } - return fmt.Sprintf("\x1b]0;%s%s T:%s R:%s I:%s(%s) O:%s C:%s\x07\x1b]9;4;%d\x07", + // 对齐 bash 版 term_title.awk：model T:turn R:req I:in+cr(pct) O:out C:ctx S:speedtok/s + return fmt.Sprintf("\x1b]0;%s%s T:%s R:%s I:%s(%s) O:%s C:%s S:%stok/s\x07\x1b]9;4;%d\x07", prefix, model, fmtInt(st.TurnCount), @@ -808,6 +815,7 @@ func (s *FileStore) FormatTitle(model, status string) string { cachePct, fmtInt(st.OutputTokens), fmtInt(st.ContextTokens), + fmtInt(st.LastCallSpeedTokPerSec), progress) } diff --git a/go/transport.go b/go/transport.go index eca356b5..977f1460 100644 --- a/go/transport.go +++ b/go/transport.go @@ -196,6 +196,10 @@ func (t *HTTPTransport) parseSSEStream(ctx context.Context, resp *http.Response, } }() + // 对齐 bash 版 claude_sse.awk：流开始时刻记录 start_ms，结束时算 speed + startMs := time.Now().UnixMilli() + endMs := func() int64 { return time.Now().UnixMilli() } + scanner := bufio.NewScanner(resp.Body) scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) @@ -227,7 +231,8 @@ func (t *HTTPTransport) parseSSEStream(ctx context.Context, resp *http.Response, if t.cfg.Provider == "responses" && (strings.HasPrefix(eventType, "response.") || eventType == "error") { if t.handleResponsesEvent(eventType, data, ch, responsesPendingCalls, responsesItemIndexes, - &responsesTextStarted, &inputTokens, &outputTokens, &cacheRead) { + &responsesTextStarted, &inputTokens, &outputTokens, &cacheRead, + startMs) { stopEmitted = true } continue @@ -239,7 +244,8 @@ func (t *HTTPTransport) parseSSEStream(ctx context.Context, resp *http.Response, openaiPendingCalls, &openaiTextStarted, &stopReason, - &inputTokens, &outputTokens, &cacheRead, &cacheCreate) + &inputTokens, &outputTokens, &cacheRead, &cacheCreate, + startMs) if data == "[DONE]" { stopEmitted = true } @@ -287,12 +293,14 @@ func (t *HTTPTransport) parseSSEStream(ctx context.Context, resp *http.Response, case "message_stop": stopEmitted = true - // 发送 USAGE + STOP + // 发送 USAGE + STOP（对齐 bash 版：带 start_ms/end_ms 用于计算 tok/s） ch <- Event{Type: EventUsage, Payload: Usage{ InputTokens: inputTokens, OutputTokens: outputTokens, CacheRead: cacheRead, CacheWrite: cacheCreate, + StartMs: startMs, + EndMs: endMs(), }} ch <- Event{Type: EventStop, Fields: []string{"STOP", stopReason}} @@ -360,7 +368,8 @@ func emitResponsesPendingCalls(ch chan<- Event, pending map[int]*responsesPendin // 返回 true 表示该事件已经终止当前响应。 func (t *HTTPTransport) handleResponsesEvent(eventType, data string, ch chan<- Event, pending map[int]*responsesPendingCall, itemIndexes map[string]int, textStarted *bool, - inputTokens, outputTokens, cacheRead *int) bool { + inputTokens, outputTokens, cacheRead *int, + startMs int64) bool { var payload struct { Delta string `json:"delta"` OutputIndex int `json:"output_index"` @@ -486,7 +495,13 @@ func (t *HTTPTransport) handleResponsesEvent(eventType, data string, ch chan<- E recordUsage() hasTools := len(pending) > 0 emitResponsesPendingCalls(ch, pending) - ch <- Event{Type: EventUsage, Payload: Usage{InputTokens: *inputTokens, OutputTokens: *outputTokens, CacheRead: *cacheRead}} + ch <- Event{Type: EventUsage, Payload: Usage{ + InputTokens: *inputTokens, + OutputTokens: *outputTokens, + CacheRead: *cacheRead, + StartMs: startMs, + EndMs: time.Now().UnixMilli(), + }} stopReason := "end_turn" if hasTools { stopReason = "tool_use" @@ -521,7 +536,13 @@ func (t *HTTPTransport) handleResponsesEvent(eventType, data string, ch chan<- E } } ch <- Event{Type: EventError, Fields: []string{"ERROR", message}} - ch <- Event{Type: EventUsage, Payload: Usage{InputTokens: *inputTokens, OutputTokens: *outputTokens, CacheRead: *cacheRead}} + ch <- Event{Type: EventUsage, Payload: Usage{ + InputTokens: *inputTokens, + OutputTokens: *outputTokens, + CacheRead: *cacheRead, + StartMs: startMs, + EndMs: time.Now().UnixMilli(), + }} ch <- Event{Type: EventStop, Fields: []string{"STOP", "error"}} return true } @@ -551,7 +572,8 @@ func (t *HTTPTransport) handleOpenAIChunk(data string, ch chan<- Event, pending map[int]*openAIPendingCall, textStarted *bool, stopReason *string, - inputTokens, outputTokens, cacheRead, cacheCreate *int) { + inputTokens, outputTokens, cacheRead, cacheCreate *int, + startMs int64) { if data == "[DONE]" { emitOpenAIPendingCalls(ch, pending) @@ -570,6 +592,8 @@ func (t *HTTPTransport) handleOpenAIChunk(data string, ch chan<- Event, OutputTokens: *outputTokens, CacheRead: *cacheRead, CacheWrite: *cacheCreate, + StartMs: startMs, + EndMs: time.Now().UnixMilli(), }} ch <- Event{Type: EventStop, Fields: []string{"STOP", sr}} return diff --git a/go/types.go b/go/types.go index 0483f62e..e134c369 100644 --- a/go/types.go +++ b/go/types.go @@ -120,9 +120,11 @@ const ( type Usage struct { InputTokens int OutputTokens int - CacheRead int // CacheReadInputTokens - CacheWrite int // CacheCreationInputTokens + CacheRead int // CacheReadInputTokens + CacheWrite int // CacheCreationInputTokens Cost float64 + StartMs int64 // 流开始时间戳（毫秒） + EndMs int64 // 流结束时间戳（毫秒） } // ToolCallInfo 表示一个工具调用 @@ -143,16 +145,17 @@ type ToolResultInfo struct { // Stats 会话统计数据 type Stats struct { - TurnCount int `json:"current_turn_count"` // user turn count - TotalRequests int `json:"agent_request_count"` // total LLM requests - TotalCompact int `json:"compact_request_count"` // number of compactions - SubAgentRequests int `json:"sub_agent_request_count"` // sub-agent requests - InputTokens int `json:"total_input_tokens"` // cumulative input - OutputTokens int `json:"total_output_tokens"` // cumulative output - CacheRead int `json:"total_cache_read_tokens"` // cumulative cache read - CacheWrite int `json:"total_cache_creation_tokens"` // cumulative cache write - ContextTokens int `json:"current_context_tokens"` // current context size - LastUpdated string `json:"last_updated"` // ISO 8601 timestamp + TurnCount int `json:"current_turn_count"` // user turn count + TotalRequests int `json:"agent_request_count"` // total LLM requests + TotalCompact int `json:"compact_request_count"` // number of compactions + SubAgentRequests int `json:"sub_agent_request_count"` // sub-agent requests + InputTokens int `json:"total_input_tokens"` // cumulative input + OutputTokens int `json:"total_output_tokens"` // cumulative output + CacheRead int `json:"total_cache_read_tokens"` // cumulative cache read + CacheWrite int `json:"total_cache_creation_tokens"` // cumulative cache write + ContextTokens int `json:"current_context_tokens"` // current context size + LastCallSpeedTokPerSec int `json:"last_call_speed_tok_per_sec"` // 最近一次 LLM call 输出速度 + LastUpdated string `json:"last_updated"` // ISO 8601 timestamp } // SessionRow session 列表行 diff --git a/rust/src/agent.rs b/rust/src/agent.rs index f7afc404..83d7a6ac 100644 --- a/rust/src/agent.rs +++ b/rust/src/agent.rs @@ -288,6 +288,8 @@ struct Agent { last_output_tokens: usize, last_cache_read_tokens: usize, last_cache_creation_tokens: usize, + last_start_ms: i64, + last_end_ms: i64, msg_tx: Arc>>>, // 主循环消息队列发送端（Arc> 以便 readline 线程退出时主动 drop） msg_rx: mpsc::Receiver, // 主循环消息队列接收端 sub_result_rx: mpsc::Receiver, // SubAgent 结果专用通道（对齐 NOTIFY_FIFO） @@ -682,7 +684,7 @@ impl Agent { let mut f = std::fs::File::create(&paths.stats)?; write!( f, - r#"{{"current_turn_count":0,"agent_request_count":0,"compact_request_count":0,"sub_agent_request_count":0,"total_input_tokens":0,"total_output_tokens":0,"total_cache_read_tokens":0,"total_cache_creation_tokens":0,"current_context_tokens":0,"last_updated":""}}{}"#, + r#"{{"current_turn_count":0,"agent_request_count":0,"compact_request_count":0,"sub_agent_request_count":0,"total_input_tokens":0,"total_output_tokens":0,"total_cache_read_tokens":0,"total_cache_creation_tokens":0,"current_context_tokens":0,"last_call_speed_tok_per_sec":0,"last_updated":""}}{}"#, '\n' )?; } @@ -726,6 +728,8 @@ impl Agent { last_output_tokens: 0, last_cache_read_tokens: 0, last_cache_creation_tokens: 0, + last_start_ms: 0, + last_end_ms: 0, msg_tx, msg_rx, sub_result_rx, @@ -867,6 +871,8 @@ impl Agent { last_output_tokens: 0, last_cache_read_tokens: 0, last_cache_creation_tokens: 0, + last_start_ms: 0, + last_end_ms: 0, msg_tx: Arc::new(Mutex::new(Some(sub_msg_tx))), msg_rx: _sub_msg_rx, sub_result_rx: sub_rrx, @@ -1823,6 +1829,8 @@ impl Agent { output_tokens, cache_read_input_tokens, cache_creation_input_tokens, + start_ms, + end_ms, }) => { self.emit_and_append_event(json!({ "type":"usage", @@ -1841,6 +1849,8 @@ impl Agent { self.last_output_tokens = *output_tokens as usize; self.last_cache_read_tokens = *cache_read_input_tokens as usize; self.last_cache_creation_tokens = *cache_creation_input_tokens as usize; + self.last_start_ms = *start_ms; + self.last_end_ms = *end_ms; } DisplayEvent::Stop(reason) => { self.emit_and_append_event(json!({"type":"stop","reason":&reason}))?; @@ -2216,6 +2226,14 @@ impl Agent { if ctx > 0 { Self::set_stat_usize(stats, "current_context_tokens", ctx); } + // 对齐 bash 版：speed = output_tokens / (end_ms - start_ms) * 1000，duration > 0 + let speed = if self.last_end_ms > self.last_start_ms { + (self.last_output_tokens as i64 * 1000 + / (self.last_end_ms - self.last_start_ms)) as usize + } else { + 0 + }; + Self::set_stat_usize(stats, "last_call_speed_tok_per_sec", speed); stats.insert( "last_updated".to_string(), Value::String(chrono_now_rfc3339()), @@ -2264,6 +2282,7 @@ impl Agent { let ao = stats_get_f64(&stats, "total_output_tokens") as usize; let ctx = stats_get_f64(&stats, "current_context_tokens") as usize; let cr = stats_get_f64(&stats, "total_cache_read_tokens") as usize; + let speed = stats_get_f64(&stats, "last_call_speed_tok_per_sec") as usize; let cache_pct = { let total = ai + cr; if total > 0 { @@ -2275,8 +2294,9 @@ impl Agent { let idle = status == "idle" && self.active_task_count == 0; let prefix = if idle { "" } else { "⏳ " }; let progress = if idle { 0 } else { 3 }; + // 对齐 bash 版 term_title.awk：model T:turn R:req I:in+cr(pct) O:out C:ctx S:speedtok/s let title = format!( - "\x1b]0;{}{} T:{} R:{} I:{}({}) O:{} C:{}\x07\x1b]9;4;{}\x07", + "\x1b]0;{}{} T:{} R:{} I:{}({}) O:{} C:{} S:{}tok/s\x07\x1b]9;4;{}\x07", prefix, self.cfg.model, Self::fmt_num(tc), @@ -2285,6 +2305,7 @@ impl Agent { cache_pct, Self::fmt_num(ao), Self::fmt_num(ctx), + Self::fmt_num(speed), progress ); // 通过 display worker 序列化输出，避免与 text delta 交织 diff --git a/rust/src/lib.rs b/rust/src/lib.rs index 9bfad410..2e665f13 100644 --- a/rust/src/lib.rs +++ b/rust/src/lib.rs @@ -456,6 +456,8 @@ pub mod types { pub output_tokens: i64, pub cache_read_input_tokens: i64, pub cache_creation_input_tokens: i64, + pub start_ms: i64, + pub end_ms: i64, } #[derive(Debug, Clone)] @@ -532,7 +534,7 @@ pub mod types { pub fn parse_usage_payload(payload: &str) -> Result { let parts: Vec<&str> = payload.split('\t').collect(); - if parts.len() != 4 { + if parts.len() != 4 && parts.len() != 6 { bail!("invalid usage payload") } Ok(UsageEvent { @@ -540,6 +542,8 @@ pub mod types { output_tokens: parts[1].parse()?, cache_read_input_tokens: parts[2].parse()?, cache_creation_input_tokens: parts[3].parse()?, + start_ms: parts.get(4).and_then(|s| s.parse().ok()).unwrap_or(0), + end_ms: parts.get(5).and_then(|s| s.parse().ok()).unwrap_or(0), }) } diff --git a/rust/src/sse.rs b/rust/src/sse.rs index edd3d56c..ee5b7749 100644 --- a/rust/src/sse.rs +++ b/rust/src/sse.rs @@ -75,11 +75,19 @@ pub mod toolcall { } } +fn now_ms() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis() as i64 +} + pub mod claude { use crate::protocol::{ ErrorEvent, Event, RetryEvent, StopEvent, TextEvent, ThinkingEvent, UsageEvent, }; use crate::sse::toolcall::build_tool_call_event; + use crate::sse::now_ms; use anyhow::{Result, anyhow}; use serde_json::Value; use std::io::{BufRead, BufReader, Read}; @@ -143,6 +151,7 @@ pub mod claude { let mut cache_creation_input_tokens = 0i64; let mut pending_usage: Option = None; let mut pending_stop: Option = None; + let start_ms = now_ms(); read_sse(reader, |evt| { if evt.event == "RETRY" { @@ -283,6 +292,8 @@ pub mod claude { output_tokens, cache_read_input_tokens, cache_creation_input_tokens, + start_ms: 0, + end_ms: 0, }); pending_stop = Some(stop_reason.clone()); } @@ -306,7 +317,9 @@ pub mod claude { }) .map_err(|e| anyhow!("parse claude sse: {e}"))?; - if let Some(usage) = pending_usage { + if let Some(mut usage) = pending_usage { + usage.start_ms = start_ms; + usage.end_ms = now_ms(); emit(Event::Usage(usage))?; } if let Some(reason) = pending_stop { @@ -329,6 +342,7 @@ pub mod openai { ErrorEvent, Event, RetryEvent, StopEvent, TextEvent, ThinkingEvent, UsageEvent, }; use crate::sse::toolcall::build_tool_call_event; + use crate::sse::now_ms; use anyhow::Result; use serde_json::Value; use std::collections::BTreeMap; @@ -352,6 +366,7 @@ pub mod openai { let mut pending_calls: BTreeMap = BTreeMap::new(); let mut pending_usage: Option = None; let mut pending_stop: Option = None; + let start_ms = now_ms(); loop { line.clear(); @@ -388,6 +403,8 @@ pub mod openai { output_tokens, cache_read_input_tokens, cache_creation_input_tokens: 0, + start_ms, + end_ms: now_ms(), }); pending_stop = Some(stop_reason.clone()); break; @@ -530,6 +547,7 @@ pub mod responses { ErrorEvent, Event, RetryEvent, StopEvent, TextEvent, ThinkingEvent, UsageEvent, }; use crate::sse::toolcall::build_tool_call_event; + use crate::sse::now_ms; use anyhow::Result; use serde_json::Value; use std::collections::BTreeMap; @@ -553,6 +571,7 @@ pub mod responses { let mut calls: BTreeMap = BTreeMap::new(); let mut item_indexes: BTreeMap = BTreeMap::new(); let mut completed = false; + let start_ms = now_ms(); while { line.clear(); br.read_line(&mut line)? != 0 @@ -632,6 +651,8 @@ pub mod responses { output_tokens, cache_read_input_tokens, cache_creation_input_tokens: 0, + start_ms, + end_ms: now_ms(), }))?; emit(Event::Stop(StopEvent { reason: if has_tools { "tool_use" } else { "end_turn" }.to_string(), @@ -667,6 +688,8 @@ pub mod responses { output_tokens, cache_read_input_tokens, cache_creation_input_tokens: 0, + start_ms, + end_ms: now_ms(), }))?; emit(Event::Stop(StopEvent { reason: "error".to_string(), diff --git a/rust/src/store.rs b/rust/src/store.rs index 075d9e4f..b89bd449 100644 --- a/rust/src/store.rs +++ b/rust/src/store.rs @@ -181,7 +181,7 @@ pub fn store_stats_write(stats_path: &Path, stats: &serde_json::Map serde_json::Map { "current_context_tokens".to_string(), Value::Number(0.into()), ); + stats.insert( + "last_call_speed_tok_per_sec".to_string(), + Value::Number(0.into()), + ); stats.insert("last_updated".to_string(), Value::String(String::new())); stats } diff --git a/tests/test.sh b/tests/test.sh index bf0b4c3f..0de6dc2c 100755 --- a/tests/test.sh +++ b/tests/test.sh @@ -2517,7 +2517,7 @@ test_agent_compact_context() { # Test 40: stats.json structure completeness (S1/S2/S4) test_agent_stats_structure() { - info "Test 40: stats.json structure completeness (S1 sub_agent_request_count, S2 last_updated, S4 no trailing spaces)" + info "Test 40: stats.json structure completeness (S1 sub_agent_request_count, S3 last_call_speed_tok_per_sec, S2 last_updated, S4 no trailing spaces)" local home_dir stats_file home_dir=$(mktemp -d) BASH_AGENT_HOME="$home_dir" "$AGENT" -p claude --base-url "$BASE/v1" -m test --api-key test 'STATS_CHECK_MARKER' >/dev/null 2>&1 || true @@ -2538,6 +2538,13 @@ test_agent_stats_structure() { red "stats structure: sub_agent_request_count field MISSING"; echo " Content: $content"; ((FAIL++)) || true fi + # S3: last_call_speed_tok_per_sec must exist (introduced in PR #87) + if [[ "$content" == *"last_call_speed_tok_per_sec"* ]]; then + green "stats structure: last_call_speed_tok_per_sec field present"; ((PASS++)) || true + else + red "stats structure: last_call_speed_tok_per_sec field MISSING"; echo " Content: $content"; ((FAIL++)) || true + fi + # S2: last_updated must be non-empty (ISO timestamp like 2025-01-01T00:00:00Z) local last_updated last_updated=$(grep -o '"last_updated":"[^"]*"' "$stats_file" 2>/dev/null | head -1) From f3ff4a857fbb108cdb5ae93b36ef03ead19132a9 Mon Sep 17 00:00:00 2001 From: lloydzhou Date: Thu, 10 Sep 2026 20:15:33 +0800 Subject: [PATCH 03/10] =?UTF-8?q?refactor(bash):=20speed=20=E8=AE=A1?= =?UTF-8?q?=E6=97=B6=E6=94=B9=E7=94=A8=20curl=20-w=20SSE=20timing=20?= =?UTF-8?q?=E5=9D=97=EF=BC=8C=E7=A7=BB=E9=99=A4=20date=5Fms=20fallback?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 原 date_ms 三级 fallback 在 macOS 上两级是坏的： - date +%s%3N: BSD date 不认 %N，返回垃圾值 - systime(): macOS bwk awk 无此函数，直接崩溃 - 实际全靠系统 perl 兜底 新方案： - curl -w 在 SSE 流末尾追加标准 event:timing 块（time_total/time_starttransfer） - 块沿现有管道自然流转，openai/responses 转换层各加 2 行透传规则 - claude_sse.awk END 里直接算：speed = out / (total - ttfb) 口径对齐 Go/Rust/C：读流开始→流结束，排除建连/TLS/TTFB - USAGE 第 5 字段携带算好的 speed，bash 侧零计算 - --retry 时 -w 只在最终成功传输输出一次，speed 不含重试等待 - 零 fallback、零外部进程、零新依赖、毫秒精度 --- src/agent.sh | 5 ++--- src/awk/claude_sse.awk | 26 +++++++++++++------------- src/awk/transport_openai_sse.awk | 4 ++++ src/awk/transport_responses_sse.awk | 5 +++++ 4 files changed, 24 insertions(+), 16 deletions(-) diff --git a/src/agent.sh b/src/agent.sh index 8ee48311..d85cdd08 100755 --- a/src/agent.sh +++ b/src/agent.sh @@ -565,7 +565,7 @@ store_session_list_rows() { # llm llm_stream_curl() { - exec 9< <(curl -sS --no-buffer -D - --retry 2 --retry-delay 1 --retry-max-time 20 --connect-timeout 5 --speed-limit 1 --speed-time 60 "${HEADER_ARGS[@]}" -d @- "$API_URL" 2>&1) + exec 9< <(curl -sS --no-buffer -D - --retry 2 --retry-delay 1 --retry-max-time 20 --connect-timeout 5 --speed-limit 1 --speed-time 60 -w '\nevent: timing\ndata: {"time_total":%{time_total},"time_starttransfer":%{time_starttransfer}}\n\n' "${HEADER_ARGS[@]}" -d @- "$API_URL" 2>&1) curl_pid=$! echo "$curl_pid" > "/tmp/agent_curl_pid.$$" 2>/dev/null || true util_awk_run -f "$AWK_DIR/http_stream.awk" <&9 @@ -1426,8 +1426,7 @@ agent_loop_stream() { STOP) stop="${REPLY_MESSAGE[1]}" ;; ERROR) loop_error="${REPLY_MESSAGE[1]}"; stop="error"; break ;; USAGE) _ctx_tokens=$(agent_record_usage "agent" agent_request_count false) - local _dur _speed; _dur=$((${REPLY_MESSAGE[6]:-0}-${REPLY_MESSAGE[5]:-0})); _speed=$((_dur>0?${REPLY_MESSAGE[2]:-0}*1000/_dur:0)) - store_stats_update last_call_speed_tok_per_sec="$_speed" current_context_tokens="${_ctx_tokens:-0}" ;; + store_stats_update last_call_speed_tok_per_sec="${REPLY_MESSAGE[5]:-0}" current_context_tokens="${_ctx_tokens:-0}" ;; esac done exec 8<&- diff --git a/src/awk/claude_sse.awk b/src/awk/claude_sse.awk index 1f0cbf86..ee0aa702 100644 --- a/src/awk/claude_sse.awk +++ b/src/awk/claude_sse.awk @@ -3,16 +3,6 @@ # Output: Unified protocol via emit1/emit/emit_flush # Requires: awk -v verbose=true/false -f json.awk -f protocol.awk -f todo_protocol.awk -f claude_sse.awk -function date_ms( cmd, ms) { - cmd = "perl -MTime::HiRes=time -e \047printf \042%d\\n\042, time * 1000\047" - if ((cmd | getline ms) > 0) { close(cmd); return ms + 0 } - close(cmd) - cmd = "date +%s%3N" - if ((cmd | getline ms) > 0) { close(cmd); return ms + 0 } - close(cmd) - return systime() * 1000 -} - BEGIN { event = "" block_type = "" @@ -25,7 +15,8 @@ BEGIN { cache_read_input_tokens = 0 cache_creation_input_tokens = 0 pending_stop_reason = "" - start_ms = date_ms() + timing_total = 0 + timing_ttfb = 0 } /^:/ { next } @@ -131,6 +122,14 @@ BEGIN { pending_cache_creation_tokens = cache_creation_input_tokens pending_stop_reason = stop_reason } + else if (event == "timing") { + # Injected by curl -w at end of stream; survives http/transport layers + # speed 口径对齐 Go/Rust/C：读流开始→流结束（排除建连/TLS/TTFB） + tt = extract_num(json, "time_total", 1) + ts = extract_num(json, "time_starttransfer", 1) + if (tt != "") timing_total = tt + 0 + if (ts != "") timing_ttfb = ts + 0 + } else if (event == "error") { msg = extract_str(json, "message", 1) emit1("ERROR"); emit(msg); emit_flush() @@ -141,8 +140,9 @@ BEGIN { END { if (pending_stop_reason != "") { - end_ms = date_ms() - emit1("USAGE"); emit(pending_input_tokens + 0); emit(pending_output_tokens + 0); emit(pending_cache_read_tokens + 0); emit(pending_cache_creation_tokens + 0); emit(start_ms + 0); emit(end_ms); emit_flush() + gen = timing_total - timing_ttfb + speed = (gen > 0) ? int(pending_output_tokens / gen) : 0 + emit1("USAGE"); emit(pending_input_tokens + 0); emit(pending_output_tokens + 0); emit(pending_cache_read_tokens + 0); emit(pending_cache_creation_tokens + 0); emit(speed + 0); emit_flush() emit1("STOP"); emit(pending_stop_reason); emit_flush() } else { # No message_stop received — emit an error STOP so the caller loop diff --git a/src/awk/transport_openai_sse.awk b/src/awk/transport_openai_sse.awk index 192dfc54..b7b57bc0 100644 --- a/src/awk/transport_openai_sse.awk +++ b/src/awk/transport_openai_sse.awk @@ -29,6 +29,10 @@ BEGIN { # Pass through ERROR lines from http_stream.awk /^ERROR:/ { print; fflush(); next } +# Pass through timing event from curl -w (captured by claude_sse.awk) +/^event: timing$/ { print; fflush(); next } +/^data: \{"time_total"/ { print; fflush(); next } + # Handle curl retry: reset all state /^RETRY:/ { _sse_reset() diff --git a/src/awk/transport_responses_sse.awk b/src/awk/transport_responses_sse.awk index 6b44183d..8a852be2 100644 --- a/src/awk/transport_responses_sse.awk +++ b/src/awk/transport_responses_sse.awk @@ -24,6 +24,11 @@ BEGIN { /^:/ { next } /^ERROR:/ { print; fflush(); next } + +# Pass through timing event from curl -w (captured by claude_sse.awk) +/^event: timing$/ { print; fflush(); next } +/^data: \{"time_total"/ { print; fflush(); next } + /^RETRY:/ { _reset(); print; fflush(); next } /^event: / { event = substr($0, 8); next } From 0ff0a27dc2106526da50762a7d057f694356fd9b Mon Sep 17 00:00:00 2001 From: lloydzhou Date: Thu, 10 Sep 2026 21:05:27 +0800 Subject: [PATCH 04/10] =?UTF-8?q?fix(go,rust,c):=20speed=20=E4=BB=85?= =?UTF-8?q?=E5=9C=A8=E6=B5=81=E6=AD=A3=E5=B8=B8=E7=BB=88=E7=BB=93=E6=97=B6?= =?UTF-8?q?=E6=9B=B4=E6=96=B0=EF=BC=8C=E5=A4=B1=E8=B4=A5=E7=BB=88=E6=80=81?= =?UTF-8?q?=E5=AF=B9=E9=BD=90=20bash?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 跨 provider review 发现的口径差异（bash 在 message_stop/[DONE]/ response.completed 才发 USAGE，失败终态/中断不发 → speed 保留旧值）： - Go responses 失败终态（response.failed/incomplete/error）也发 USAGE → speed 被部分输出的偏低值覆盖；Rust 同路径同问题； C 中断场景（收到 message_start 的 in 但未到 message_stop） 流中 USAGE 使守卫通过 → 写 speed=0 - 修复：三版本统一"正常终结"标志 - Go: Usage.Stopped（message_stop/[DONE]/completed=true， failed=false），store.RecordUsage 仅 Stopped 时写 speed - Rust: UsageEvent.stopped（同上四处置值），agent 仅 last_stopped 时写 speed - C: accum->stopped（SSE_STOP 置位）守卫 speed 写入 C 版计时口径修复（对齐 bash 传输时间）： - 原 start_ms 在 curl_easy_perform 前打点，含 DNS/TCP/TLS/TTFB - 改为 perform 成功后 curl_easy_getinfo 查 CURLINFO_TOTAL_TIME_T / CURLINFO_STARTTRANSFER_TIME_T， 与 bash 的 curl -w 同一套计时器（time_total - time_starttransfer， 仅传输时间）；补发最终 USAGE 覆盖流中 now_ms 口径时间戳 （token 字段为 0，>0 守卫不重复累加） C 版 speed 恒 0 的根因修复： - agent 层 stream_display_callback 的 SSE_USAGE case 漏记 start_ms/end_ms（transport 层另一份 accumulator 有记，agent 层读自己这份）→ dur 恒 0 亚毫秒截断（三版本）：整数 ms dur 截断为 0 → speed=0 偶发； 传输真实发生时向上取整至少 1ms（对齐 bash awk 浮点行为） tests/test.sh Test 40 补 S5 断言 speed > 0（原只查字段存在， C 版恒 0 未被发现的盲区） 四版本 e2e 各 223 passed --- c/agent.c | 7 +++++-- c/transport.c | 16 ++++++++++++++++ go/store.go | 13 +++++++++++-- go/transport.go | 4 ++++ go/types.go | 1 + rust/src/agent.rs | 26 +++++++++++++++++++------- rust/src/lib.rs | 3 +++ rust/src/sse.rs | 4 ++++ tests/test.sh | 10 ++++++++++ 9 files changed, 73 insertions(+), 11 deletions(-) diff --git a/c/agent.c b/c/agent.c index 3eeae244..5cccadfa 100644 --- a/c/agent.c +++ b/c/agent.c @@ -171,6 +171,8 @@ static void stream_display_callback(void *ctx, const SseEvent *evt) { if (evt->out_tokens > 0) sctx->accum.out_tokens = evt->out_tokens; if (evt->cache_read_tokens > 0) sctx->accum.cache_read_tokens = evt->cache_read_tokens; if (evt->cache_creation_tokens > 0) sctx->accum.cache_creation_tokens = evt->cache_creation_tokens; + sctx->accum.start_ms = evt->start_ms; + sctx->accum.end_ms = evt->end_ms; break; case SSE_TOOL_CALL_START: { @@ -1029,8 +1031,9 @@ int agent_loop(Agent *agent, const char *user_input, const char *turn_kind) { store_stats_set_int_file(agent->paths.stats, "current_context_tokens", agent->last_context_tokens); } - /* 对齐 bash 版：speed = output_tokens / (end_ms - start_ms) * 1000，duration > 0 */ - { + /* 对齐 bash 版：speed = output_tokens / (end_ms - start_ms) * 1000，duration > 0 + * 且仅在流正常终结（收到 stop）时更新——中断场景 bash 无 USAGE、保留旧值 */ + if (accum->stopped) { long long dur = agent->last_end_ms - agent->last_start_ms; int speed = (dur > 0) ? (int)((long long)agent->last_output_tokens * 1000LL / dur) : 0; store_stats_set_int_file(agent->paths.stats, "last_call_speed_tok_per_sec", speed); diff --git a/c/transport.c b/c/transport.c index a902a1c6..4270b914 100644 --- a/c/transport.c +++ b/c/transport.c @@ -650,6 +650,22 @@ int http_post_sse(const char *url, const char **headers, int header_count, /* 成功或不可重试错误 → 处理残留 JSON */ if (rc == CURLE_OK) { process_residual_json(&sctx, provider, callback, ctx); + /* 用 curl 内部计时器对齐 bash 版 curl -w 的 + * time_total - time_starttransfer（同一套计时器，口径完全一致， + * 排除 DNS/TCP/TLS/TTFB，只算传输时间）。 + * 补发最终 USAGE 覆盖流中事件的 now_ms 口径时间戳； + * token 字段为 0，accum 侧 >0 守卫不会重复累加 */ + curl_off_t total_us = 0, ttfb_us = 0; + curl_easy_getinfo(curl, CURLINFO_TOTAL_TIME_T, &total_us); + curl_easy_getinfo(curl, CURLINFO_STARTTRANSFER_TIME_T, &ttfb_us); + curl_off_t dur_us = total_us - ttfb_us; + SseEvent tevt; + memset(&tevt, 0, sizeof(tevt)); + tevt.type = SSE_USAGE; + tevt.start_ms = 0; + /* 微秒向上取整到毫秒：亚毫秒传输至少 1ms，避免整数截断 speed=0 */ + tevt.end_ms = (dur_us > 0) ? (long long)((dur_us + 999) / 1000) : 0; + callback(ctx, &tevt); } sb_free(&sctx.line_buf); FREE_PTR(sctx.event); diff --git a/go/store.go b/go/store.go index b5121b92..c3ea3327 100644 --- a/go/store.go +++ b/go/store.go @@ -56,8 +56,17 @@ func (s *FileStore) UpdateStats(usage Usage, model string) error { s.stats.CacheWrite += usage.CacheWrite s.stats.CacheRead += usage.CacheRead // 对齐 bash 版：speed = output_tokens / (end_ms - start_ms) * 1000，duration > 0 - if usage.EndMs > usage.StartMs { - s.stats.LastCallSpeedTokPerSec = usage.OutputTokens * 1000 / int(usage.EndMs-usage.StartMs) + // 且仅在流正常终结时更新（失败终态/中断 bash 不发 USAGE，保留旧值） + // 亚毫秒传输向上取整至少 1ms，避免整数截断导致 speed=0（对齐 bash awk 浮点行为） + if !usage.Stopped { + return s.flushStats() + } + dur := usage.EndMs - usage.StartMs + if dur <= 0 && usage.OutputTokens > 0 { + dur = 1 + } + if dur > 0 { + s.stats.LastCallSpeedTokPerSec = usage.OutputTokens * 1000 / int(dur) } else { s.stats.LastCallSpeedTokPerSec = 0 } diff --git a/go/transport.go b/go/transport.go index 977f1460..e1ff0608 100644 --- a/go/transport.go +++ b/go/transport.go @@ -295,6 +295,7 @@ func (t *HTTPTransport) parseSSEStream(ctx context.Context, resp *http.Response, stopEmitted = true // 发送 USAGE + STOP（对齐 bash 版：带 start_ms/end_ms 用于计算 tok/s） ch <- Event{Type: EventUsage, Payload: Usage{ + Stopped: true, InputTokens: inputTokens, OutputTokens: outputTokens, CacheRead: cacheRead, @@ -501,6 +502,7 @@ func (t *HTTPTransport) handleResponsesEvent(eventType, data string, ch chan<- E CacheRead: *cacheRead, StartMs: startMs, EndMs: time.Now().UnixMilli(), + Stopped: true, }} stopReason := "end_turn" if hasTools { @@ -542,6 +544,7 @@ func (t *HTTPTransport) handleResponsesEvent(eventType, data string, ch chan<- E CacheRead: *cacheRead, StartMs: startMs, EndMs: time.Now().UnixMilli(), + Stopped: false, }} ch <- Event{Type: EventStop, Fields: []string{"STOP", "error"}} return true @@ -594,6 +597,7 @@ func (t *HTTPTransport) handleOpenAIChunk(data string, ch chan<- Event, CacheWrite: *cacheCreate, StartMs: startMs, EndMs: time.Now().UnixMilli(), + Stopped: true, }} ch <- Event{Type: EventStop, Fields: []string{"STOP", sr}} return diff --git a/go/types.go b/go/types.go index e134c369..b56f6351 100644 --- a/go/types.go +++ b/go/types.go @@ -125,6 +125,7 @@ type Usage struct { Cost float64 StartMs int64 // 流开始时间戳（毫秒） EndMs int64 // 流结束时间戳（毫秒） + Stopped bool // 流正常终结（message_stop/[DONE]/response.completed）；失败终态 false，不更新 speed } // ToolCallInfo 表示一个工具调用 diff --git a/rust/src/agent.rs b/rust/src/agent.rs index 83d7a6ac..67934a68 100644 --- a/rust/src/agent.rs +++ b/rust/src/agent.rs @@ -290,6 +290,7 @@ struct Agent { last_cache_creation_tokens: usize, last_start_ms: i64, last_end_ms: i64, + last_stopped: bool, msg_tx: Arc>>>, // 主循环消息队列发送端（Arc> 以便 readline 线程退出时主动 drop） msg_rx: mpsc::Receiver, // 主循环消息队列接收端 sub_result_rx: mpsc::Receiver, // SubAgent 结果专用通道（对齐 NOTIFY_FIFO） @@ -730,6 +731,7 @@ impl Agent { last_cache_creation_tokens: 0, last_start_ms: 0, last_end_ms: 0, + last_stopped: false, msg_tx, msg_rx, sub_result_rx, @@ -873,6 +875,7 @@ impl Agent { last_cache_creation_tokens: 0, last_start_ms: 0, last_end_ms: 0, + last_stopped: false, msg_tx: Arc::new(Mutex::new(Some(sub_msg_tx))), msg_rx: _sub_msg_rx, sub_result_rx: sub_rrx, @@ -1831,6 +1834,7 @@ impl Agent { cache_creation_input_tokens, start_ms, end_ms, + stopped, }) => { self.emit_and_append_event(json!({ "type":"usage", @@ -1851,6 +1855,7 @@ impl Agent { self.last_cache_creation_tokens = *cache_creation_input_tokens as usize; self.last_start_ms = *start_ms; self.last_end_ms = *end_ms; + self.last_stopped = *stopped; } DisplayEvent::Stop(reason) => { self.emit_and_append_event(json!({"type":"stop","reason":&reason}))?; @@ -2227,13 +2232,20 @@ impl Agent { Self::set_stat_usize(stats, "current_context_tokens", ctx); } // 对齐 bash 版：speed = output_tokens / (end_ms - start_ms) * 1000，duration > 0 - let speed = if self.last_end_ms > self.last_start_ms { - (self.last_output_tokens as i64 * 1000 - / (self.last_end_ms - self.last_start_ms)) as usize - } else { - 0 - }; - Self::set_stat_usize(stats, "last_call_speed_tok_per_sec", speed); + // 且仅在流正常终结时更新（失败终态/中断 bash 不发 USAGE，保留旧值） + // 亚毫秒传输向上取整至少 1ms，避免整数截断导致 speed=0（对齐 bash awk 浮点行为） + if self.last_stopped { + let mut dur = self.last_end_ms - self.last_start_ms; + if dur <= 0 && self.last_output_tokens > 0 { + dur = 1; + } + let speed = if dur > 0 { + (self.last_output_tokens as i64 * 1000 / dur) as usize + } else { + 0 + }; + Self::set_stat_usize(stats, "last_call_speed_tok_per_sec", speed); + } stats.insert( "last_updated".to_string(), Value::String(chrono_now_rfc3339()), diff --git a/rust/src/lib.rs b/rust/src/lib.rs index 2e665f13..90c1a8a4 100644 --- a/rust/src/lib.rs +++ b/rust/src/lib.rs @@ -458,6 +458,8 @@ pub mod types { pub cache_creation_input_tokens: i64, pub start_ms: i64, pub end_ms: i64, + /// 流正常终结（message_stop/[DONE]/response.completed）；失败终态 false，不更新 speed + pub stopped: bool, } #[derive(Debug, Clone)] @@ -544,6 +546,7 @@ pub mod types { cache_creation_input_tokens: parts[3].parse()?, start_ms: parts.get(4).and_then(|s| s.parse().ok()).unwrap_or(0), end_ms: parts.get(5).and_then(|s| s.parse().ok()).unwrap_or(0), + stopped: false, }) } diff --git a/rust/src/sse.rs b/rust/src/sse.rs index ee5b7749..af6782b8 100644 --- a/rust/src/sse.rs +++ b/rust/src/sse.rs @@ -294,6 +294,7 @@ pub mod claude { cache_creation_input_tokens, start_ms: 0, end_ms: 0, + stopped: true, }); pending_stop = Some(stop_reason.clone()); } @@ -405,6 +406,7 @@ pub mod openai { cache_creation_input_tokens: 0, start_ms, end_ms: now_ms(), + stopped: true, }); pending_stop = Some(stop_reason.clone()); break; @@ -653,6 +655,7 @@ pub mod responses { cache_creation_input_tokens: 0, start_ms, end_ms: now_ms(), + stopped: true, }))?; emit(Event::Stop(StopEvent { reason: if has_tools { "tool_use" } else { "end_turn" }.to_string(), @@ -690,6 +693,7 @@ pub mod responses { cache_creation_input_tokens: 0, start_ms, end_ms: now_ms(), + stopped: false, }))?; emit(Event::Stop(StopEvent { reason: "error".to_string(), diff --git a/tests/test.sh b/tests/test.sh index 0de6dc2c..3eaa1d79 100755 --- a/tests/test.sh +++ b/tests/test.sh @@ -2545,6 +2545,16 @@ test_agent_stats_structure() { red "stats structure: last_call_speed_tok_per_sec field MISSING"; echo " Content: $content"; ((FAIL++)) || true fi + # S5: speed value must be positive after a real LLM call + # (guards against accumulator wiring bugs like cagent missing start_ms/end_ms) + local speed_val + speed_val=$(grep -o '"last_call_speed_tok_per_sec":[0-9-]*' "$stats_file" 2>/dev/null | grep -o '[0-9-]*$' | head -1) + if [[ "$speed_val" =~ ^[0-9]+$ ]] && (( speed_val > 0 )); then + green "stats structure: last_call_speed_tok_per_sec positive ($speed_val)"; ((PASS++)) || true + else + red "stats structure: last_call_speed_tok_per_sec not positive"; echo " Value: $speed_val"; ((FAIL++)) || true + fi + # S2: last_updated must be non-empty (ISO timestamp like 2025-01-01T00:00:00Z) local last_updated last_updated=$(grep -o '"last_updated":"[^"]*"' "$stats_file" 2>/dev/null | head -1) From ce82e610c66c11769d8f6cfd885227b52c79f3ec Mon Sep 17 00:00:00 2001 From: lloydzhou Date: Thu, 10 Sep 2026 21:18:46 +0800 Subject: [PATCH 05/10] =?UTF-8?q?fix(ci):=20=E5=AE=B9=E5=99=A8=E5=86=85=20?= =?UTF-8?q?apt-get=20update=20=E8=B7=B3=E8=BF=87=20Release=20=E8=BF=87?= =?UTF-8?q?=E6=9C=9F=E6=A3=80=E6=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit debian:11 (bullseye) 的 bullseye-security InRelease 元数据过期 （invalid since 2d），导致所有 Debian 11 container 构建 job 的 apt-get update 退出 100。加 -o Acquire::Check-Valid-Until=false 跳过检查（与代码改动无关的 CI 基建问题） --- .github/workflows/ci.yml | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 21ea20d1..3a269bd8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -102,7 +102,7 @@ jobs: export GOCACHE="$PWD/go/.gocache" GOMODCACHE="$PWD/go/.gomodcache" go version sed -i "s|deb.debian.org|deb.debian.org|g; s|security.debian.org/debian-security|security.debian.org/debian-security|g" /etc/apt/sources.list 2>/dev/null || true - apt-get update + apt-get update -o Acquire::Check-Valid-Until=false apt-get install -y --no-install-recommends gcc make libc6-dev mkdir -p dist if [ "${{ matrix.goarch }}" = "arm64" ]; then @@ -187,7 +187,7 @@ jobs: docker run --rm -v "$PWD":/github/workspace -w /github/workspace debian:11 \ bash -euxo pipefail -c ' sed -i "s|deb.debian.org|deb.debian.org|g" /etc/apt/sources.list 2>/dev/null || true - apt-get update + apt-get update -o Acquire::Check-Valid-Until=false apt-get install -y --no-install-recommends curl ca-certificates build-essential pkg-config if [ "${{ matrix.target }}" = "aarch64-unknown-linux-gnu" ]; then apt-get install -y --no-install-recommends gcc-aarch64-linux-gnu libc6-dev-arm64-cross @@ -263,7 +263,7 @@ jobs: docker run --rm -v "$PWD":/github/workspace -w /github/workspace debian:11 \ bash -euxo pipefail -c ' sed -i "s|deb.debian.org|deb.debian.org|g" /etc/apt/sources.list 2>/dev/null || true - apt-get update + apt-get update -o Acquire::Check-Valid-Until=false apt-get install -y --no-install-recommends curl ca-certificates build-essential pkg-config if [ "${{ matrix.target }}" = "aarch64-unknown-linux-gnu" ]; then apt-get install -y --no-install-recommends gcc-aarch64-linux-gnu libc6-dev-arm64-cross @@ -319,7 +319,7 @@ jobs: docker run --rm -v "$PWD":/github/workspace -w /github/workspace debian:11 \ bash -euxo pipefail -c ' sed -i "s|deb.debian.org|deb.debian.org|g" /etc/apt/sources.list 2>/dev/null || true - apt-get update + apt-get update -o Acquire::Check-Valid-Until=false apt-get install -y --no-install-recommends gcc make xxd libc6-dev libcurl4-openssl-dev pkg-config mkdir -p dist make -C c CC=gcc From 43694d99c3ee683e5985ba7ab1773fc9ad06ce2c Mon Sep 17 00:00:00 2001 From: lloydzhou Date: Thu, 10 Sep 2026 21:22:21 +0800 Subject: [PATCH 06/10] =?UTF-8?q?fix(ci):=20debian:11=20=E6=BA=90=E5=88=87?= =?UTF-8?q?=E6=8D=A2=E5=88=B0=20archive.debian.org?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit bullseye LTS 已于 2026-08-31 结束，deb.debian.org 上的 bullseye-security 旧版包已清理（install 阶段 404）。 sed 替换占位行改为指向 archive.debian.org（EOL 官方归档， 包永久可用），配合已有的 Check-Valid-Until=false --- .github/workflows/ci.yml | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3a269bd8..7f729740 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -101,7 +101,7 @@ jobs: export PATH="$(dirname "$GO_BIN"):$PATH" export GOCACHE="$PWD/go/.gocache" GOMODCACHE="$PWD/go/.gomodcache" go version - sed -i "s|deb.debian.org|deb.debian.org|g; s|security.debian.org/debian-security|security.debian.org/debian-security|g" /etc/apt/sources.list 2>/dev/null || true + sed -i "s|deb.debian.org|archive.debian.org|g; s|security.debian.org/debian-security|archive.debian.org/debian-security|g" /etc/apt/sources.list 2>/dev/null || true apt-get update -o Acquire::Check-Valid-Until=false apt-get install -y --no-install-recommends gcc make libc6-dev mkdir -p dist @@ -186,7 +186,7 @@ jobs: run: | docker run --rm -v "$PWD":/github/workspace -w /github/workspace debian:11 \ bash -euxo pipefail -c ' - sed -i "s|deb.debian.org|deb.debian.org|g" /etc/apt/sources.list 2>/dev/null || true + sed -i "s|deb.debian.org|archive.debian.org|g" /etc/apt/sources.list 2>/dev/null || true apt-get update -o Acquire::Check-Valid-Until=false apt-get install -y --no-install-recommends curl ca-certificates build-essential pkg-config if [ "${{ matrix.target }}" = "aarch64-unknown-linux-gnu" ]; then @@ -262,7 +262,7 @@ jobs: run: | docker run --rm -v "$PWD":/github/workspace -w /github/workspace debian:11 \ bash -euxo pipefail -c ' - sed -i "s|deb.debian.org|deb.debian.org|g" /etc/apt/sources.list 2>/dev/null || true + sed -i "s|deb.debian.org|archive.debian.org|g" /etc/apt/sources.list 2>/dev/null || true apt-get update -o Acquire::Check-Valid-Until=false apt-get install -y --no-install-recommends curl ca-certificates build-essential pkg-config if [ "${{ matrix.target }}" = "aarch64-unknown-linux-gnu" ]; then @@ -318,7 +318,7 @@ jobs: run: | docker run --rm -v "$PWD":/github/workspace -w /github/workspace debian:11 \ bash -euxo pipefail -c ' - sed -i "s|deb.debian.org|deb.debian.org|g" /etc/apt/sources.list 2>/dev/null || true + sed -i "s|deb.debian.org|archive.debian.org|g" /etc/apt/sources.list 2>/dev/null || true apt-get update -o Acquire::Check-Valid-Until=false apt-get install -y --no-install-recommends gcc make xxd libc6-dev libcurl4-openssl-dev pkg-config mkdir -p dist From 34ac629b4fcb6de975565995806bb70cfcebd45a Mon Sep 17 00:00:00 2001 From: lloydzhou Date: Thu, 10 Sep 2026 21:28:35 +0800 Subject: [PATCH 07/10] =?UTF-8?q?ci:=20=E6=9E=84=E5=BB=BA=E5=AE=B9?= =?UTF-8?q?=E5=99=A8=20debian:11=20=E2=86=92=20debian:12=20(bookworm)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit bullseye 已于 2026-08-31 结束 LTS，镜像进入元数据过期+pool 清理 的半死状态。切换 bookworm（支持至 2026+6/2028 LTS）。 ⚠️ 兼容性变化：Linux 构建产物 glibc 下限从 2.31 升至 2.36， 不再支持 Ubuntu 22.04 (2.35) 及更早；Ubuntu 24.04+/Debian 12+ 不受影响。 同时清理 bullseye EOL 的临时 workaround（archive/snapshot sed 替换与 Check-Valid-Until=false，debian:12 活仓库无需） --- .github/workflows/ci.yml | 20 ++++++++------------ 1 file changed, 8 insertions(+), 12 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7f729740..355e223b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -95,14 +95,13 @@ jobs: - name: Build goagent (linux, Debian 11 container) if: matrix.goos == 'linux' run: | - docker run --rm -v /opt/hostedtoolcache:/opt/hostedtoolcache -v "$PWD":/github/workspace -w /github/workspace debian:11 \ + docker run --rm -v /opt/hostedtoolcache:/opt/hostedtoolcache -v "$PWD":/github/workspace -w /github/workspace debian:12 \ bash -euxo pipefail -c ' GO_BIN="$(ls -d /opt/hostedtoolcache/go/*/x64/bin | sort -V | tail -1)/go" export PATH="$(dirname "$GO_BIN"):$PATH" export GOCACHE="$PWD/go/.gocache" GOMODCACHE="$PWD/go/.gomodcache" go version - sed -i "s|deb.debian.org|archive.debian.org|g; s|security.debian.org/debian-security|archive.debian.org/debian-security|g" /etc/apt/sources.list 2>/dev/null || true - apt-get update -o Acquire::Check-Valid-Until=false + apt-get update apt-get install -y --no-install-recommends gcc make libc6-dev mkdir -p dist if [ "${{ matrix.goarch }}" = "arm64" ]; then @@ -184,10 +183,9 @@ jobs: - name: Build rustagent (linux, Debian 11 container) if: matrix.runner == 'ubuntu-latest' run: | - docker run --rm -v "$PWD":/github/workspace -w /github/workspace debian:11 \ + docker run --rm -v "$PWD":/github/workspace -w /github/workspace debian:12 \ bash -euxo pipefail -c ' - sed -i "s|deb.debian.org|archive.debian.org|g" /etc/apt/sources.list 2>/dev/null || true - apt-get update -o Acquire::Check-Valid-Until=false + apt-get update apt-get install -y --no-install-recommends curl ca-certificates build-essential pkg-config if [ "${{ matrix.target }}" = "aarch64-unknown-linux-gnu" ]; then apt-get install -y --no-install-recommends gcc-aarch64-linux-gnu libc6-dev-arm64-cross @@ -260,10 +258,9 @@ jobs: - name: Build webagent (linux, Debian 11 container) if: matrix.runner == 'ubuntu-latest' run: | - docker run --rm -v "$PWD":/github/workspace -w /github/workspace debian:11 \ + docker run --rm -v "$PWD":/github/workspace -w /github/workspace debian:12 \ bash -euxo pipefail -c ' - sed -i "s|deb.debian.org|archive.debian.org|g" /etc/apt/sources.list 2>/dev/null || true - apt-get update -o Acquire::Check-Valid-Until=false + apt-get update apt-get install -y --no-install-recommends curl ca-certificates build-essential pkg-config if [ "${{ matrix.target }}" = "aarch64-unknown-linux-gnu" ]; then apt-get install -y --no-install-recommends gcc-aarch64-linux-gnu libc6-dev-arm64-cross @@ -316,10 +313,9 @@ jobs: - name: Build cagent (linux, Debian 11 container) if: runner.os == 'Linux' run: | - docker run --rm -v "$PWD":/github/workspace -w /github/workspace debian:11 \ + docker run --rm -v "$PWD":/github/workspace -w /github/workspace debian:12 \ bash -euxo pipefail -c ' - sed -i "s|deb.debian.org|archive.debian.org|g" /etc/apt/sources.list 2>/dev/null || true - apt-get update -o Acquire::Check-Valid-Until=false + apt-get update apt-get install -y --no-install-recommends gcc make xxd libc6-dev libcurl4-openssl-dev pkg-config mkdir -p dist make -C c CC=gcc From 1e2023e2debf209070b28f49fd4815beccec7eb7 Mon Sep 17 00:00:00 2001 From: lloydzhou Date: Thu, 10 Sep 2026 21:35:15 +0800 Subject: [PATCH 08/10] =?UTF-8?q?ci:=20=E6=9E=84=E5=BB=BA=E5=AE=B9?= =?UTF-8?q?=E5=99=A8=E6=94=B9=E7=94=A8=20ubuntu:20.04=EF=BC=88glibc=202.31?= =?UTF-8?q?=20=E5=9F=BA=E7=BA=BF=E4=B8=8D=E5=8F=98=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 替代 debian:11（bullseye 已 EOL，镜像进入元数据过期+pool 清理 半死状态）和 debian:12（glibc 2.36 会放弃 Ubuntu 22.04 等系统， 违背 v4.3.1 的 2.31 基线承诺）。 ubuntu:20.04 (focal) glibc 同为 2.31： - 门禁 GLIBC_MIN=2.31 / deb 声明 libc6 (>= 2.31) / README 承诺全部不变 - focal 仍在 archive.ubuntu.com 正常服务（主仓/security 均 200）， 且 Ubuntu Release 无 Valid-Until 字段，不存在 Debian 的过期检查问题 - 产物兼容性完全等价：Debian 11/12/13、Ubuntu 20.04/22.04/24.04+ --- .github/workflows/ci.yml | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 355e223b..b97cbc0b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -95,7 +95,7 @@ jobs: - name: Build goagent (linux, Debian 11 container) if: matrix.goos == 'linux' run: | - docker run --rm -v /opt/hostedtoolcache:/opt/hostedtoolcache -v "$PWD":/github/workspace -w /github/workspace debian:12 \ + docker run --rm -v /opt/hostedtoolcache:/opt/hostedtoolcache -v "$PWD":/github/workspace -w /github/workspace ubuntu:20.04 \ bash -euxo pipefail -c ' GO_BIN="$(ls -d /opt/hostedtoolcache/go/*/x64/bin | sort -V | tail -1)/go" export PATH="$(dirname "$GO_BIN"):$PATH" @@ -183,7 +183,7 @@ jobs: - name: Build rustagent (linux, Debian 11 container) if: matrix.runner == 'ubuntu-latest' run: | - docker run --rm -v "$PWD":/github/workspace -w /github/workspace debian:12 \ + docker run --rm -v "$PWD":/github/workspace -w /github/workspace ubuntu:20.04 \ bash -euxo pipefail -c ' apt-get update apt-get install -y --no-install-recommends curl ca-certificates build-essential pkg-config @@ -258,7 +258,7 @@ jobs: - name: Build webagent (linux, Debian 11 container) if: matrix.runner == 'ubuntu-latest' run: | - docker run --rm -v "$PWD":/github/workspace -w /github/workspace debian:12 \ + docker run --rm -v "$PWD":/github/workspace -w /github/workspace ubuntu:20.04 \ bash -euxo pipefail -c ' apt-get update apt-get install -y --no-install-recommends curl ca-certificates build-essential pkg-config @@ -313,7 +313,7 @@ jobs: - name: Build cagent (linux, Debian 11 container) if: runner.os == 'Linux' run: | - docker run --rm -v "$PWD":/github/workspace -w /github/workspace debian:12 \ + docker run --rm -v "$PWD":/github/workspace -w /github/workspace ubuntu:20.04 \ bash -euxo pipefail -c ' apt-get update apt-get install -y --no-install-recommends gcc make xxd libc6-dev libcurl4-openssl-dev pkg-config From a17de1c2648ab81495b5063600ff124752cd4a95 Mon Sep 17 00:00:00 2001 From: lloydzhou Date: Thu, 10 Sep 2026 21:38:11 +0800 Subject: [PATCH 09/10] =?UTF-8?q?ci:=20step=20=E5=90=8D=E5=90=8C=E6=AD=A5?= =?UTF-8?q?=E4=B8=BA=20Ubuntu=2020.04=20container=EF=BC=88=E6=98=BE?= =?UTF-8?q?=E7=A4=BA=E5=90=8D=E6=AE=8B=E7=95=99=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .github/workflows/ci.yml | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b97cbc0b..f84b315c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -92,7 +92,7 @@ jobs: sudo apt-get update sudo apt-get install -y gcc-aarch64-linux-gnu - - name: Build goagent (linux, Debian 11 container) + - name: Build goagent (linux, Ubuntu 20.04 container) if: matrix.goos == 'linux' run: | docker run --rm -v /opt/hostedtoolcache:/opt/hostedtoolcache -v "$PWD":/github/workspace -w /github/workspace ubuntu:20.04 \ @@ -180,7 +180,7 @@ jobs: cd rust && cargo build --release --target ${{ matrix.target }} cp target/${{ matrix.target }}/release/${{ matrix.bin }} ../dist/rustagent-${{ matrix.suffix }}${{ matrix.ext }} - - name: Build rustagent (linux, Debian 11 container) + - name: Build rustagent (linux, Ubuntu 20.04 container) if: matrix.runner == 'ubuntu-latest' run: | docker run --rm -v "$PWD":/github/workspace -w /github/workspace ubuntu:20.04 \ @@ -255,7 +255,7 @@ jobs: cd webagent && cargo build --release --target ${{ matrix.target }} cp target/${{ matrix.target }}/release/${{ matrix.bin }} ../dist/webagent-${{ matrix.suffix }}${{ matrix.ext }} - - name: Build webagent (linux, Debian 11 container) + - name: Build webagent (linux, Ubuntu 20.04 container) if: matrix.runner == 'ubuntu-latest' run: | docker run --rm -v "$PWD":/github/workspace -w /github/workspace ubuntu:20.04 \ @@ -310,7 +310,7 @@ jobs: sudo apt-get update sudo apt-get install -y ripgrep gcc-aarch64-linux-gnu - - name: Build cagent (linux, Debian 11 container) + - name: Build cagent (linux, Ubuntu 20.04 container) if: runner.os == 'Linux' run: | docker run --rm -v "$PWD":/github/workspace -w /github/workspace ubuntu:20.04 \ From 6ccb0266a5fc1f71a07a8f86770b475631f71fca Mon Sep 17 00:00:00 2001 From: lloydzhou Date: Thu, 10 Sep 2026 22:13:42 +0800 Subject: [PATCH 10/10] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E8=BE=93=E5=87=BA?= =?UTF-8?q?=E9=80=9F=E5=BA=A6=E7=BB=9F=E8=AE=A1=E4=B8=8E=E5=AE=8C=E6=95=B4?= =?UTF-8?q?=E4=BC=A0=E8=BE=93=E8=AE=A1=E6=97=B6=E5=AF=B9=E9=BD=90=20#87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- c/agent.c | 21 ++--- c/test_transport.c | 75 ++++++++++++++++++ c/transport.c | 34 +++++++- c/transport.h | 2 + go/speed_test.go | 188 ++++++++++++++++++++++++++++++++++++++++++++ go/transport.go | 68 +++++++++++++--- rust/src/sse.rs | 27 +++++-- tests/test_speed.py | 103 ++++++++++++++++++++++++ 8 files changed, 488 insertions(+), 30 deletions(-) create mode 100644 go/speed_test.go create mode 100644 tests/test_speed.py diff --git a/c/agent.c b/c/agent.c index 5cccadfa..2600fb5e 100644 --- a/c/agent.c +++ b/c/agent.c @@ -173,6 +173,7 @@ static void stream_display_callback(void *ctx, const SseEvent *evt) { if (evt->cache_creation_tokens > 0) sctx->accum.cache_creation_tokens = evt->cache_creation_tokens; sctx->accum.start_ms = evt->start_ms; sctx->accum.end_ms = evt->end_ms; + sctx->accum.speed_ready = evt->speed_ready; break; case SSE_TOOL_CALL_START: { @@ -224,6 +225,7 @@ static void stream_display_callback(void *ctx, const SseEvent *evt) { } sctx->accum.tool_count = 0; sctx->accum.stopped = 0; + sctx->accum.speed_ready = 0; if (sctx->accum.stop_reason) { free(sctx->accum.stop_reason); sctx->accum.stop_reason = NULL; } if (sctx->accum.error) { free(sctx->accum.error); sctx->accum.error = NULL; } sctx->accum.in_tokens = 0; @@ -1031,16 +1033,17 @@ int agent_loop(Agent *agent, const char *user_input, const char *turn_kind) { store_stats_set_int_file(agent->paths.stats, "current_context_tokens", agent->last_context_tokens); } - /* 对齐 bash 版：speed = output_tokens / (end_ms - start_ms) * 1000，duration > 0 - * 且仅在流正常终结（收到 stop）时更新——中断场景 bash 无 USAGE、保留旧值 */ - if (accum->stopped) { - long long dur = agent->last_end_ms - agent->last_start_ms; - int speed = (dur > 0) ? (int)((long long)agent->last_output_tokens * 1000LL / dur) : 0; - store_stats_set_int_file(agent->paths.stats, "last_call_speed_tok_per_sec", speed); - } - /* 对齐 bash 版 store_stats_update 末尾的 display_term_title */ - agent_update_title(agent); } + /* 仅完整协议终结且 HTTP 成功后的最终 USAGE 才更新速度。 + * 独立于正 token 守卫：零用量成功需要清零，失败和中断保留旧值。 */ + if (accum->speed_ready && !accum->error && !agent->interrupted) { + long long dur = agent->last_end_ms - agent->last_start_ms; + int speed = (dur > 0) ? (int)((long long)agent->last_output_tokens * 1000LL / dur) : 0; + store_stats_set_int_file(agent->paths.stats, "last_call_speed_tok_per_sec", speed); + } + /* 对齐 bash 版 store_stats_update 末尾的 display_term_title */ + if (accum->in_tokens > 0 || accum->out_tokens > 0 || accum->speed_ready) + agent_update_title(agent); /* ---- 执行工具调用 ---- */ const char **result_ids = NULL; diff --git a/c/test_transport.c b/c/test_transport.c index 3c3f2885..cee3f2c9 100644 --- a/c/test_transport.c +++ b/c/test_transport.c @@ -42,7 +42,80 @@ static void check(int condition, const char *label) { } } +/* 走实际逐行解析器，分别检查提前 STOP、协议终结和最终 HTTP 门槛。 */ +static void feed_stream(StreamCtx *stream, const char *wire) { + check(stream_cb((char *)wire, 1, strlen(wire), stream) == strlen(wire), "流数据完整接收"); +} + +static void test_speed_completion(void) { + SseAccumulator acc; + sse_accum_init(&acc); + StreamCtx stream = {0}; + stream.callback = sse_accum_callback; + stream.ctx = &acc; + stream.provider = "claude"; + sb_init(&stream.line_buf); + + feed_stream(&stream, "data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\"},\"usage\":{\"output_tokens\":10}}\n\n"); + check(acc.stopped && !acc.speed_ready, "提前停止原因不是最终用量"); + check(!stream_speed_ready(&stream, CURLE_OK, 200), "Claude 有停止原因但缺少 message_stop 的提前 EOF 保留速度"); + feed_stream(&stream, "data: {\"type\":\"message_stop\"}\n\n"); + check(stream_speed_ready(&stream, CURLE_OK, 200), "Claude 完整终结且 HTTP 成功允许更新"); + check(!acc.speed_ready, "仅协议终结尚未发最终用量时不更新"); + check(!stream_speed_ready(&stream, CURLE_PARTIAL_FILE, 200), "协议完整但 HTTP 截断保留速度"); + check(!stream_speed_ready(&stream, CURLE_OK, 401), "HTTP 失败保留速度"); + volatile int cancelled = 1; + stream.cancelled = &cancelled; + check(!stream_speed_ready(&stream, CURLE_OK, 200), "协议终结后取消仍保留速度"); + check(!stream_speed_ready(&stream, CURLE_WRITE_ERROR, 200), "取消引发写入错误保留速度"); + emit_simple_event(stream.callback, stream.ctx, SSE_STOP, "interrupted"); + check(acc.stopped && !acc.speed_ready, "中断 STOP 不允许更新速度"); + stream.cancelled = NULL; + + SseEvent final_usage = {0}; + final_usage.type = SSE_USAGE; + final_usage.speed_ready = stream_speed_ready(&stream, CURLE_OK, 200); + final_usage.end_ms = 100; + sse_accum_callback(&acc, &final_usage); + check(acc.speed_ready && acc.out_tokens == 10, "最终用量允许更新且保留累计 token"); + emit_simple_event(stream.callback, stream.ctx, SSE_RETRY, NULL); + check(!acc.speed_ready && acc.out_tokens == 0, "重试清除速度门槛及用量"); + sb_free(&stream.line_buf); + + /* 与传输重试一样，每次尝试使用全新的上下文。 */ + memset(&stream, 0, sizeof(stream)); + stream.callback = sse_accum_callback; + stream.ctx = &acc; + stream.provider = "openai"; + sb_init(&stream.line_buf); + check(!stream_speed_ready(&stream, CURLE_OK, 200), "重试不继承协议终结状态"); + feed_stream(&stream, "data: {\"object\":\"chat.completion.chunk\",\"choices\":[{\"finish_reason\":\"stop\"}]}\n\n"); + check(acc.stopped && !stream_speed_ready(&stream, CURLE_OK, 200), "OpenAI 有 finish_reason 但缺少 DONE 的提前 EOF 保留速度"); + feed_stream(&stream, "data: [DONE]\n\n"); + check(stream_speed_ready(&stream, CURLE_OK, 200), "OpenAI 完整终结允许更新"); + final_usage.speed_ready = stream_speed_ready(&stream, CURLE_OK, 200); + sse_accum_callback(&acc, &final_usage); + check(acc.speed_ready && acc.out_tokens == 0, "零用量成功仍允许清零速度"); + stream.protocol_complete = stream.openai_finished = 0; + feed_stream(&stream, "data: [DONE]\n\n"); + check(!stream_speed_ready(&stream, CURLE_OK, 200), "仅 DONE 无 finish_reason 不更新"); + sb_free(&stream.line_buf); + + const char *failed_events[] = {"response.failed", "response.incomplete"}; + for (int i = 0; i < 2; i++) { + memset(&stream, 0, sizeof(stream)); + stream.callback = sse_accum_callback; + stream.ctx = &acc; + parse_responses_sse_event(&stream, failed_events[i], "{}", 2); + check(stream.responses_terminal && !stream_speed_ready(&stream, CURLE_OK, 200), + "Responses 失败或不完整终结保留速度"); + streamctx_free_openai_tools(&stream); + } + sse_accum_free(&acc); +} + int main(void) { + test_speed_completion(); Capture capture = {0}; StreamCtx stream = {0}; stream.callback = capture_event; @@ -55,6 +128,7 @@ int main(void) { parse_responses_sse_event(&stream, "response.function_call_arguments.delta", delta, strlen(delta)); parse_responses_sse_event(&stream, "response.completed", completed, strlen(completed)); + check(stream_speed_ready(&stream, CURLE_OK, 200), "Responses completed 允许最终用量更新速度"); check(capture.count == 3, "Responses completed emits tool, usage, stop"); check(capture.events[0].type == SSE_TOOL_CALL && strcmp(capture.events[0].tool_name, "Read") == 0 && strcmp(capture.events[0].tool_id, "call_1") == 0 && strcmp(capture.events[0].tool_input, "{\"path\":\"/tmp/a\"}") == 0, "Responses maps item id and argument delta"); check(capture.events[1].type == SSE_USAGE && capture.events[1].in_tokens == 9 && capture.events[1].out_tokens == 8 && capture.events[1].cache_read_tokens == 6, "Responses prefers nested cached tokens"); @@ -68,6 +142,7 @@ int main(void) { stream.ctx = &capture; const char *error = "{\"reason\":\"upstream failed\"}"; parse_responses_sse_event(&stream, "error", error, strlen(error)); + check(!stream_speed_ready(&stream, CURLE_OK, 200), "Responses 错误终结不更新速度"); check(capture.count == 3, "Responses error emits error, usage, stop"); check(capture.events[0].type == SSE_ERROR && strcmp(capture.events[0].content, "upstream failed") == 0, "Responses error uses reason fallback"); check(capture.events[1].type == SSE_USAGE, "Responses error emits usage event"); diff --git a/c/transport.c b/c/transport.c index 4270b914..2e412b14 100644 --- a/c/transport.c +++ b/c/transport.c @@ -59,6 +59,8 @@ typedef struct { int openai_tool_cap; int responses_saw_text; int responses_terminal; + int protocol_complete; /* 独立于提前发出的 STOP */ + int openai_finished; int responses_input_tokens; int responses_output_tokens; int responses_cache_read_tokens; @@ -132,7 +134,10 @@ static void streamctx_emit_openai_tool_calls(StreamCtx *sctx) { static void parse_openai_sse_event(StreamCtx *sctx, const char *data, size_t data_len) { if (data_len == 0) return; - if (strcmp(data, "[DONE]") == 0) return; + if (strcmp(data, "[DONE]") == 0) { + sctx->protocol_complete = sctx->openai_finished; + return; + } size_t pos = 0; JsonParse jp = json_parse(data, &pos); @@ -195,6 +200,7 @@ static void parse_openai_sse_event(StreamCtx *sctx, const char *data, size_t dat /* 非标准 API 可能用空字符串 "" 代替 null（如 sensenova）， * 空字符串不应触发 STOP */ if (finish && finish[0]) { + sctx->openai_finished = 1; if (strcmp(finish, "tool_calls") == 0) { streamctx_emit_openai_tool_calls(sctx); emit_simple_event(sctx->callback, sctx->ctx, SSE_STOP, "tool_use"); @@ -325,6 +331,7 @@ static void parse_responses_sse_event(StreamCtx *sctx, const char *event, const responses_emit_usage(sctx); emit_simple_event(sctx->callback, sctx->ctx, SSE_STOP, has_tools ? "tool_use" : "end_turn"); sctx->responses_terminal = 1; + sctx->protocol_complete = 1; } else if (strcmp(event, "response.failed") == 0 || strcmp(event, "response.incomplete") == 0 || strcmp(event, "error") == 0) { JsonVal response = json_get(root, "response"); if (response.type == JSON_NULL) response = root; @@ -342,6 +349,24 @@ static void parse_responses_sse_event(StreamCtx *sctx, const char *event, const } } +static void parse_claude_stream_event(StreamCtx *sctx, const char *data) { + size_t pos = 0; + JsonParse jp = json_parse(data, &pos); + if (!jp.error) { + char *type = json_get_string(jp.val, "type"); + if (type && strcmp(type, "message_stop") == 0) + sctx->protocol_complete = 1; + FREE_PTR(type); + } + sse_parse_event(sctx->provider, data, strlen(data), sctx->callback, sctx->ctx, sctx->start_ms); +} + +/* 每次重试重新初始化 StreamCtx，协议完成状态不会跨尝试继承。 */ +static int stream_speed_ready(const StreamCtx *sctx, CURLcode rc, long http_code) { + return rc == CURLE_OK && http_code >= 200 && http_code < 300 && + !(sctx->cancelled && *sctx->cancelled) && sctx->protocol_complete; +} + static size_t stream_cb(char *ptr, size_t size, size_t nmemb, void *userdata) { StreamCtx *sctx = (StreamCtx *)userdata; @@ -373,13 +398,13 @@ static size_t stream_cb(char *ptr, size_t size, size_t nmemb, void *userdata) { const char *data = line + 6; if (strcmp(sctx->provider, "openai") == 0) parse_openai_sse_event(sctx, data, strlen(data)); else if (strcmp(sctx->provider, "responses") == 0) parse_responses_sse_event(sctx, sctx->event ? sctx->event : "", data, strlen(data)); - else sse_parse_event(sctx->provider, data, strlen(data), sctx->callback, sctx->ctx, sctx->start_ms); + else parse_claude_stream_event(sctx, data); } else if (strncmp(line, "data:", 5) == 0) { const char *data = line + 5; while (*data == ' ') data++; if (strcmp(sctx->provider, "openai") == 0) parse_openai_sse_event(sctx, data, strlen(data)); else if (strcmp(sctx->provider, "responses") == 0) parse_responses_sse_event(sctx, sctx->event ? sctx->event : "", data, strlen(data)); - else sse_parse_event(sctx->provider, data, strlen(data), sctx->callback, sctx->ctx, sctx->start_ms); + else parse_claude_stream_event(sctx, data); } /* 重置行缓冲 */ sb_truncate(&sctx->line_buf, 0); @@ -662,6 +687,7 @@ int http_post_sse(const char *url, const char **headers, int header_count, SseEvent tevt; memset(&tevt, 0, sizeof(tevt)); tevt.type = SSE_USAGE; + tevt.speed_ready = stream_speed_ready(&sctx, rc, http_code); tevt.start_ms = 0; /* 微秒向上取整到毫秒：亚毫秒传输至少 1ms，避免整数截断 speed=0 */ tevt.end_ms = (dur_us > 0) ? (long long)((dur_us + 999) / 1000) : 0; @@ -1020,6 +1046,7 @@ void sse_accum_callback(void *ctx, const SseEvent *evt) { if (evt->cache_creation_tokens > 0) acc->cache_creation_tokens = evt->cache_creation_tokens; acc->start_ms = evt->start_ms; acc->end_ms = evt->end_ms; + acc->speed_ready = evt->speed_ready; break; case SSE_STOP: @@ -1046,6 +1073,7 @@ void sse_accum_callback(void *ctx, const SseEvent *evt) { } acc->tool_count = 0; acc->stopped = 0; + acc->speed_ready = 0; FREE_PTR(acc->stop_reason); acc->in_tokens = 0; acc->out_tokens = 0; diff --git a/c/transport.h b/c/transport.h index 6bcff7bd..92d10c1f 100644 --- a/c/transport.h +++ b/c/transport.h @@ -36,6 +36,7 @@ typedef struct { int cache_creation_tokens; /* USAGE: 缓存创建 token */ long long start_ms; /* USAGE: 流开始时间戳（毫秒） */ long long end_ms; /* USAGE: 流结束时间戳（毫秒） */ + int speed_ready; /* 仅协议完整且 HTTP 成功结束后的最终 USAGE */ } SseEvent; typedef void (*sse_callback_fn)(void *ctx, const SseEvent *evt); @@ -110,6 +111,7 @@ typedef struct { /* 状态标记 */ int stopped; /* 收到 stop 事件 */ + int speed_ready; /* 最终 USAGE 允许更新速度，不等同于 stop */ } SseAccumulator; /* 初始化/释放累积器 */ diff --git a/go/speed_test.go b/go/speed_test.go new file mode 100644 index 00000000..3ff5c96e --- /dev/null +++ b/go/speed_test.go @@ -0,0 +1,188 @@ +package agent + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "path/filepath" + "testing" + "time" +) + +var speedStreams = map[string]string{ + "claude": "event: message_start\ndata: {\"message\":{\"usage\":{\"input_tokens\":12}}}\n\nevent: message_delta\ndata: {\"delta\":{\"stop_reason\":\"end_turn\"},\"usage\":{\"output_tokens\":100}}\n\nevent: message_stop\ndata: {}\n\n", + "openai": "data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}],\"usage\":{\"prompt_tokens\":12,\"completion_tokens\":100}}\n\ndata: [DONE]\n\n", + "responses": "event: response.completed\ndata: {\"response\":{\"usage\":{\"input_tokens\":12,\"output_tokens\":100}}}\n\n", +} + +// 终态已刷新到客户端后，服务器仍保持 HTTP 正文打开。 +func TestSpeedWaitsForHTTPEOF(t *testing.T) { + for provider, stream := range speedStreams { + t.Run(provider, func(t *testing.T) { + terminal := make(chan struct{}) + release := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprint(w, stream) + w.(http.Flusher).Flush() + close(terminal) + <-release + fmt.Fprint(w, ": 尾部\n\n") + })) + defer server.Close() + tr := NewHTTPTransport(Config{Provider: provider, BaseURL: server.URL}) + ch, err := tr.Call(context.Background(), "[]", "", "[]", 128, "disabled") + if err != nil { + close(release) + t.Fatal(err) + } + <-terminal + timer := time.NewTimer(180 * time.Millisecond) + early := false + wait: + for { + select { + case ev, ok := <-ch: + if !ok { + early = true + break wait + } + if ev.Type == EventUsage || ev.Type == EventStop { + early = true + } + case <-timer.C: + break wait + } + } + timer.Stop() + releasedAt := time.Now().UnixMilli() + close(release) + var usage Usage + count, stops := 0, 0 + for ev := range ch { + if ev.Type == EventError { + t.Errorf("意外错误: %v", ev.Fields) + } + if ev.Type == EventUsage { + usage = ev.Payload.(Usage) + count++ + } + if ev.Type == EventStop { + stops++ + } + } + if early { + t.Error("HTTP 结束前发布了用量或停止事件") + } + if count != 1 || stops != 1 || !usage.Stopped || usage.OutputTokens != 100 || usage.InputTokens != 12 { + t.Fatalf("终态不正确: 用量=%+v 次数=%d 停止=%d", usage, count, stops) + } + if usage.EndMs < releasedAt || usage.EndMs-usage.StartMs < 170 { + t.Fatalf("计时未包含尾部等待: %+v", usage) + } + store := &FileStore{statsFile: filepath.Join(t.TempDir(), "stats.json")} + if err := store.UpdateStats(usage, ""); err != nil { + t.Fatal(err) + } + want := 100000 / int(usage.EndMs-usage.StartMs) + if got := store.GetStats().LastCallSpeedTokPerSec; got != want { + t.Fatalf("速度=%d 预期=%d", got, want) + } + }) + } +} + +func TestSpeedFailurePreservesPrevious(t *testing.T) { + for provider, stream := range speedStreams { + modes := []string{"truncated", "cancel", "missing", "error", "retry-missing", "retry-success"} + if provider == "responses" { + modes = append(modes, "response.failed", "response.incomplete") + } + for _, mode := range modes { + t.Run(provider+"/"+mode, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + if mode == "truncated" { + w.Header().Set("Content-Length", "100000") + } + if mode != "missing" { + fmt.Fprint(w, stream) + } else { + fmt.Fprint(w, ": 未终结\n\n") + } + w.(http.Flusher).Flush() + switch mode { + case "cancel": + time.Sleep(40 * time.Millisecond) + cancel() + <-r.Context().Done() + case "error": + fmt.Fprint(w, "event: error\ndata: {\"error\":{\"message\":\"失败\"}}\n\n") + case "response.failed", "response.incomplete": + fmt.Fprintf(w, "event: %s\ndata: {\"response\":{\"error\":{\"message\":\"失败\"},\"usage\":{\"output_tokens\":10}}}\n\n", mode) + case "retry-missing", "retry-success": + fmt.Fprint(w, "event: retry\ndata: {}\n\n") + w.(http.Flusher).Flush() + time.Sleep(40 * time.Millisecond) + if mode == "retry-success" { + fmt.Fprint(w, stream) + } + } + })) + defer server.Close() + store := &FileStore{statsFile: filepath.Join(t.TempDir(), "stats.json")} + if err := store.UpdateStats(Usage{Stopped: true, OutputTokens: 37, StartMs: 1000, EndMs: 2000}, ""); err != nil { + t.Fatal(err) + } + tr := NewHTTPTransport(Config{Provider: provider, BaseURL: server.URL}) + ch, err := tr.Call(ctx, "[]", "", "[]", 128, "disabled") + if err != nil { + t.Fatal(err) + } + successes, retries, errors, stops := 0, 0, 0, 0 + for ev := range ch { + switch ev.Type { + case EventUsage: + u := ev.Payload.(Usage) + if u.Stopped { + successes++ + } + if err := store.UpdateStats(u, ""); err != nil { + t.Fatal(err) + } + case EventRetry: + retries++ + case EventError: + errors++ + case EventStop: + stops++ + if mode != "retry-success" && (len(ev.Fields) < 2 || ev.Fields[1] != "error") { + t.Errorf("失败流停止原因错误: %v", ev.Fields) + } + } + } + if stops != 1 { + t.Errorf("停止事件数=%d", stops) + } + if mode == "retry-success" { + if successes != 1 || retries != 1 || errors != 0 { + t.Fatalf("重试成功状态错误: 成功=%d 重试=%d 错误=%d", successes, retries, errors) + } + } else { + if successes != 0 || errors == 0 { + t.Errorf("失败流状态错误: 成功=%d 错误=%d", successes, errors) + } + if got := store.GetStats().LastCallSpeedTokPerSec; got != 37 { + t.Errorf("失败覆盖旧速度: %d", got) + } + if mode == "retry-missing" && retries != 1 { + t.Errorf("重试事件数=%d", retries) + } + } + }) + } + } +} diff --git a/go/transport.go b/go/transport.go index e1ff0608..058c89a6 100644 --- a/go/transport.go +++ b/go/transport.go @@ -172,6 +172,50 @@ func NewHTTPTransport(cfg Config) *HTTPTransport { // parseSSEStream 从 HTTP response body 读取 SSE 流，解析为 Event func (t *HTTPTransport) parseSSEStream(ctx context.Context, resp *http.Response, ch chan<- Event) { + defer close(ch) + events := make(chan Event) + go t.decodeSSEStream(ctx, resp, events) + + // 协议终态不代表 HTTP 已结束。暂存用量和停止事件，读完后再发布。 + var pendingUsage *Usage + var pendingStop *Event + var failed bool + for ev := range events { + switch ev.Type { + case EventUsage: + u := ev.Payload.(Usage) + pendingUsage = &u + case EventStop: + stop := ev + pendingStop = &stop + case EventError: + failed = true + ch <- ev + case EventRetry: + pendingUsage = nil + pendingStop = nil + failed = false + ch <- ev + default: + ch <- ev + } + } + if ctx.Err() != nil { + failed = true + } + if pendingUsage != nil { + pendingUsage.EndMs = time.Now().UnixMilli() + pendingUsage.Stopped = pendingUsage.Stopped && !failed + ch <- Event{Type: EventUsage, Payload: *pendingUsage} + } + if failed { + ch <- Event{Type: EventStop, Fields: []string{"STOP", "error"}} + } else if pendingStop != nil { + ch <- *pendingStop + } +} + +func (t *HTTPTransport) decodeSSEStream(ctx context.Context, resp *http.Response, ch chan<- Event) { var stopEmitted bool streamDone := make(chan struct{}) defer close(streamDone) @@ -196,9 +240,8 @@ func (t *HTTPTransport) parseSSEStream(ctx context.Context, resp *http.Response, } }() - // 对齐 bash 版 claude_sse.awk：流开始时刻记录 start_ms，结束时算 speed + // 起点保持在响应头之后，终点由 parseSSEStream 在 HTTP 读完后设置。 startMs := time.Now().UnixMilli() - endMs := func() int64 { return time.Now().UnixMilli() } scanner := bufio.NewScanner(resp.Body) scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) @@ -293,15 +336,14 @@ func (t *HTTPTransport) parseSSEStream(ctx context.Context, resp *http.Response, case "message_stop": stopEmitted = true - // 发送 USAGE + STOP（对齐 bash 版：带 start_ms/end_ms 用于计算 tok/s） + // 交给外层暂存，等待完整 HTTP 结束后发布。 ch <- Event{Type: EventUsage, Payload: Usage{ - Stopped: true, + Stopped: true, InputTokens: inputTokens, OutputTokens: outputTokens, CacheRead: cacheRead, CacheWrite: cacheCreate, StartMs: startMs, - EndMs: endMs(), }} ch <- Event{Type: EventStop, Fields: []string{"STOP", stopReason}} @@ -311,6 +353,8 @@ func (t *HTTPTransport) parseSSEStream(ctx context.Context, resp *http.Response, case "retry": // 对齐 Rust/C: 重置所有累积状态 + stopEmitted = false + eventType = "" blockType = "" toolName = "" toolID = "" @@ -329,6 +373,11 @@ func (t *HTTPTransport) parseSSEStream(ctx context.Context, resp *http.Response, } } } + if err := scanner.Err(); err != nil { + ch <- Event{Type: EventError, Fields: []string{"ERROR", err.Error()}} + } else if err := ctx.Err(); err != nil { + ch <- Event{Type: EventError, Fields: []string{"ERROR", err.Error()}} + } } type openAIPendingCall struct { @@ -501,8 +550,7 @@ func (t *HTTPTransport) handleResponsesEvent(eventType, data string, ch chan<- E OutputTokens: *outputTokens, CacheRead: *cacheRead, StartMs: startMs, - EndMs: time.Now().UnixMilli(), - Stopped: true, + Stopped: true, }} stopReason := "end_turn" if hasTools { @@ -543,8 +591,7 @@ func (t *HTTPTransport) handleResponsesEvent(eventType, data string, ch chan<- E OutputTokens: *outputTokens, CacheRead: *cacheRead, StartMs: startMs, - EndMs: time.Now().UnixMilli(), - Stopped: false, + Stopped: false, }} ch <- Event{Type: EventStop, Fields: []string{"STOP", "error"}} return true @@ -596,8 +643,7 @@ func (t *HTTPTransport) handleOpenAIChunk(data string, ch chan<- Event, CacheRead: *cacheRead, CacheWrite: *cacheCreate, StartMs: startMs, - EndMs: time.Now().UnixMilli(), - Stopped: true, + Stopped: true, }} ch <- Event{Type: EventStop, Fields: []string{"STOP", sr}} return diff --git a/rust/src/sse.rs b/rust/src/sse.rs index af6782b8..455c27e0 100644 --- a/rust/src/sse.rs +++ b/rust/src/sse.rs @@ -409,7 +409,7 @@ pub mod openai { stopped: true, }); pending_stop = Some(stop_reason.clone()); - break; + continue; } let body: Value = serde_json::from_str(payload)?; @@ -511,7 +511,9 @@ pub mod openai { } } - if let Some(usage) = pending_usage { + if let Some(mut usage) = pending_usage { + // 与 Bash 一致：终态只暂存用量，完整读流结束后才结束计时。 + usage.end_ms = now_ms(); emit(Event::Usage(usage))?; } if let Some(reason) = pending_stop { @@ -573,6 +575,8 @@ pub mod responses { let mut calls: BTreeMap = BTreeMap::new(); let mut item_indexes: BTreeMap = BTreeMap::new(); let mut completed = false; + let mut pending_usage = None; + let mut pending_stop = None; let start_ms = now_ms(); while { line.clear(); @@ -588,6 +592,8 @@ pub mod responses { calls.clear(); item_indexes.clear(); completed = false; + pending_usage = None; + pending_stop = None; emit(Event::Retry(RetryEvent {}))?; continue; } @@ -648,18 +654,18 @@ pub mod responses { ); let has_tools = !calls.is_empty(); emit_calls(&mut calls, &mut emit)?; - emit(Event::Usage(UsageEvent { + pending_usage = Some(UsageEvent { input_tokens, output_tokens, cache_read_input_tokens, cache_creation_input_tokens: 0, start_ms, - end_ms: now_ms(), + end_ms: 0, stopped: true, - }))?; - emit(Event::Stop(StopEvent { + }); + pending_stop = Some(StopEvent { reason: if has_tools { "tool_use" } else { "end_turn" }.to_string(), - }))?; + }); completed = true; } "response.failed" | "response.incomplete" | "error" => { @@ -703,6 +709,13 @@ pub mod responses { _ => {} } } + if let Some(mut usage) = pending_usage { + usage.end_ms = now_ms(); + emit(Event::Usage(usage))?; + } + if let Some(stop) = pending_stop { + emit(Event::Stop(stop))?; + } if !completed { emit(Event::Error(ErrorEvent { message: "Stream interrupted (no response.completed received)".to_string(), diff --git a/tests/test_speed.py b/tests/test_speed.py new file mode 100644 index 00000000..8c345e6b --- /dev/null +++ b/tests/test_speed.py @@ -0,0 +1,103 @@ +#!/usr/bin/env python3 +"""串行本地 SSE 回归：速度必须包含终态事件之后的读流时间。""" +import http.server +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import threading +import time + + +class Handler(http.server.BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def do_POST(self): + self.rfile.read(int(self.headers.get('Content-Length', 0))) + mode, provider = self.server.mode, self.server.provider + self.send_response(200) + self.send_header('Content-Type', 'text/event-stream') + self.end_headers() + self.wfile.flush() + + def event(name, data): + if provider == 'claude' and isinstance(data, dict): + data = dict(data, type=name) + elif provider == 'openai' and isinstance(data, dict) and 'choices' in data: + data = dict(data, object='chat.completion.chunk') + prefix = f'event: {name}\n' if name else '' + value = data if isinstance(data, str) else json.dumps(data) + self.wfile.write(f'{prefix}data: {value}\n\n'.encode()) + self.wfile.flush() + + if provider == 'claude': + event('message_start', {'message': {'usage': {'input_tokens': 10}}}) + time.sleep(0.15) + if mode == 'failure': + event('error', {'type': 'error', 'error': {'message': 'test failure'}, 'message': 'test failure'}) + return + if mode == 'eof': + if provider == 'claude': + event('message_delta', {'delta': {'stop_reason': 'end_turn'}, 'usage': {'output_tokens': 50}}) + elif provider == 'openai': + event('', {'choices': [{'index': 0, 'delta': {}, 'finish_reason': 'stop'}], + 'usage': {'prompt_tokens': 10, 'completion_tokens': 50}}) + return + output = 0 if mode == 'zero' else 100 + input_tokens = 0 if mode == 'zero' else 10 + if provider == 'claude': + event('message_start', {'message': {'usage': {'input_tokens': input_tokens}}}) + event('message_delta', {'delta': {'stop_reason': 'end_turn'}, 'usage': {'output_tokens': output}}) + event('message_stop', {}) + elif provider == 'openai': + event('', {'choices': [{'index': 0, 'delta': {'content': 'ok'}, 'finish_reason': 'stop'}], + 'usage': {'prompt_tokens': input_tokens, 'completion_tokens': output}}) + event('', '[DONE]') + else: + event('response.completed', {'response': {'status': 'completed', 'output': [], + 'usage': {'input_tokens': input_tokens, 'output_tokens': output}}}) + # 若在 message_stop / DONE / completed 计时，速度约 666，而非约 86。 + time.sleep(1.0) + + +def main(): + agent = str(Path(sys.argv[1]).resolve()) + server = http.server.ThreadingHTTPServer(('127.0.0.1', 0), Handler) + threading.Thread(target=server.serve_forever, daemon=True).start() + try: + for provider in ('claude', 'openai', 'responses'): + server.provider = provider + with tempfile.TemporaryDirectory(prefix='agent-speed-') as home: + env = dict(os.environ, BASH_AGENT_HOME=home) + command = [agent, '-p', provider, '--base-url', f'http://127.0.0.1:{server.server_port}/v1', + '-m', 'test', '--api-key', 'test', '--session', 'speed-regression', 'speed test'] + + def run(mode): + server.mode = mode + result = subprocess.run(command, env=env, capture_output=True, timeout=30) + files = list(Path(home).rglob('stats.json')) + assert len(files) == 1, (provider, mode, result.stderr.decode(errors='replace')) + speed = json.loads(files[0].read_text())['last_call_speed_tok_per_sec'] + assert type(speed) is int, (provider, mode, speed) + return speed, result + + speed, result = run('success') + assert result.returncode == 0, result.stderr.decode(errors='replace') + assert 40 <= speed <= 110, (provider, '计时未覆盖完整流', speed) + for mode in ('failure', 'eof'): + actual, _ = run(mode) + assert actual == speed, (provider, mode, '覆盖上次速度', speed, actual) + actual, result = run('zero') + assert result.returncode == 0, result.stderr.decode(errors='replace') + assert actual == 0, (provider, '零输出应清零速度', actual) + print(f'{Path(agent).name} {provider}: 速度={speed}，失败/提前结束保留，零输出清零', flush=True) + finally: + server.shutdown() + server.server_close() + + +if __name__ == '__main__': + main()