From 7527a832f74bcc8a9e37a5d5267e7ec8cf94c116 Mon Sep 17 00:00:00 2001 From: "wujiaming.ai" Date: Wed, 28 Oct 2026 13:04:45 +0800 Subject: [PATCH 1/2] feat(tracing): add request and first-token timestamps --- docs/content/docs/framework/tracing.en.mdx | 19 + docs/content/docs/framework/tracing.mdx | 19 + .../2026-10-28-first-token-timestamps.md | 114 ++++++ .../2026-10-28-first-token-timestamps.zh.md | 114 ++++++ specs/model-timing/README.md | 30 ++ specs/model-timing/README.zh.md | 30 ++ tests/cloud/test_harness_app_http.py | 2 + tests/models/test_model_timing_adapters.py | 374 ++++++++++++++++++ tests/test_model_timing.py | 309 +++++++++++++++ veadk/cloud/harness_app/app.py | 3 + veadk/integrations/agentkit/app.py | 2 + veadk/models/ark_llm.py | 33 +- veadk/models/retrying_lite_llm.py | 6 +- veadk/models/timed_lite_llm.py | 82 ++++ veadk/runner.py | 3 +- .../telemetry/exporters/inmemory_exporter.py | 2 + veadk/tracing/telemetry/telemetry.py | 10 + veadk/tracing/telemetry/timing.py | 222 +++++++++++ 18 files changed, 1360 insertions(+), 14 deletions(-) create mode 100644 prd-spec/features/first-token-timestamps/2026-10-28-first-token-timestamps.md create mode 100644 prd-spec/features/first-token-timestamps/2026-10-28-first-token-timestamps.zh.md create mode 100644 specs/model-timing/README.md create mode 100644 specs/model-timing/README.zh.md create mode 100644 tests/models/test_model_timing_adapters.py create mode 100644 tests/test_model_timing.py create mode 100644 veadk/models/timed_lite_llm.py create mode 100644 veadk/tracing/telemetry/timing.py diff --git a/docs/content/docs/framework/tracing.en.mdx b/docs/content/docs/framework/tracing.en.mdx index 1b7a16663..5b69bad3d 100644 --- a/docs/content/docs/framework/tracing.en.mdx +++ b/docs/content/docs/framework/tracing.en.mdx @@ -187,3 +187,22 @@ In a multi-agent scenario, spans are linked through `parent_span_id`, showing th VeADK's field naming follows the OpenTelemetry generative-AI conventions, so you can import tracing data directly into any OpenTelemetry-compatible system for analysis and visualization. + + +## Request, model, and first-token timestamps + +VeADK's default LiteLLM and Ark Responses adapters add the following extension attributes to model spans when OpenTelemetry tracing is enabled. UTC Unix nanoseconds are used throughout. + +| Attribute | Meaning | +| --- | --- | +| `veadk.request.start_time_unix_ns` | HTTP ASGI entry in VeADK AgentKit/Harness apps, or direct VeADK Runner execution entry. | +| `veadk.request.start_source` | `http` or `runner`; an inherited HTTP timestamp takes precedence. | +| `veadk.model.start_time_unix_ns` | Start of one adapter SDK request, not provider inference start. | +| `veadk.model.streaming` | Actual model SDK request mode, independent of HTTP SSE transport. | +| `veadk.model.first_token_time_unix_ns` | Receipt of the first nonempty generated content delta. | +| `veadk.model.first_token_type` | `text`, `reasoning`, or `tool_call`. | +| `veadk.model.first_token_status` | `observed`, `non_streaming`, `no_output`, or `unsupported`. | + +First-token capture excludes role/id/signature metadata, empty deltas, usage, heartbeats, and completion-only aggregates. It does not require token usage or content tracing. One chunk may contain several tokens, so this is an SDK-observed first-output timestamp. Non-streaming calls have no first-token timestamp; completion time is never substituted. + +Each SDK request creates a `model_request` child span; the owning model span also receives the latest attempt's fields. VeADK-issued retries and fallbacks create separate attempts. Internal LiteLLM/provider retries remain within the observed SDK request (`veadk.model.request_scope=sdk_request`). Unsupported external/custom engines are explicitly marked; invocation span start is not used as model start. Applications outside the VeADK HTTP factories must install the request timing middleware or use VeADK Runner to establish a request timestamp. Client-render timing and TTFT/TPOT histogram reporting are separate capabilities. diff --git a/docs/content/docs/framework/tracing.mdx b/docs/content/docs/framework/tracing.mdx index a5adaf3fc..a03ca75ee 100644 --- a/docs/content/docs/framework/tracing.mdx +++ b/docs/content/docs/framework/tracing.mdx @@ -185,3 +185,22 @@ Tracing 数据由一系列 **Span** 组成，每个 Span 记录一次操作（ VeADK 的字段命名遵循 OpenTelemetry 生成式 AI 规范，你可以直接将 Tracing 数据导入到 OpenTelemetry 兼容的系统中进行分析和可视化。 + + +## 请求、模型与首 Token 时间戳 + +启用 OpenTelemetry Tracing 后，VeADK 默认 LiteLLM 和 Ark Responses 适配器会在模型 Span 上添加以下扩展属性。时间戳统一采用 UTC Unix 纳秒。 + +| 属性 | 含义 | +| --- | --- | +| `veadk.request.start_time_unix_ns` | VeADK AgentKit/Harness 应用的 HTTP ASGI 入口，或直接调用 VeADK Runner 的执行入口。 | +| `veadk.request.start_source` | `http` 或 `runner`；继承的 HTTP 时间戳优先。 | +| `veadk.model.start_time_unix_ns` | 一次适配器 SDK 请求的开始，不是服务端开始推理。 | +| `veadk.model.streaming` | 实际模型 SDK 请求模式，与 HTTP SSE 传输独立。 | +| `veadk.model.first_token_time_unix_ns` | 收到首个非空生成内容增量的时间。 | +| `veadk.model.first_token_type` | `text`、`reasoning` 或 `tool_call`。 | +| `veadk.model.first_token_status` | `observed`、`non_streaming`、`no_output` 或 `unsupported`。 | + +首 Token 采集排除角色/id/签名元数据、空增量、usage、心跳和仅完成时返回的聚合内容。不依赖 token usage 或内容采集。一个 chunk 可含多个 token，因此这里表示 SDK 侧观测到首输出的时间。非流式调用没有首 Token 时间戳，不能用响应完成时间替代。 + +每次 SDK 请求建立 `model_request` 子 Span，所属模型 Span 同时记录最近一次尝试的字段。VeADK 发起的重试和 fallback 分别记录尝试；LiteLLM/Provider 内部重试计入所观测的 SDK 请求（`veadk.model.request_scope=sdk_request`）。不支持的外部/自定义引擎明确标记，不用 invocation Span 起点替代模型开始。未使用 VeADK HTTP 工厂的应用需要安装请求计时中间件或使用 VeADK Runner，才能建立请求时间戳。客户端渲染计时和 TTFT/TPOT 直方图上报属于独立能力。 diff --git a/prd-spec/features/first-token-timestamps/2026-10-28-first-token-timestamps.md b/prd-spec/features/first-token-timestamps/2026-10-28-first-token-timestamps.md new file mode 100644 index 000000000..a22609b66 --- /dev/null +++ b/prd-spec/features/first-token-timestamps/2026-10-28-first-token-timestamps.md @@ -0,0 +1,114 @@ +# First-token timestamps + +[中文版](2026-10-28-first-token-timestamps.zh.md) + +Change ID: `first-token-timestamps`; created/revised: 2026-10-28; status: approved (implementation and feature verification complete; repository default regression is not fully green). +Component: [model-timing](../../../specs/model-timing/README.md). Base: `23014771`. + +## Background, scope, and approval + +A360 needs timestamps to analyze model and Agent waiting time. Existing `portal_metrics.py` defines TTFT/TPOT instruments but leaves recording as TODO; response tracing alone does not preserve first-output receipt. The user explicitly narrowed scope to first-token timestamps, timing definitions/ranges, and non-streaming distinction, then authorized implementation on a new branch, changed-line coverage >=95%, and real cloud E2E. This approval covers isolated test deployment. The user subsequently authorized committing and pushing this feature branch. PR creation is not requested. + +Goals: additive timing attributes on request/model spans, receipt-time capture without usage dependency, documented scopes, live cloud proof. Non-goals: TTFT/TPOT histograms, client-render timing, provider inference timestamps, hidden provider retry instrumentation, external-engine model interception, frontend changes, dependency upgrades. + +## Requirements and scenarios + +- FR-1: HTTP requests and direct Runner calls establish isolated request timing before preprocessing (CON-1/5). +- FR-2: SDK requests capture model start and first generated delta for text/reasoning/tool-call streams; empty/metadata/final aggregate events are ignored (CON-2/3/6). +- FR-3: Non-streaming and unsupported/no-output paths expose explicit status with absent first-token timestamp; retries and cancellation retain accurate attempt evidence (CON-4/5). +- FR-4: Existing cloud exporters and JSON dumps retain the timing attributes without exposing content or changing responses (CON-6). + +Given a stream beginning with role metadata, when text arrives, then its receipt timestamp is latched once. Given reasoning or tool arguments precede text, the first generated delta is tagged by type. Given no usage, timing remains present. Given a non-streaming request, completion never becomes a first token. Given two parallel requests or resumed generator iterations, timestamps remain associated with the correct invocation. Given a failed pre-output attempt and retry, child attempt spans are distinct. Given generator cancellation, attempt spans end and the original cancellation propagates. + +## Design and impact + +Use a small shared timing module with request ContextVar, ASGI middleware, yield-safe Runner propagation, content-delta classification, and a per-SDK-request span recorder. A timed LiteLLM client subclass intercepts the actual SDK call before ADK aggregation; Ark capture is at its raw event loop. This avoids estimating first output from final telemetry and avoids globally patching third-party clients. Supported HTTP factories install the middleware once. Existing annotation processing copies request timing to invocation/agent spans. Final telemetry marks unsupported paths without inventing model start. + +Affected code: timing module, LiteLLM client/RetryingLiteLlm, ArkLlm, Runner, telemetry, in-memory annotation processor, AgentKit/Harness factories. Generated AgentKit projects already use the shared factory. No auth or database contracts change. Existing tools, content redaction, retry/fallback behavior, and exports remain intact. Rollback removes additive fields/code; no migration is required. + +## Tasks and acceptance + +| Requirement | Task | Acceptance | Test/command | Result | +| --- | --- | --- | --- | --- | +| FR-1 | T-1: request boundary/context | AC-1: correct start/source; no context leakage | `uv run --extra dev pytest tests/test_model_timing.py -q` | pass: request/context contracts | +| FR-2/3 | T-2: adapter instrumentation | AC-2: first delta latched; non-streaming absent; attempts terminate | timing and adapter regression tests | pass: 55 timing/adapter cases | +| FR-4 | T-3: span/export wiring | AC-3: attributes survive export and dump | tracing/integration regression tests | pass: actual cloud exporter/dump | +| FR-1/2/3/4 | T-4: verification | AC-4: changed executable line coverage >=95%; required checks pass | coverage JSON against `23014771`; `uv run --extra dev pre-commit run --all-files`; `uv run --extra dev pytest -n 2 -m "not codex_smoke and not piagent_smoke"` | pass: 100% coverage, static/docs checks; fail: one existing default-regression case | +| FR-1/2/3/4 | T-5: isolated live cloud E2E | AC-5: deployed branch code, real streaming/non-streaming model calls, exported timestamps verified | isolated AgentKit Runtime; protected environment credentials; HTTP calls and exported span assertions | pass: 6 real cloud cases, source hashes verified, Runtime deleted | + +## Review, risks, and delivery + +Direct design review completed because `review-spec` is unavailable. PRD/spec responsibilities, bilingual identifiers, no-usage/empty/non-streaming semantics, generator context and resource cleanup, request source, privacy, compatibility, testability, and deployment authorization have been reconciled; no design blocker remains. User authorization is the implementation request following the agreed scope. Implementation review verified raw SDK interception, yield-safe request context, stream cleanup, compatible client injection, no-content timing, and factory/export consumers. The feature-specific acceptance evidence is complete. AC-4's repository-wide default-regression condition remains fail for one existing Sidecar test reproduced on the comparison base; this PRD stays approved rather than claiming all delivery gates passed. SDK boundaries do not expose internal provider retry start. HTTP timing starts at ASGI entry and is not browser-visible wait. + +## Verification conclusions (2026-10-28) + +Comparison base: `2301477104f6f77e76c8182529545950e2d8ae6e`; branch: `feat/first-token-timestamps`. The inspected diff includes every modified tracked file and new untracked production/test/design file. The user's original dirty checkout is preserved in a separate worktree. Python 3.12.13, google-adk 2.7.1, LiteLLM 1.83.14, and the repository environment with all extras were used; test commands retained that environment with `--no-sync`. No dependency metadata changed. + +Old behavior did not preserve first generated delta receipt. CON-1..6 now define additive UTC Unix-nanosecond attributes, source/mode/type/status, per-SDK-request attempt spans, and unchanged response/error propagation. HTTP factories, Runner, raw adapter loops, final telemetry, existing exporters/dumps, and bilingual user documentation are the affected consumers. Generated AgentKit projects already call the shared factory. No frontend UI, schema migration, authentication, or metric histogram contract changes. + +### Gate status + +| Gate | Status | Evidence or blocking reason | +| --- | --- | --- | +| Change Basis | PASS | Complete tracked/untracked diff inspected against `23014771`; original work preserved. | +| Contract | PASS | FR-1..4 and CON-1..6 agree with code and both document languages. | +| Impact | PASS | R-1..7 map request, model, lifecycle, export, factories, compatibility, and docs consumers. | +| Portfolio | PASS | Controlled clocks, raw SDK boundaries, failure injection, ASGI contracts, cloud E2E, and existing regression. | +| Baseline | PASS | Missing timing module was RED; close-before-iteration was RED then GREEN. Existing Harness HTTP test preconditions were corrected without changing assertions; two collection-context cases are GREEN. | +| Execution | PASS | Changed behavior checks executed; exact final counts below. One unrelated default-regression failure is reproduced on the comparison base and remains `fail`. | +| Evaluation | PASS | 208/208 changed executable production lines; 53/54 changed branch arcs; all material feature risks verified; the remaining Ark optional-close guard branch is explained below. | +| Handoff | PASS | Bilingual contracts/docs reconciled, live evidence saved without content/credentials, isolated cloud Runtime deleted; commit/push subsequently authorized. | + +### Change Impact + +| Change ID | Changed contract | Affected surface | Repository evidence | Failure mode | Risk | +| --- | --- | --- | --- | --- | --- | +| FR-1 | CON-1/5 request source and propagation | Runner, ASGI request/invocation spans | `timing.py`, `runner.py`, application factories | Request timing leaks across yields/concurrent requests or begins after preprocessing | R-1 High | +| FR-2 | CON-2/3 first delta | LiteLLM and Ark raw SDK responses | `timed_lite_llm.py`, `ark_llm.py` | Metadata/aggregates counted; reasoning/tool output missed; usage dependency | R-2 High | +| FR-3 | CON-4 mode/status | Model spans and unsupported engines | `ModelTiming`, `mark_uninstrumented_model` | Non-streaming completion or invocation start falsely becomes first-token/model start | R-3 High | +| FR-3 | CON-2/5 attempts and termination | Retry, error, cancellation, stream closure | `RetryingLiteLlm`, `_TimedStream`, Ark response loop | Leaked spans/resources, replayed output, lost first-output evidence | R-4 High | +| FR-4 | CON-6 export | OTel processors/exporters and trace dump | `inmemory_exporter.py`, `telemetry.py`, live `/e2e/evidence` | Timing disappears on export or sensitive content is added | R-5 Medium | +| FR-1/4 | Additive compatible wiring | AgentKit/Harness factories, generated apps, existing callers | Existing AgentKit/Harness/Runner/model test contracts and generator factory call sites | Request events, route/session behavior, or public client injection regresses | R-6 High | +| FR-1..4 | Maintained definitions | PRD/spec and tracing pages in both languages | Paired `.md`/`.zh.md`, `.mdx`/`.en.mdx` | Scope/identifiers disagree or MDX breaks the docs build | R-7 Medium | + +High ratings reflect common public exposure plus timing/lifecycle hazards; Medium ratings reflect cross-module export/document boundaries. No new authorization or persistence risk is introduced. Coverage reduces execution uncertainty; it does not reduce failure impact severity. + +### Test Portfolio + +| Risk ID | Boundary | Method | Cases | Oracle | Evidence | Status | +| --- | --- | --- | --- | --- | --- | --- | +| R-1 | ASGI/async generators | Clock control, concurrency, context/closure contracts | HTTP vs Runner, inherited start, parallel yields, errors/close, non-HTTP scope | Correct source/start; caller context restored; request isolation | `tests/test_model_timing.py`; 2 concurrent live cloud calls | VERIFIED | +| R-2 | Actual owned adapters; provider boundary controlled | Equivalence partitions and adapter integration | Text/reasoning/tool deltas; empty/metadata/aggregate; missing usage | Timestamp latches once on first generated delta before ADK aggregation | Both new test modules; 4 real cloud adapter/mode cases | VERIFIED | +| R-3 | Timing state/telemetry | Mode/status contracts | Non-streaming, no output, unsupported, no recording tracer | Explicit status and absence of fabricated model/first-token time | Unit/tracing tests and both real cloud non-streaming cases | VERIFIED | +| R-4 | SDK request/iterator lifecycle | Failure injection and real VeADK retry integration | Pre-output 429 retry, error/cancel, close before/after iteration, close error | Distinct child spans, preserved output/errors, original stream closes once, terminal outcome | `tests/models/test_model_timing_adapters.py` and existing retry/fallback tests | VERIFIED | +| R-5 | Actual exporter/dump | Integration and privacy inspection | Cloud-exported request/model fields; source fingerprint | Expected attributes/parent/order survive actual dump; timing adds no content | 6 sanitized live evidence files, credential-content check, pre-commit scans | VERIFIED | +| R-6 | Factories and existing SDK callers | Contract/affected/default regression | Routes, session creation/config, Runner events, adapters/retries/fallbacks | Existing assertions retained; additive timing middleware and fields | 244 affected tests; final default-regression and baseline conclusion below | VERIFIED | +| R-7 | Docs build and paired files | Type/build, identifiers and relative links | Chinese/English definitions and valid MDX | Equivalent fields/IDs/scopes; site compiles and type-checks | Docs commands, paired-language/link check, `git diff --check` | VERIFIED | + +### Executed checks + +- `uv run --no-sync --extra dev pytest tests/models/test_model_timing_adapters.py tests/test_model_timing.py -q`: **55 passed**, zero failures/skips. Provider calls are controlled at the SDK boundary; these are local adapter tests. +- `COVERAGE_FILE=/tmp/veadk-first-token-delivery.coverage uv run --no-sync --extra dev pytest tests/test_tracing_content.py tests/test_tracing.py tests/test_inmemory_exporter.py tests/test_runner.py tests/runner/test_runner_contract.py tests/integrations/agentkit/test_app.py tests/cloud/test_harness_app_contract.py tests/cloud/test_harness_app_http.py tests/models tests/test_model_timing.py --cov=veadk --cov-branch --cov-report=json:/tmp/veadk-first-token-coverage-delivery.json --cov-report= -q`: **244 selected/executed/passed**, zero failures/skips. +- `uv run --no-sync --extra dev pytest -n 2 -m "not codex_smoke and not piagent_smoke" -q`: **7,987 selected; 7,945 executed (including 4 expected failures); 7,940 passed, 1 failed, 42 skipped, 4 xfailed; 6 subtests passed; 472.86s; status: fail (existing comparison-base failure)**. Two workers match repository policy. The Codex/Pi smoke markers are excluded by the repository default; external-engine model instrumentation is outside this feature's scope. Skipped/expected-failure cases provide no live evidence. +- Comparison-base execution with full test collection, `-n 2 -k test_run_wraps_child_with_sidecar_environment -q`: **1 failed**. The remaining default-regression failure is the same existing Sidecar child-model-URL assertion (fixed port versus the relay URL). It is not changed or suppressed by this feature. +- `uv run --no-sync --extra dev pre-commit run --all-files` and explicit `--files` for newly added Python/design/spec files: **pass** for Ruff, formatting and applicable secret/dependency checks. Hooks with no matching file are reported skipped, not executed. +- `npm --prefix docs run types:check` and `npm --prefix docs run build`: **pass**. Docs dependencies came from the existing pnpm lock; no package/lock changes. +- Paired-language identifiers, relative links, full production-file SHA256 versus the cloud build, credential-content inspection, and `git diff --check`: **pass**. + +Changed-line coverage is computed from the entire Git diff plus all lines in new untracked production Python modules, intersected with coverage.py executable lines. Tests and Markdown are not production executable lines. No coverage exclusions were added. Result: **100% (208/208)**. Both new production modules have 100% line and branch coverage. The overall changed branch arcs are **53/54**; the unexecuted arc is Ark's optional missing-`close` guard, while the verified real Ark SDK stream implements async `close`. Normal/early-close/error paths use that actual interface. + +The timestamp, status/absence, one-event latch, request isolation, parent relationships, original output and terminal cleanup assertions discriminate plausible defects; aggregate coverage alone is not the oracle. + +### Real cloud E2E + +Region: `cn-beijing`; isolated Runtime: `r-yey87ec8hs2hv32antzi`; actual model: `doubao-seed-2-1-pro-260628`; result: **6/6 pass**. The branch wheel was deployed to the cloud Runtime with all nine changed production files matching SHA256. Calls used the real LiteLLM Chat and Ark Responses adapters and real model service, through the actual VeADK app factory/middleware and Runner; evidence came from the built-in trace exporter/dump. This is distinct from locally controlled SDK-boundary tests. + +Chat streaming/non-streaming, Responses streaming/non-streaming, and concurrent Chat/Responses streams returned nonempty real output and completed. Streaming spans had `http` source, correct `call_llm` parent, `sdk_request` scope, `observed` status/type and request-start <= model-start <= first-token <= span-end. Non-streaming spans had actual SDK `streaming=false`, `non_streaming` status, and no first-token time/type. Concurrent requests retained distinct request starts/session evidence. Exported parent fields agreed with child attempts. The isolated Runtime was deleted after evidence capture. + +Sanitized evidence is retained outside Git in `veadk-first-token-20261028`; conclusions: `cloud-verification.json`; individual cases: `chat-stream.json`, `chat-nonstream.json`, `responses-stream.json`, `responses-nonstream.json`, `concurrent-chat.json`, `concurrent-responses.json`; diff coverage: `changed-coverage.json`. Only verification conclusions are recorded here; no credential or troubleshooting history is included. + +### Delivery limits + +Request start is server ASGI ingress or direct Runner execution, not browser submission/render. Model start is adapter SDK request, not provider inference. Opaque LiteLLM/provider retries remain within one SDK request; external/custom engines have explicit unavailable status. No new TTFT/TPOT histogram or cross-host duration guarantee is added. Cloud proof covers one Beijing Runtime and the two supported adapters/model modes; no broad provider matrix or load benchmark is claimed. The existing Sidecar regression above remains outside this change. Commit and push are subsequently authorized; no PR is requested. Git history and the remote feature branch record publication. + +Pre-publication verification (2026-10-28): fetched and rebased against `origin/main` (`23014771`, unchanged); all applicable pre-commit hooks passed and all 244 affected tests passed again. Production SHA256 values still match the real cloud E2E build. diff --git a/prd-spec/features/first-token-timestamps/2026-10-28-first-token-timestamps.zh.md b/prd-spec/features/first-token-timestamps/2026-10-28-first-token-timestamps.zh.md new file mode 100644 index 000000000..e47c32f71 --- /dev/null +++ b/prd-spec/features/first-token-timestamps/2026-10-28-first-token-timestamps.zh.md @@ -0,0 +1,114 @@ +# 首 Token 时间戳 + +[English](2026-10-28-first-token-timestamps.md) + +变更 ID：`first-token-timestamps`；创建/修订：2026-10-28；状态：approved（实现与需求验证完成；仓库默认回归未全绿）。 +组件：[model-timing](../../../specs/model-timing/README.zh.md)。基线：`23014771`。 + +## 背景、范围和授权 + +A360 需要时间戳分析模型和 Agent 等待时间。现有 `portal_metrics.py` 定义了 TTFT/TPOT 仪表，但实际记录仍为 TODO；单靠响应 Trace 无法保留首输出接收时间。用户明确将范围收敛为首 Token 时间戳、计时定义/范围、区分非流式，并授权新建分支实现、变更行覆盖率 >=95% 和真实云端 E2E。授权覆盖隔离测试部署。用户随后授权提交并推送本功能分支，未要求创建 PR。 + +目标：在请求/模型 Span 上新增计时属性，不依赖 usage 采集接收时间，记录范围定义，提供真实云端证据。非目标：TTFT/TPOT 直方图、客户端渲染计时、服务端推理时间、Provider 隐藏重试采集、外部执行引擎模型拦截、前端修改、依赖升级。 + +## 需求和场景 + +- FR-1：HTTP 请求和直接 Runner 调用在预处理前建立隔离请求计时（CON-1/5）。 +- FR-2：SDK 请求记录模型开始及正文/推理/工具调用流的首个生成增量；忽略空包、元数据和最终聚合事件（CON-2/3/6）。 +- FR-3：非流式、不支持和无输出路径明确标记状态且不提供首 Token 时间；重试和取消保留准确尝试证据（CON-4/5）。 +- FR-4：现有云端导出器和 JSON dump 保留计时属性，不泄露内容，不改变响应（CON-6）。 + +流以角色元数据开始时，正文到达后只锁存一次接收时间。推理或工具参数先于正文时，首个生成增量按类型标记。没有 usage 时仍有计时。非流式响应完成不能作为首 Token。两个并发请求或恢复生成器迭代时，时间戳仍属于正确调用。首输出前失败并重试时，各尝试子 Span 独立。取消生成器时，尝试 Span 结束且原取消异常继续传播。 + +## 设计与影响 + +使用小型公共计时模块，包含请求 ContextVar、ASGI 中间件、不会跨 yield 泄漏的 Runner 传播、生成增量识别和每次 SDK 请求的 Span 记录器。计时 LiteLLM 客户端子类在 ADK 聚合之前拦截实际 SDK 调用；Ark 在原始事件循环采集。避免从最终遥测估计首输出，也避免全局修改第三方客户端。支持的 HTTP 工厂仅安装一次中间件。现有标注处理器将请求计时复制到 invocation/agent Span。最终遥测为不支持的路径标记状态，不伪造模型开始。 + +影响代码：计时模块、LiteLLM 客户端/RetryingLiteLlm、ArkLlm、Runner、telemetry、内存标注处理器、AgentKit/Harness 工厂。生成的 AgentKit 项目已使用共享工厂。鉴权和数据库契约不变。现有工具、内容脱敏、重试/fallback 和导出行为保持一致。回滚移除新增字段与代码，无需迁移。 + +## 任务与验收 + +| 需求 | 任务 | 验收 | 测试/命令 | 结果 | +| --- | --- | --- | --- | --- | +| FR-1 | T-1：请求入口/上下文 | AC-1：起点和来源准确；上下文不泄漏 | `uv run --extra dev pytest tests/test_model_timing.py -q` | pass：请求/上下文契约 | +| FR-2/3 | T-2：适配器采集 | AC-2：锁存首增量；非流式缺省；尝试结束 | 计时与适配器回归测试 | pass：55 项计时/适配器用例 | +| FR-4 | T-3：Span/导出接入 | AC-3：导出和 dump 保留属性 | tracing/integration 回归测试 | pass：真实云端 exporter/dump | +| FR-1/2/3/4 | T-4：验证 | AC-4：变更可执行行覆盖率 >=95%；必需检查通过 | 相对 `23014771` 的 coverage JSON；`uv run --extra dev pre-commit run --all-files`；`uv run --extra dev pytest -n 2 -m "not codex_smoke and not piagent_smoke"` | pass：100% 覆盖率、静态/文档检查；fail：一项既有默认回归 | +| FR-1/2/3/4 | T-5：隔离真实云端 E2E | AC-5：部署分支代码、真实流式/非流式模型调用、核对导出时间戳 | 隔离 AgentKit Runtime；受保护的环境凭证；HTTP 调用与导出 Span 断言 | pass：6 项真实云端用例、源码指纹一致、Runtime 已删除 | + +## 评审、风险和交付 + +因 `review-spec` 不可用，已直接完成设计评审。PRD/spec 职责、双语标识、无 usage/空包/非流式语义、生成器上下文和资源清理、请求来源、隐私、兼容性、可测试性及部署授权已对齐，无设计阻塞。用户授权来自确认范围后的实现请求。实现评审核对了原始 SDK 拦截、yield 安全的请求上下文、流清理、兼容的 client 注入、不采集内容的计时，以及工厂/导出消费者。需求自身验收证据完整。AC-4 的仓库默认回归条件仍为 fail：一项既有 Sidecar 测试已在对比基线复现，因此 PRD 保持 approved，不宣称全部交付门禁通过。SDK 边界无法暴露 Provider 内部重试开始。HTTP 计时起于 ASGI 入口，不代表浏览器可见等待。 + +## 验证结论（2026-10-28） + +对比基线：`2301477104f6f77e76c8182529545950e2d8ae6e`；分支：`feat/first-token-timestamps`。检查范围包含所有已修改的受版本管理文件，以及尚未跟踪的新增生产代码、测试和设计文件。独立 worktree 保留了用户原工作区的未提交改动。验证环境为 Python 3.12.13、google-adk 2.7.1、LiteLLM 1.83.14，安装仓库全部 extras；测试命令通过 `--no-sync` 保留该环境。依赖元数据未修改。 + +旧行为不保留首个生成增量的接收时间。CON-1..6 现在定义增量添加的 UTC Unix 纳秒属性、来源/模式/类型/状态、每次 SDK 请求的尝试 Span，以及保持原有响应和异常传播。受影响的消费者包括 HTTP 工厂、Runner、适配器原始事件循环、最终遥测、现有 exporter/dump 和双语用户文档。生成的 AgentKit 项目已调用共享工厂。前端 UI、数据迁移、认证和指标直方图契约均无变更。 + +### Gate 状态 + +| Gate | Status | 证据或阻断原因 | +| --- | --- | --- | +| Change Basis | PASS | 对 `23014771` 检查完整的已跟踪/未跟踪 diff；保留原工作区。 | +| Contract | PASS | FR-1..4、CON-1..6 与代码和两种语言的文档一致。 | +| Impact | PASS | R-1..7 映射请求、模型、生命周期、导出、工厂、兼容性和文档消费者。 | +| Portfolio | PASS | 受控时钟、原始 SDK 边界、故障注入、ASGI 契约、云端 E2E 和既有回归。 | +| Baseline | PASS | 计时模块缺失时为 RED；迭代前关闭用例 RED 后转为 GREEN。既有 Harness HTTP 测试补齐前置条件，断言未变；完整收集上下文中的两项用例为 GREEN。 | +| Execution | PASS | 已执行变更行为验证，最终精确数量见下文。一项无关的默认回归失败在对比基线上复现，其检查状态仍为 `fail`。 | +| Evaluation | PASS | 变更可执行生产行 208/208，变更分支弧 53/54；所有实质性需求风险均已验证，下文说明 Ark 可选关闭保护分支。 | +| Handoff | PASS | 双语契约/文档对齐，保存不含内容/凭证的真实证据，删除独立云端 Runtime；用户随后授权提交/推送。 | + +### Change Impact + +| Change ID | Changed contract | Affected surface | Repository evidence | Failure mode | Risk | +| --- | --- | --- | --- | --- | --- | +| FR-1 | CON-1/5 请求来源与传播 | Runner、ASGI 请求/invocation Span | `timing.py`、`runner.py`、应用工厂 | 时间在 yield/并发请求间串用，或晚于预处理开始 | R-1 High | +| FR-2 | CON-2/3 首个增量 | LiteLLM 与 Ark 原始 SDK 响应 | `timed_lite_llm.py`、`ark_llm.py` | 把元数据/聚合响应算成首 Token，漏掉推理/工具增量，依赖 usage | R-2 High | +| FR-3 | CON-4 模式/状态 | 模型 Span 与不支持的引擎 | `ModelTiming`、`mark_uninstrumented_model` | 把非流式完成或 invocation 起点伪装成首 Token/模型起点 | R-3 High | +| FR-3 | CON-2/5 尝试与终止 | 重试、异常、取消、流关闭 | `RetryingLiteLlm`、`_TimedStream`、Ark 响应循环 | Span/资源泄漏、输出重放、首输出证据丢失 | R-4 High | +| FR-4 | CON-6 导出 | OTel processor/exporter 与 Trace dump | `inmemory_exporter.py`、`telemetry.py`、真实 `/e2e/evidence` | 导出丢失时间属性，或添加敏感内容 | R-5 Medium | +| FR-1/4 | 兼容的增量接入 | AgentKit/Harness 工厂、生成应用、既有调用方 | 既有 AgentKit/Harness/Runner/模型测试契约和生成器工厂调用点 | 请求事件、路由/会话行为或公共 client 注入回归 | R-6 High | +| FR-1..4 | 持续维护的定义 | 两种语言的 PRD/spec 和 Tracing 页面 | 成对 `.md`/`.zh.md`、`.mdx`/`.en.mdx` | 范围/标识符不一致，或 MDX 破坏文档构建 | R-7 Medium | + +High 对应常用公共路径与计时/生命周期变更风险；Medium 对应跨模块的导出/文档边界。未引入新的授权或持久化风险。覆盖率降低执行范围的不确定性，不降低故障影响严重度。 + +### Test Portfolio + +| Risk ID | Boundary | Method | Cases | Oracle | Evidence | Status | +| --- | --- | --- | --- | --- | --- | --- | +| R-1 | ASGI/异步生成器 | 时钟控制、并发、上下文/关闭契约 | HTTP 与 Runner、继承起点、并行 yield、异常/关闭、非 HTTP scope | 来源/起点正确，恢复调用方上下文，请求相互隔离 | `tests/test_model_timing.py`；2 次并发真实云端调用 | VERIFIED | +| R-2 | 真实自有适配器；受控 Provider 边界 | 等价类与适配器集成 | 文本/推理/工具增量，空值/元数据/聚合，缺少 usage | ADK 聚合前首个生成增量锁定时间，且只记录一次 | 两个新增测试模块；4 个真实云端适配器/模式用例 | VERIFIED | +| R-3 | 计时状态/遥测 | 模式/状态契约 | 非流式、无输出、不支持（`unsupported`）、无 recording tracer | 状态明确，不伪造模型/首 Token 时间 | 单元/Tracing 测试及两个真实云端非流式用例 | VERIFIED | +| R-4 | SDK 请求/迭代器生命周期 | 故障注入与真实 VeADK 重试集成 | 输出前 429 重试，异常/取消，迭代前后关闭，关闭异常 | 独立子 Span，保留输出/异常，原始流只关闭一次，终态正确 | `tests/models/test_model_timing_adapters.py` 及既有重试/fallback 测试 | VERIFIED | +| R-5 | 实际 exporter/dump | 集成与隐私检查 | 云端导出的请求/模型字段，源码指纹 | 真实 dump 保留属性/父子关系/时间顺序；计时不添加内容 | 6 份脱敏真实证据，凭证内容检查，pre-commit 扫描 | VERIFIED | +| R-6 | 工厂和既有 SDK 调用方 | 契约/受影响/默认回归 | 路由、会话创建/配置、Runner 事件、适配器/重试/fallback | 保留既有断言；计时中间件和字段为增量变更 | 244 项受影响测试；下文最终默认回归和基线结论 | VERIFIED | +| R-7 | 文档构建和成对文件 | 类型/构建、标识符和相对链接 | 中英文定义与有效 MDX | 字段/ID/范围等价；文档编译及类型检查通过 | 文档命令、双语/链接检查、`git diff --check` | VERIFIED | + +### 已执行检查 + +- `uv run --no-sync --extra dev pytest tests/models/test_model_timing_adapters.py tests/test_model_timing.py -q`：**55 项通过**，无失败/跳过。Provider 调用在 SDK 边界受控，这是本地适配器测试。 +- `COVERAGE_FILE=/tmp/veadk-first-token-delivery.coverage uv run --no-sync --extra dev pytest tests/test_tracing_content.py tests/test_tracing.py tests/test_inmemory_exporter.py tests/test_runner.py tests/runner/test_runner_contract.py tests/integrations/agentkit/test_app.py tests/cloud/test_harness_app_contract.py tests/cloud/test_harness_app_http.py tests/models tests/test_model_timing.py --cov=veadk --cov-branch --cov-report=json:/tmp/veadk-first-token-coverage-delivery.json --cov-report= -q`：**244 项选中/执行/通过**，无失败/跳过。 +- `uv run --no-sync --extra dev pytest -n 2 -m "not codex_smoke and not piagent_smoke" -q`：**选中 7,987 项；执行 7,945 项（含 4 项预期失败）；7,940 项通过、1 项失败、42 项跳过、4 项 xfailed；另有 6 个子测试通过；472.86s；状态：fail（对比基线的既有失败）**。两个 worker 符合仓库要求。仓库默认回归排除 Codex/Pi smoke 标记；外部引擎模型拦截不在本需求范围。跳过和预期失败的用例不作为真实验证证据。 +- 对比基线以完整测试收集上下文执行 `-n 2 -k test_run_wraps_child_with_sidecar_environment -q`：**1 项失败**。剩余默认回归失败为相同的既有 Sidecar 子进程模型 URL 断言（固定端口与 relay URL），本需求未修改或屏蔽该用例。 +- `uv run --no-sync --extra dev pre-commit run --all-files` 及对新增 Python/设计/spec 文件显式执行 `--files`：Ruff、格式及适用的凭证/依赖检查均 **pass**。无匹配文件的 hook 标记为跳过，不算执行通过。 +- `npm --prefix docs run types:check` 和 `npm --prefix docs run build`：均 **pass**。文档依赖来自既有 pnpm lock，package/lock 未修改。 +- 双语标识符、相对链接、全部生产文件 SHA256 与云端构建一致性、凭证内容检查以及 `git diff --check`：均 **pass**。 + +变更行覆盖率按完整 Git diff 加新增未跟踪生产 Python 模块的全部行，再与 coverage.py 可执行行求交集计算。测试与 Markdown 不属于生产可执行行。未添加任何覆盖率排除项。结果：**100%（208/208）**。两个新增生产模块的行和分支覆盖率均为 100%。整体变更分支弧为 **53/54**；未执行分支是 Ark 可选的缺少 `close` 保护分支，而经验证的真实 Ark SDK Stream 具备异步 `close`。正常/提前关闭/异常路径使用该真实接口。 + +时间值、状态/字段缺失、事件只锁定一次、请求隔离、父子关系、原始输出和终态资源关闭断言，能够区分相应缺陷；累计覆盖率本身不是测试判据。 + +### 真实云端 E2E + +地域：`cn-beijing`；独立 Runtime：`r-yey87ec8hs2hv32antzi`；真实模型：`doubao-seed-2-1-pro-260628`；结果：**6/6 通过**。云端 Runtime 部署本分支 wheel，全部 9 个生产变更文件的 SHA256 一致。通过实际 VeADK 应用工厂/中间件及 Runner，调用真实 LiteLLM Chat、Ark Responses 适配器和模型服务；证据来自内置 Trace exporter/dump。它与本地受控 SDK 边界测试分别记录。 + +Chat 流式/非流式、Responses 流式/非流式及并发 Chat/Responses 流式调用，均返回非空真实输出并完成。流式 Span 的来源为 `http`，父 Span 为 `call_llm`，范围为 `sdk_request`，状态/类型为 `observed`，且请求开始 <= 模型开始 <= 首 Token <= Span 结束。非流式 Span 的实际 SDK `streaming=false`、状态为 `non_streaming`，无首 Token 时间/类型。并发请求保留不同的请求起点与会话证据。导出的父 Span 字段与子尝试一致。采集证据后已删除独立 Runtime。 + +脱敏证据保留在 Git 之外的 `veadk-first-token-20261028` 目录；结论：`cloud-verification.json`；各场景：`chat-stream.json`、`chat-nonstream.json`、`responses-stream.json`、`responses-nonstream.json`、`concurrent-chat.json`、`concurrent-responses.json`；变更覆盖率：`changed-coverage.json`。本文仅保留验证结论，不包含凭证和排障历史。 + +### 交付边界 + +请求起点是服务端 ASGI 入口或直接 Runner 执行，不是浏览器提交/渲染。模型起点是适配器 SDK 请求，不是 Provider 推理。不可见的 LiteLLM/Provider 内部重试计入一次 SDK 请求；外部/自定义引擎明确标记不可用。未新增 TTFT/TPOT 直方图或跨主机耗时保证。云端证据覆盖北京一个 Runtime、两个支持的适配器及模型调用模式，不宣称完整 Provider 矩阵或负载基准。上述既有 Sidecar 回归在本次变更范围之外。用户随后授权 commit 和 push，未要求创建 PR。发布记录见 Git 历史和远端功能分支。 + +发布前验证（2026-10-28）：已 fetch 并相对 `origin/main`（`23014771`，未变化）执行 rebase；全部适用的 pre-commit hook 通过，244 项受影响测试再次全部通过。生产文件 SHA256 仍与真实云端 E2E 构建一致。 diff --git a/specs/model-timing/README.md b/specs/model-timing/README.md new file mode 100644 index 000000000..b3e838d33 --- /dev/null +++ b/specs/model-timing/README.md @@ -0,0 +1,30 @@ +# Model timing + +[中文版](README.zh.md) + +Component ID: `model-timing`; revision: 2026-10-28; status: active. + +## Responsibility and interfaces + +Own SDK-observed request/model/first-output timestamps in `veadk/tracing/telemetry/timing.py`, model adapters, Runner, and VeADK HTTP application factories. Consumers are existing OpenTelemetry exporters and trace dumps. No new configuration, credentials, dependencies, or response payload fields are required. + +## Contracts + +- CON-1: `veadk.request.start_time_unix_ns` is captured at HTTP ASGI entry, before request parsing, in VeADK AgentKit/Harness applications. `veadk.request.start_source` is `http`. Direct VeADK Runner calls use Runner execution entry and source `runner`. An inherited HTTP timestamp takes precedence. This excludes client/network/gateway delays before ASGI entry. +- CON-2: `veadk.model.start_time_unix_ns` is captured immediately before each adapter SDK request (`LiteLLMClient.acompletion` or Ark `aresponses`). Each request has a child `model_request` span under the owning model span; its model attribute identifies the requested model. Internal provider/LiteLLM retry/fallback activity is opaque and included in that SDK request. VeADK-issued retries/fallbacks create separate child spans. Successful/latest attempt fields also annotate the owning model span. This is not provider inference start. +- CON-3: For streaming requests, `veadk.model.first_token_time_unix_ns` is the local receipt time of the first nonempty generated content delta. `veadk.model.first_token_type` is `text`, `reasoning`, or `tool_call`. Empty deltas, role/id/signature metadata, heartbeats, usage, and completion-only aggregates do not count. A chunk can contain several tokens; no provider-side token generation timestamp is claimed. Record once, without requiring usage or storing content. +- CON-4: `veadk.model.streaming` describes the actual SDK request. `veadk.model.first_token_status` is `observed`, `non_streaming`, `no_output`, or `unsupported`. Non-streaming requests never synthesize a first-token timestamp from their complete response. Unsupported/custom/external engines explicitly expose absent model timing rather than treating an invocation span start as model start. +- CON-5: Each attempt ends on completion, error, cancellation, or generator closure. Already observed timestamps survive interruption. Model/request context does not leak between concurrent invocations or through async-generator yields. Original outputs, exceptions, cancellation, retry rules, and close behavior remain intact. +- CON-6: Wall-clock timestamps are UTC Unix nanoseconds. They may be used for trace correlation; no cross-host subtraction or monotonic-duration/TPOT guarantee is introduced. Trace attributes carry timestamps so existing JSON dumps retain them. First output also emits a timestamped `veadk.model.first_token` event. + +## Security and compatibility + +Only timing, bounded status/type labels, and existing model identifiers are added. No prompts, response content, or credentials are stored by timing instrumentation. HTTP middleware does not buffer bodies or alter SSE. Existing content-tracing controls remain applicable. Public APIs and serialized agent events stay compatible. Rollback removes additive instrumentation without migrating stored data. Generic ADK/custom application factories without the middleware can only provide a Runner timestamp when using VeADK Runner, or no request timestamp; no earlier start is fabricated. + +## Verification + +CON-1/5: ASGI boundary, async-generator/context/cancellation tests. CON-2/3/4/5: real adapter tests with controlled provider boundaries, plus real cloud Runtime/provider E2E. CON-6: exporter/dump contract assertions. Changed executable production lines must reach at least 95% coverage against the branch base, with no coverage exclusions added to satisfy the gate. + +Change design and evidence: [first-token-timestamps](../../prd-spec/features/first-token-timestamps/2026-10-28-first-token-timestamps.md). + +Contract verification: 55 focused timing/adapter tests, 244 affected tests and 6 real cloud cases passed; changed production-line coverage is 100% (208/208). The unrelated default-regression failure is documented in the linked PRD. diff --git a/specs/model-timing/README.zh.md b/specs/model-timing/README.zh.md new file mode 100644 index 000000000..eb283de3e --- /dev/null +++ b/specs/model-timing/README.zh.md @@ -0,0 +1,30 @@ +# 模型计时 + +[English](README.md) + +组件 ID：`model-timing`；修订日期：2026-10-28；状态：active。 + +## 职责与接口 + +负责 `veadk/tracing/telemetry/timing.py`、模型适配器、Runner 和 VeADK HTTP 应用工厂中的 SDK 侧请求开始、模型开始及首输出时间戳。消费者是现有 OpenTelemetry 导出器和 Trace dump。无需新增配置、凭证、依赖或响应载荷字段。 + +## 契约 + +- CON-1：VeADK AgentKit/Harness 应用在 HTTP ASGI 入口、请求解析前记录 `veadk.request.start_time_unix_ns`，`veadk.request.start_source` 为 `http`。直接调用 VeADK Runner 时以 Runner 执行入口为起点，来源为 `runner`。继承的 HTTP 时间戳优先。不包含 ASGI 入口之前的客户端、网络和网关耗时。 +- CON-2：每次适配器 SDK 请求（`LiteLLMClient.acompletion` 或 Ark `aresponses`）之前记录 `veadk.model.start_time_unix_ns`。每次请求在所属模型 Span 下建立 `model_request` 子 Span，其模型属性标识请求模型。Provider/LiteLLM 内部重试和 fallback 不可见，计入本次 SDK 请求；VeADK 发起的重试和 fallback 分别建立子 Span。成功或最近一次尝试的字段同时写入所属模型 Span。这不是模型服务端开始推理的时间。 +- CON-3：流式请求的 `veadk.model.first_token_time_unix_ns` 为首个非空生成内容增量在本地收到的时间，`veadk.model.first_token_type` 为 `text`、`reasoning` 或 `tool_call`。空增量、角色/id/签名元数据、心跳、usage 和仅完成时返回的聚合内容不计入。一个 chunk 可含多个 token，不声称取得服务端生成 token 的时间。仅记录一次，不依赖 usage，也不存储内容。 +- CON-4：`veadk.model.streaming` 表示实际 SDK 请求模式。`veadk.model.first_token_status` 为 `observed`、`non_streaming`、`no_output` 或 `unsupported`。非流式请求不使用完整响应时间伪造首 Token 时间。不支持的自定义/外部执行引擎明确标记模型计时缺失，不把 invocation Span 起点当成模型开始。 +- CON-5：每次尝试在完成、异常、取消或生成器关闭时结束。已观测到的时间戳在中断后保留。模型/请求上下文不会在并发调用之间或通过异步生成器 yield 泄漏。原有输出、异常、取消、重试规则和关闭行为保持一致。 +- CON-6：时钟时间戳统一为 UTC Unix 纳秒，用于 Trace 关联；不引入跨主机相减、单调时钟耗时或 TPOT 保证。时间戳写入 Trace 属性，现有 JSON dump 可以保留；首输出另产生带时间戳的 `veadk.model.first_token` 事件。 + +## 安全与兼容性 + +只添加时间、有限状态/类型标签和现有模型标识。计时采集不存储提示词、输出内容和凭证。HTTP 中间件不缓存请求体，也不改变 SSE。现有内容采集控制仍适用。公开接口及 Agent 事件序列化保持兼容。回滚只需移除新增采集，无需迁移历史数据。未安装中间件的通用 ADK/自定义应用，仅在使用 VeADK Runner 时提供 Runner 时间，否则不提供请求时间；不伪造更早的起点。 + +## 验证 + +CON-1/5：ASGI 入口、异步生成器/上下文/取消测试。CON-2/3/4/5：控制外部 Provider 边界的真实适配器测试，以及真实云端 Runtime/Provider E2E。CON-6：导出与 dump 契约断言。相对分支基线，生产代码变更可执行行覆盖率必须至少 95%，不得为了满足门槛新增覆盖率排除。 + +变更设计与证据：[first-token-timestamps](../../prd-spec/features/first-token-timestamps/2026-10-28-first-token-timestamps.zh.md)。 + +契约验证：55 项计时/适配器测试、244 项受影响测试及 6 项真实云端用例通过；生产变更行覆盖率为 100%（208/208）。无关的默认回归失败记录于所链接 PRD。 diff --git a/tests/cloud/test_harness_app_http.py b/tests/cloud/test_harness_app_http.py index 08a3ad612..f9f223e0b 100644 --- a/tests/cloud/test_harness_app_http.py +++ b/tests/cloud/test_harness_app_http.py @@ -39,6 +39,7 @@ def _closure_cell(function, name): def test_harness_app_exposes_agent_info(monkeypatch): + monkeypatch.delenv("MODEL_AGENT_NAME", raising=False) monkeypatch.setenv("MODEL_AGENT_API_KEY", "test-api-key") monkeypatch.setenv("MODEL_NAME", "test-model") monkeypatch.setenv("HARNESS_NAME", "test-harness") @@ -100,6 +101,7 @@ def test_harness_app_disables_bff_tool_host_by_default(monkeypatch): def test_harness_session_create_accepts_id_and_get_agent_config(monkeypatch): + monkeypatch.delenv("MODEL_AGENT_NAME", raising=False) monkeypatch.setenv("MODEL_AGENT_API_KEY", "test-api-key") monkeypatch.setenv("MODEL_NAME", "test-model") monkeypatch.setenv("HARNESS_NAME", "test-harness") diff --git a/tests/models/test_model_timing_adapters.py b/tests/models/test_model_timing_adapters.py new file mode 100644 index 000000000..7790035af --- /dev/null +++ b/tests/models/test_model_timing_adapters.py @@ -0,0 +1,374 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from __future__ import annotations + +import asyncio +from types import SimpleNamespace as NS + +import pytest +from google.adk.models.lite_llm import LiteLLMClient +from google.adk.models.llm_request import LlmRequest +from google.genai import types +from litellm import ModelResponse +from litellm.types.utils import ModelResponseStream +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from volcenginesdkarkruntime.types.responses import ( + ResponseTextDeltaEvent, + ResponseFunctionCallArgumentsDeltaEvent, +) + +from veadk.models.ark_llm import ArkLlm, ArkLlmClient +from veadk.models.retrying_lite_llm import RetryingLiteLlm +from veadk.models.timed_lite_llm import TimedLiteLLMClient +from veadk.tracing.telemetry import timing + + +@pytest.fixture +def spans(monkeypatch): + provider = TracerProvider() + exporter = InMemorySpanExporter() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + tracer = provider.get_tracer("adapter-test") + monkeypatch.setattr(timing.trace, "get_tracer", lambda *a, **kw: tracer) + yield tracer, exporter + provider.shutdown() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("stream", [True, False]) +async def test_default_litellm_adapter_records_before_adk_aggregation( + monkeypatch, spans, stream +): + tracer, exporter = spans + + async def chunks(): + yield ModelResponseStream( + model="openai/test", + choices=[ + {"index": 0, "delta": {"role": "assistant"}, "finish_reason": None} + ], + ) + yield ModelResponseStream( + model="openai/test", + choices=[ + {"index": 0, "delta": {"content": "hello"}, "finish_reason": None} + ], + ) + yield ModelResponseStream( + model="openai/test", + choices=[{"index": 0, "delta": {"content": "!"}, "finish_reason": "stop"}], + ) + + async def complete(self, model, messages, tools, **kwargs): + assert isinstance(self, TimedLiteLLMClient) + if kwargs.get("stream"): + return chunks() + return ModelResponse( + model=model, + choices=[ + { + "message": {"role": "assistant", "content": "hello!"}, + "finish_reason": "stop", + "index": 0, + } + ], + ) + + monkeypatch.setattr(LiteLLMClient, "acompletion", complete) + model = RetryingLiteLlm(model="openai/test") + request = LlmRequest( + model=model.model, + contents=[types.Content(role="user", parts=[types.Part(text="hello")])], + ) + with tracer.start_as_current_span("call_llm") as owner: + responses = [ + r async for r in model.generate_content_async(request, stream=stream) + ] + assert responses + assert owner.attributes["veadk.model.streaming"] is stream + assert owner.attributes["veadk.model.first_token_status"] == ( + "observed" if stream else "non_streaming" + ) + if stream: + assert ( + owner.attributes["veadk.model.first_token_time_unix_ns"] + >= owner.attributes["veadk.model.start_time_unix_ns"] + ) + else: + assert "veadk.model.first_token_time_unix_ns" not in owner.attributes + assert ( + len([s for s in exporter.get_finished_spans() if s.name == "model_request"]) + == 1 + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "mode", ["error", "cancel", "empty", "close", "no_close_method"] +) +async def test_litellm_terminal_paths_close_original_stream(monkeypatch, spans, mode): + tracer, exporter = spans + closed = [] + + async def chunks(): + try: + if mode == "error": + raise RuntimeError("provider failed") + if mode == "cancel": + raise asyncio.CancelledError() + if mode == "empty": + return + yield NS(choices=[NS(delta=NS(content="hello"))]) + finally: + closed.append(True) + + class IteratorOnly: + def __aiter__(self): + return chunks() + + async def complete(*args, **kwargs): + return IteratorOnly() if mode == "no_close_method" else chunks() + + monkeypatch.setattr(LiteLLMClient, "acompletion", complete) + with tracer.start_as_current_span("call_llm"): + response = await TimedLiteLLMClient().acompletion("test", [], [], stream=True) + if mode in {"error", "cancel"}: + with pytest.raises( + RuntimeError if mode == "error" else asyncio.CancelledError + ): + await anext(response) + elif mode == "close": + await anext(response) + await response.aclose() + else: + _ = [r async for r in response] + attempt = exporter.get_finished_spans()[0] + assert closed == [True] + assert attempt.attributes["veadk.model.outcome"] == { + "error": "error", + "cancel": "cancelled", + "close": "cancelled", + }.get(mode, "completed") + + +@pytest.mark.asyncio +async def test_litellm_request_failure_does_not_leak_span(monkeypatch, spans): + _, exporter = spans + + async def complete(*args, **kwargs): + raise RuntimeError("request failed") + + monkeypatch.setattr(LiteLLMClient, "acompletion", complete) + with pytest.raises(RuntimeError, match="request failed"): + await TimedLiteLLMClient().acompletion("test", [], [], stream=True) + assert exporter.get_finished_spans()[0].attributes["veadk.model.outcome"] == "error" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "mode", ["text", "tool_call", "empty", "close", "error", "non_streaming"] +) +async def test_ark_raw_events_and_terminal_paths(monkeypatch, spans, mode): + tracer, exporter = spans + closed = [] + + async def chunks(): + try: + yield NS(type="response.created", delta="") + if mode == "error": + raise RuntimeError("provider failed") + if mode == "empty": + return + if mode == "tool_call": + yield ResponseFunctionCallArgumentsDeltaEvent( + type="response.function_call_arguments.delta", + item_id="item", + output_index=0, + delta="{", + ) + yield ResponseTextDeltaEvent( + type="response.output_text.delta", + item_id="item", + output_index=0, + content_index=0, + delta="hello", + ) + finally: + closed.append(True) + + class Stream: + def __init__(self): + self.events = chunks() + + def __aiter__(self): + return self.events + + async def close(self): + await self.events.aclose() + + async def responses(*args, **kwargs): + if mode == "non_streaming": + return NS() + return Stream() + + monkeypatch.setattr(ArkLlmClient, "aresponses", responses) + if mode == "non_streaming": + from google.adk.models.llm_response import LlmResponse + + monkeypatch.setattr( + "veadk.models.ark_llm.ark_response_to_generate_content_response", + lambda _: LlmResponse( + content=types.Content(role="model", parts=[types.Part(text="hello")]) + ), + ) + model = ArkLlm(model="openai/test") + with tracer.start_as_current_span("call_llm") as owner: + stream = model.generate_content_via_responses( + {"model": "openai/test", "input": []}, stream=mode != "non_streaming" + ) + if mode == "error": + with pytest.raises(RuntimeError, match="provider failed"): + _ = [r async for r in stream] + elif mode == "close": + assert (await anext(stream)).content.parts[0].text == "hello" + await stream.aclose() + else: + responses_seen = [r async for r in stream] + if mode not in {"empty", "non_streaming"}: + assert responses_seen[0].content.parts[0].text == "hello" + attempt = exporter.get_finished_spans()[0] + status = ( + "non_streaming" + if mode == "non_streaming" + else "no_output" + if mode in {"empty", "error"} + else "observed" + ) + assert owner.attributes["veadk.model.first_token_status"] == status + if status == "observed": + assert owner.attributes["veadk.model.first_token_type"] == ( + "tool_call" if mode == "tool_call" else "text" + ) + else: + assert "veadk.model.first_token_time_unix_ns" not in owner.attributes + assert attempt.end_time is not None + if mode != "non_streaming": + assert closed == [True] + + +@pytest.mark.asyncio +async def test_litellm_close_before_iteration_ends_attempt(monkeypatch, spans): + _, exporter = spans + closed = [] + + class Stream: + def __aiter__(self): + return self + + async def __anext__(self): + raise AssertionError("must not consume") + + async def aclose(self): + closed.append(True) + + async def complete(*args, **kwargs): + return Stream() + + monkeypatch.setattr(LiteLLMClient, "acompletion", complete) + response = await TimedLiteLLMClient().acompletion("test", [], [], stream=True) + await response.aclose() + await response.aclose() + assert closed == [True] + attempt = exporter.get_finished_spans()[0] + assert attempt.attributes["veadk.model.outcome"] == "cancelled" + assert attempt.attributes["veadk.model.first_token_status"] == "no_output" + with pytest.raises(StopAsyncIteration): + await anext(response) + + +@pytest.mark.asyncio +async def test_litellm_close_failure_propagates_and_ends_span(monkeypatch, spans): + _, exporter = spans + + class Stream: + def __aiter__(self): + return self + + async def __anext__(self): + raise StopAsyncIteration + + async def aclose(self): + raise RuntimeError("close failed") + + async def complete(*args, **kwargs): + return Stream() + + monkeypatch.setattr(LiteLLMClient, "acompletion", complete) + response = await TimedLiteLLMClient().acompletion("test", [], [], stream=True) + with pytest.raises(RuntimeError, match="close failed"): + await anext(response) + assert exporter.get_finished_spans()[0].attributes["veadk.model.outcome"] == "error" + + +@pytest.mark.asyncio +async def test_veadk_retry_exports_distinct_sdk_attempts(monkeypatch, spans): + tracer, exporter = spans + requests = [] + + class RateLimitError(RuntimeError): + status_code = 429 + response = NS(headers={"Retry-After": "0"}) + + async def chunks(): + yield ModelResponseStream( + model="openai/test", + choices=[ + {"index": 0, "delta": {"content": "hello"}, "finish_reason": "stop"} + ], + ) + + async def complete(self, model, messages, tools, **kwargs): + requests.append(model) + if len(requests) == 1: + raise RateLimitError("pre-output quota failure") + return chunks() + + monkeypatch.setattr(LiteLLMClient, "acompletion", complete) + model = RetryingLiteLlm(model="openai/test") + request = LlmRequest( + model=model.model, + contents=[types.Content(role="user", parts=[types.Part(text="hello")])], + ) + with timing.bind_request_start(timing.RequestStart(1, "runner")): + with tracer.start_as_current_span("call_llm") as owner: + responses = [ + r async for r in model.generate_content_async(request, stream=True) + ] + assert responses and requests == ["openai/test", "openai/test"] + attempts = [s for s in exporter.get_finished_spans() if s.name == "model_request"] + assert len(attempts) == 2 + assert attempts[0].attributes["veadk.model.outcome"] == "error" + assert attempts[0].attributes["veadk.model.first_token_status"] == "no_output" + assert "veadk.model.first_token_time_unix_ns" not in attempts[0].attributes + assert attempts[1].attributes["veadk.model.outcome"] == "completed" + assert attempts[1].attributes["veadk.model.first_token_status"] == "observed" + assert ( + owner.attributes["veadk.model.first_token_time_unix_ns"] + == attempts[1].attributes["veadk.model.first_token_time_unix_ns"] + ) + assert all(s.parent.span_id == owner.context.span_id for s in attempts) + assert all(s.attributes["veadk.request.start_time_unix_ns"] == 1 for s in attempts) diff --git a/tests/test_model_timing.py b/tests/test_model_timing.py new file mode 100644 index 000000000..5323d930d --- /dev/null +++ b/tests/test_model_timing.py @@ -0,0 +1,309 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from __future__ import annotations + +import asyncio +from types import SimpleNamespace as NS + +import pytest +from fastapi import FastAPI +from httpx import ASGITransport, AsyncClient +from opentelemetry import trace +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + +from veadk.tracing.telemetry import timing + + +@pytest.fixture +def spans(monkeypatch): + provider = TracerProvider() + exporter = InMemorySpanExporter() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + tracer = provider.get_tracer("timing-test") + monkeypatch.setattr(timing.trace, "get_tracer", lambda *a, **kw: tracer) + yield tracer, exporter + provider.shutdown() + + +@pytest.mark.parametrize("kind", ["text", "reasoning", "tool_call"]) +def test_first_output_is_latched_without_usage(monkeypatch, spans, kind): + tracer, exporter = spans + clock = iter([100, 200, 300]) + monkeypatch.setattr(timing.time, "time_ns", lambda: next(clock)) + with timing.bind_request_start(timing.RequestStart(50, "http")): + with tracer.start_as_current_span("call_llm") as owner: + with timing.ModelTiming(streaming=True, model="openai/test") as call: + call.observe(None) + call.observe(kind) + call.observe("text") + assert owner.attributes["veadk.request.start_time_unix_ns"] == 50 + assert owner.attributes["veadk.model.start_time_unix_ns"] == 100 + assert owner.attributes["veadk.model.first_token_time_unix_ns"] == 200 + assert owner.attributes["veadk.model.first_token_type"] == kind + assert owner.attributes["veadk.model.first_token_status"] == "observed" + assert len(owner.events) == 1 + assert owner.events[0].timestamp == 200 + attempt = next( + s for s in exporter.get_finished_spans() if s.name == "model_request" + ) + assert attempt.parent.span_id == owner.context.span_id + assert attempt.attributes["veadk.model.request_scope"] == "sdk_request" + assert attempt.attributes["veadk.model.outcome"] == "completed" + assert "content" not in str(dict(attempt.attributes)) + + +@pytest.mark.parametrize( + "streaming,status", [(False, "non_streaming"), (True, "no_output")] +) +def test_no_first_token_is_fabricated(spans, streaming, status): + tracer, exporter = spans + with tracer.start_as_current_span("call_llm") as owner: + with timing.ModelTiming(streaming=streaming, model="test") as call: + if not streaming: + call.observe("text") + assert owner.attributes["veadk.model.first_token_status"] == status + assert "veadk.model.first_token_time_unix_ns" not in owner.attributes + assert len(exporter.get_finished_spans()) == 2 + + +@pytest.mark.parametrize( + "error,outcome", + [ + (RuntimeError("failed"), "error"), + (asyncio.CancelledError(), "cancelled"), + (GeneratorExit(), "cancelled"), + ], +) +def test_interrupted_attempt_retains_output_and_ends(spans, error, outcome): + tracer, exporter = spans + with tracer.start_as_current_span("call_llm"): + with pytest.raises(type(error)): + with timing.ModelTiming(streaming=True, model="test") as call: + call.observe("text") + raise error + attempt = exporter.get_finished_spans()[0] + assert attempt.end_time is not None + assert attempt.attributes["veadk.model.first_token_status"] == "observed" + assert attempt.attributes["veadk.model.outcome"] == outcome + + +def test_retry_attempts_have_independent_timestamps(spans): + tracer, exporter = spans + with tracer.start_as_current_span("call_llm") as owner: + with pytest.raises(RuntimeError): + with timing.ModelTiming(streaming=True, model="primary"): + raise RuntimeError("retry") + with timing.ModelTiming(streaming=True, model="fallback") as call: + call.observe("text") + attempts = [s for s in exporter.get_finished_spans() if s.name == "model_request"] + assert len(attempts) == 2 + assert attempts[0].attributes["veadk.model.first_token_status"] == "no_output" + assert "veadk.model.first_token_time_unix_ns" not in attempts[0].attributes + assert attempts[1].attributes["veadk.model.first_token_status"] == "observed" + assert ( + owner.attributes["veadk.model.start_time_unix_ns"] + == attempts[1].attributes["veadk.model.start_time_unix_ns"] + ) + + +@pytest.mark.parametrize( + "streaming,status", [(True, "unsupported"), (False, "non_streaming")] +) +def test_unsupported_models_do_not_use_invocation_start(spans, streaming, status): + tracer, _ = spans + with tracer.start_as_current_span("call_llm") as owner: + timing.mark_uninstrumented_model(owner, streaming=streaming) + assert owner.attributes["veadk.model.first_token_status"] == status + assert "veadk.model.start_time_unix_ns" not in owner.attributes + assert owner.attributes["veadk.model.streaming"] is streaming + with timing.ModelTiming(streaming=True, model="test") as call: + call.observe("text") + timing.mark_uninstrumented_model(owner, streaming=False) + assert owner.attributes["veadk.model.first_token_status"] == "observed" + + +@pytest.mark.parametrize( + "delta,expected", + [ + ({}, None), + ({"role": "assistant"}, None), + ({"content": ""}, None), + ({"content": " "}, "text"), + ({"content": "hello"}, "text"), + ({"reasoning_content": "think"}, "reasoning"), + ({"reasoning": "think"}, "reasoning"), + ({"thinking_blocks": [{"type": "thinking", "thinking": "think"}]}, "reasoning"), + ({"thinking_blocks": [{"type": "thinking", "signature": "metadata"}]}, None), + ({"tool_calls": [{"id": "metadata"}]}, None), + ( + {"tool_calls": [{"function": {"name": "search", "arguments": ""}}]}, + "tool_call", + ), + ({"tool_calls": [{"function": {"arguments": "{"}}]}, "tool_call"), + ({"function_call": {"arguments": "{"}}, "tool_call"), + ], +) +def test_litellm_delta_classification(delta, expected): + assert timing.litellm_output_type(NS(choices=[NS(delta=delta)])) == expected + + +def test_litellm_metadata_and_object_shapes(): + assert timing.litellm_output_type(NS(choices=[])) is None + assert timing.litellm_output_type(NS(choices=[NS(delta=None)])) is None + assert ( + timing.litellm_output_type(NS(choices=[NS(delta=NS(content="text"))])) == "text" + ) + assert ( + timing.litellm_output_type( + {"choices": [{"delta": {"reasoning_content": "think"}}]} + ) + == "reasoning" + ) + + +@pytest.mark.parametrize( + "event_type,delta,expected", + [ + ("response.output_text.delta", "hello", "text"), + ("response.reasoning_summary_text.delta", "think", "reasoning"), + ("response.reasoning_text.delta", "think", "reasoning"), + ("response.function_call_arguments.delta", "{", "tool_call"), + ("response.output_text.delta", "", None), + ("response.completed", "full", None), + ("response.output_item.added", "", None), + ], +) +def test_ark_delta_classification(event_type, delta, expected): + assert timing.ark_output_type(NS(type=event_type, delta=delta)) == expected + + +@pytest.mark.asyncio +async def test_request_context_is_isolated_across_yields_and_tasks(): + async def events(expected): + assert timing.current_request_start() == expected + yield expected.time_ns + await asyncio.sleep(0) + assert timing.current_request_start() == expected + yield expected.source + + first = timing.RequestStart(1, "runner") + second = timing.RequestStart(2, "http") + a = timing.with_request_timing(events(first), first) + b = timing.with_request_timing(events(second), second) + assert await asyncio.gather(anext(a), anext(b)) == [1, 2] + assert timing.current_request_start() is None + assert await asyncio.gather(anext(a), anext(b)) == ["runner", "http"] + assert timing.current_request_start() is None + await a.aclose() + await b.aclose() + + +@pytest.mark.asyncio +async def test_request_context_restored_on_error_and_close(): + inner = timing.RequestStart(2, "runner") + closed = [] + + async def events(): + try: + yield 1 + raise ValueError("original") + finally: + closed.append(timing.current_request_start()) + + with timing.bind_request_start(timing.RequestStart(1, "http")) as outer: + wrapped = timing.with_request_timing(events(), inner) + assert await anext(wrapped) == 1 + assert timing.current_request_start() == outer + with pytest.raises(ValueError, match="original"): + await anext(wrapped) + assert timing.current_request_start() == outer + assert closed == [inner] + wrapped = timing.with_request_timing(events(), inner) + await anext(wrapped) + await wrapped.aclose() + assert closed == [inner, inner] + assert timing.current_request_start() is None + + +@pytest.mark.asyncio +async def test_timed_runner_uses_inherited_or_local_start(monkeypatch): + monkeypatch.setattr(timing.time, "time_ns", lambda: 100) + + @timing.timed_runner + async def run(value): + yield (value, timing.current_request_start()) + + assert [e async for e in run("hello")] == [ + ("hello", timing.RequestStart(100, "runner")) + ] + with timing.bind_request_start(timing.RequestStart(50, "http")): + generator = run("http") + assert await anext(generator) == ("http", timing.RequestStart(50, "http")) + await generator.aclose() + assert timing.current_request_start() is None + + +@pytest.mark.asyncio +async def test_http_boundary_is_before_body_parsing_and_install_is_idempotent( + monkeypatch, +): + monkeypatch.setattr(timing.time, "time_ns", lambda: 100) + app = FastAPI() + + @app.post("/run") + async def run(payload: dict): + start = timing.current_request_start() + return { + "start": start.time_ns, + "source": start.source, + "value": payload["value"], + } + + timing.install_request_timing(app) + timing.install_request_timing(app) + assert len(app.user_middleware) == 1 + async with AsyncClient( + transport=ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.post("/run", json={"value": "hello"}) + assert response.json() == {"start": 100, "source": "http", "value": "hello"} + assert timing.current_request_start() is None + + +@pytest.mark.asyncio +async def test_non_http_scopes_and_failures_preserve_context(): + received = [] + + async def app(scope, receive, send): + received.append(timing.current_request_start()) + if scope["type"] == "http": + raise RuntimeError("original") + + middleware = timing.RequestTimingMiddleware(app) + await middleware({"type": "lifespan"}, None, None) + assert received == [None] + with pytest.raises(RuntimeError, match="original"): + await middleware({"type": "http"}, None, None) + assert received[1].source == "http" + assert timing.current_request_start() is None + + +def test_without_recording_tracer(): + with trace.use_span(trace.INVALID_SPAN): + with timing.ModelTiming(streaming=True, model="test") as call: + call.observe("text") + timing.mark_uninstrumented_model(trace.INVALID_SPAN, streaming=True) diff --git a/veadk/cloud/harness_app/app.py b/veadk/cloud/harness_app/app.py index fdbb4d629..aa10d0d88 100644 --- a/veadk/cloud/harness_app/app.py +++ b/veadk/cloud/harness_app/app.py @@ -270,6 +270,9 @@ async def lifespan(app: FastAPI): # Base app = ADK api routes; then add /harness/invoke; mount A2A last so # it catches the well-known / RPC paths the ADK routes don't claim. self.app = self._server.get_fast_api_app(lifespan=lifespan) + from veadk.tracing.telemetry.timing import install_request_timing + + install_request_timing(self.app) setattr(self.app.state, _ADK_SERVER_STATE_KEY, self._server) _configure_studio_tool_routes( self.app, diff --git a/veadk/integrations/agentkit/app.py b/veadk/integrations/agentkit/app.py index 49a33cfa1..b72110bfa 100644 --- a/veadk/integrations/agentkit/app.py +++ b/veadk/integrations/agentkit/app.py @@ -49,6 +49,7 @@ from veadk.agent_search import search_agent_component from veadk.cli.frontend_invocation import FrontendInvocationPlugin from veadk.memory.short_term_memory import ShortTermMemory +from veadk.tracing.telemetry.timing import install_request_timing from veadk.utils.logger import get_logger if TYPE_CHECKING: @@ -1055,6 +1056,7 @@ def create_agentkit_app( agent_server_kwargs["identity_health_routes"] = ("/ping",) agent_server = AgentkitAgentServerApp(**agent_server_kwargs) fastapi_app = cast(FastAPI, agent_server.app) + install_request_timing(fastapi_app) setattr(fastapi_app.state, _SERVER_STATE_KEY, agent_server) _configure_dynamic_a2a_routes(fastapi_app, root_agent, app_plugins) _configure_studio_tool_routes( diff --git a/veadk/models/ark_llm.py b/veadk/models/ark_llm.py index 4c8588d56..130a96a3b 100644 --- a/veadk/models/ark_llm.py +++ b/veadk/models/ark_llm.py @@ -61,6 +61,7 @@ from veadk.config import settings from veadk.consts import DEFAULT_VIDEO_MODEL_API_BASE +from veadk.tracing.telemetry.timing import ModelTiming, ark_output_type from veadk.utils.adk_compat import ( get_previous_interaction_id, llm_request_has_field, @@ -855,18 +856,26 @@ async def generate_content_via_responses( responses_args = request_reorganization_by_ark( responses_args, enable_responses_cache=self.enable_responses_cache ) - if stream: - responses_args["stream"] = True - async for part in await self.llm_client.aresponses(**responses_args): - llm_response = event_to_generate_content_response( - event=part, is_partial=True, model_version=model - ) - if llm_response: - yield llm_response - else: - raw_response = await self.llm_client.aresponses(**responses_args) - llm_response = ark_response_to_generate_content_response(raw_response) - yield llm_response + with ModelTiming(streaming=stream, model=model) as timing: + if stream: + responses_args["stream"] = True + response = await self.llm_client.aresponses(**responses_args) + try: + async for part in response: + timing.observe(ark_output_type(part)) + llm_response = event_to_generate_content_response( + event=part, is_partial=True, model_version=model + ) + if llm_response: + yield llm_response + finally: + close = getattr(response, "close", None) + if close is not None: + await close() + else: + raw_response = await self.llm_client.aresponses(**responses_args) + llm_response = ark_response_to_generate_content_response(raw_response) + yield llm_response @classmethod @override diff --git a/veadk/models/retrying_lite_llm.py b/veadk/models/retrying_lite_llm.py index 2b8431ba1..109f750e2 100644 --- a/veadk/models/retrying_lite_llm.py +++ b/veadk/models/retrying_lite_llm.py @@ -22,11 +22,13 @@ from collections.abc import AsyncGenerator from typing import Any -from google.adk.models.lite_llm import LiteLlm +from google.adk.models.lite_llm import LiteLlm, LiteLLMClient from google.adk.models.llm_request import LlmRequest from google.adk.models.llm_response import LlmResponse +from pydantic import Field from typing_extensions import override +from veadk.models.timed_lite_llm import TimedLiteLLMClient from veadk.utils.logger import get_logger logger = get_logger(__name__) @@ -91,6 +93,8 @@ class RetryingLiteLlm(LiteLlm): replay is unsafe because it could duplicate text, reasoning, or tool calls. """ + llm_client: LiteLLMClient = Field(default_factory=TimedLiteLLMClient, exclude=True) + def __init__(self, *, model: str, **kwargs: Any) -> None: super().__init__(model=model, **kwargs) self._fallbacks_template = copy.deepcopy( diff --git a/veadk/models/timed_lite_llm.py b/veadk/models/timed_lite_llm.py new file mode 100644 index 000000000..fd7719aec --- /dev/null +++ b/veadk/models/timed_lite_llm.py @@ -0,0 +1,82 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Capture LiteLLM SDK receipt timestamps before response aggregation.""" + +from google.adk.models.lite_llm import LiteLLMClient + +from veadk.tracing.telemetry.timing import ModelTiming, litellm_output_type + + +class TimedLiteLLMClient(LiteLLMClient): + async def acompletion(self, model, messages, tools, **kwargs): + timing = ModelTiming(streaming=bool(kwargs.get("stream")), model=model) + try: + response = await super().acompletion(model, messages, tools, **kwargs) + except BaseException as error: + timing.__exit__(type(error), error, error.__traceback__) + raise + if not timing.streaming: + timing.__exit__(None, None, None) + return response + return _TimedStream(response, timing) + + +class _TimedStream: + """End an attempt even when a returned stream is closed before iteration.""" + + def __init__(self, response, timing): + self._response = response + self._iterator = response.__aiter__() + self._timing = timing + self._closed = False + + def __aiter__(self): + return self + + async def __anext__(self): + if self._closed: + raise StopAsyncIteration + try: + chunk = await anext(self._iterator) + except StopAsyncIteration: + await self._finish(None) + raise + except BaseException as error: + await self._finish(error) + raise + self._timing.observe(litellm_output_type(chunk)) + return chunk + + async def _finish(self, error): + if self._closed: + return + self._closed = True + try: + close = getattr(self._response, "aclose", None) + if close is not None: + await close() + except BaseException as close_error: + self._timing.__exit__( + type(close_error), close_error, close_error.__traceback__ + ) + raise + self._timing.__exit__( + type(error) if error is not None else None, + error, + error.__traceback__ if error is not None else None, + ) + + async def aclose(self): + await self._finish(GeneratorExit()) diff --git a/veadk/runner.py b/veadk/runner.py index 30761488a..1e9185a17 100644 --- a/veadk/runner.py +++ b/veadk/runner.py @@ -34,6 +34,7 @@ from veadk.memory.short_term_memory import ShortTermMemory from veadk.processors.base_run_processor import BaseRunProcessor from veadk.types import MediaMessage +from veadk.tracing.telemetry.timing import timed_runner from veadk.utils.adk_compat import ( get_event_function_calls, get_event_function_responses, @@ -193,7 +194,7 @@ def flush_thinking(metadata: str) -> None: post_run_process(self) - return wrapper + return timed_runner(wrapper) return decorator diff --git a/veadk/tracing/telemetry/exporters/inmemory_exporter.py b/veadk/tracing/telemetry/exporters/inmemory_exporter.py index b15a87751..c787cfd83 100644 --- a/veadk/tracing/telemetry/exporters/inmemory_exporter.py +++ b/veadk/tracing/telemetry/exporters/inmemory_exporter.py @@ -29,6 +29,7 @@ from veadk.tracing.telemetry.exporters.base_exporter import BaseExporter from veadk.utils.logger import get_logger +from veadk.tracing.telemetry.timing import annotate_request_start logger = get_logger(__name__) _registration_lock = Lock() @@ -196,6 +197,7 @@ def on_start(self, span, parent_context) -> None: span: The span being started parent_context: Parent OpenTelemetry context """ + annotate_request_start(span) if span.name.startswith("invocation"): span.set_attribute("gen_ai.operation.name", "chain") span.set_attribute("gen_ai.span.kind", "workflow") diff --git a/veadk/tracing/telemetry/telemetry.py b/veadk/tracing/telemetry/telemetry.py index 9f2992f2b..eabdb66bb 100644 --- a/veadk/tracing/telemetry/telemetry.py +++ b/veadk/tracing/telemetry/telemetry.py @@ -18,6 +18,7 @@ from google.adk.events import Event from google.adk.models.llm_request import LlmRequest from google.adk.models.llm_response import LlmResponse +from google.adk.agents.run_config import StreamingMode from google.adk.tools import BaseTool from opentelemetry import trace from opentelemetry.context import get_value @@ -31,6 +32,7 @@ ) from veadk.tracing.telemetry.content_tracing import should_trace_content from veadk.tracing.telemetry import portal_metrics +from veadk.tracing.telemetry.timing import mark_uninstrumented_model from veadk.utils.logger import get_logger from veadk.utils.misc import safe_json_serialize @@ -392,6 +394,14 @@ def trace_call_llm( if span is None: span = trace.get_current_span() # type: ignore + mark_uninstrumented_model( + span, + streaming=bool( + invocation_context.run_config + and invocation_context.run_config.streaming_mode != StreamingMode.NONE + ), + ) + from veadk.agent import Agent set_common_attributes_on_model_span( diff --git a/veadk/tracing/telemetry/timing.py b/veadk/tracing/telemetry/timing.py new file mode 100644 index 000000000..a69b0f5c8 --- /dev/null +++ b/veadk/tracing/telemetry/timing.py @@ -0,0 +1,222 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""SDK-observed model timestamps and isolated request start propagation.""" + +from __future__ import annotations + +import asyncio +import functools +import time +from collections.abc import AsyncGenerator, Callable, Mapping +from contextlib import contextmanager +from contextvars import ContextVar +from dataclasses import dataclass +from typing import Any + +from opentelemetry import trace +from opentelemetry.trace import Span + + +@dataclass(frozen=True) +class RequestStart: + time_ns: int + source: str + + +_REQUEST_START: ContextVar[RequestStart | None] = ContextVar( + "veadk_request_start", default=None +) + + +def current_request_start() -> RequestStart | None: + return _REQUEST_START.get() + + +@contextmanager +def bind_request_start(start: RequestStart): + token = _REQUEST_START.set(start) + try: + yield start + finally: + _REQUEST_START.reset(token) + + +async def with_request_timing( + events: AsyncGenerator, start: RequestStart +) -> AsyncGenerator: + """Bind only while advancing/closing; never retain context across yields.""" + try: + while True: + with bind_request_start(start): + try: + event = await anext(events) + except StopAsyncIteration: + return + yield event + finally: + with bind_request_start(start): + await events.aclose() + + +def timed_runner(func: Callable) -> Callable: + @functools.wraps(func) + async def wrapper(*args, **kwargs): + start = current_request_start() or RequestStart(time.time_ns(), "runner") + events = with_request_timing(func(*args, **kwargs), start) + try: + async for event in events: + yield event + finally: + await events.aclose() + + return wrapper + + +class RequestTimingMiddleware: + """Capture ASGI ingress without inspecting or buffering request bodies.""" + + def __init__(self, app): + self.app = app + + async def __call__(self, scope, receive, send): + if scope["type"] != "http": + await self.app(scope, receive, send) + return + with bind_request_start(RequestStart(time.time_ns(), "http")): + await self.app(scope, receive, send) + + +def install_request_timing(app) -> None: + if getattr(app.state, "_veadk_request_timing_installed", False): + return + app.add_middleware(RequestTimingMiddleware) + app.state._veadk_request_timing_installed = True + + +def annotate_request_start(span: Span) -> None: + start = current_request_start() + if start is not None and span.is_recording(): + span.set_attribute("veadk.request.start_time_unix_ns", start.time_ns) + span.set_attribute("veadk.request.start_source", start.source) + + +class ModelTiming: + """One SDK request; this context manager never attaches OTel context.""" + + def __init__(self, *, streaming: bool, model: str): + self.streaming = streaming + self.first_token_time: int | None = None + self.owner = trace.get_current_span() + self.span = trace.get_tracer(__name__).start_span( + "model_request", attributes={"gen_ai.request.model": str(model)} + ) + self._spans = (self.span, self.owner) + for span in self._spans: + annotate_request_start(span) + self._set("veadk.model.start_time_unix_ns", time.time_ns()) + self._set("veadk.model.streaming", streaming) + self._set("veadk.model.request_scope", "sdk_request") + self._set( + "veadk.model.first_token_status", + "no_output" if streaming else "non_streaming", + ) + + def _set(self, name: str, value: Any) -> None: + for span in self._spans: + if span.is_recording(): + span.set_attribute(name, value) + + def __enter__(self) -> ModelTiming: + return self + + def observe(self, output_type: str | None) -> None: + if not self.streaming or self.first_token_time is not None or not output_type: + return + self.first_token_time = time.time_ns() + self._set("veadk.model.first_token_time_unix_ns", self.first_token_time) + self._set("veadk.model.first_token_type", output_type) + self._set("veadk.model.first_token_status", "observed") + for span in self._spans: + if span.is_recording(): + span.add_event( + "veadk.model.first_token", + {"veadk.model.first_token_type": output_type}, + timestamp=self.first_token_time, + ) + + def __exit__(self, exc_type, exc_value, traceback) -> None: + outcome = "completed" + if exc_type is not None: + outcome = ( + "cancelled" + if issubclass(exc_type, (asyncio.CancelledError, GeneratorExit)) + else "error" + ) + self._set("veadk.model.outcome", outcome) + self.span.end() + + +def mark_uninstrumented_model(span: Span, *, streaming: bool) -> None: + annotate_request_start(span) + attributes = getattr(span, "attributes", None) or {} + if "veadk.model.start_time_unix_ns" in attributes or not span.is_recording(): + return + span.set_attribute("veadk.model.streaming", streaming) + span.set_attribute( + "veadk.model.first_token_status", + "unsupported" if streaming else "non_streaming", + ) + + +def _field(value: Any, name: str, default=None): + return ( + value.get(name, default) + if isinstance(value, Mapping) + else getattr(value, name, default) + ) + + +def litellm_output_type(chunk: Any) -> str | None: + """Recognize generated deltas before ADK buffers tool/reasoning content.""" + for choice in _field(chunk, "choices", []) or []: + delta = _field(choice, "delta") + if _field(delta, "content"): + return "text" + if _field(delta, "reasoning_content") or _field(delta, "reasoning"): + return "reasoning" + for block in _field(delta, "thinking_blocks", []) or []: + if _field(block, "thinking"): + return "reasoning" + functions = [ + _field(tool, "function") for tool in _field(delta, "tool_calls", []) or [] + ] + functions.append(_field(delta, "function_call")) + if any( + _field(function, "name") or _field(function, "arguments") + for function in functions + ): + return "tool_call" + return None + + +def ark_output_type(event: Any) -> str | None: + if not _field(event, "delta"): + return None + return { + "response.output_text.delta": "text", + "response.reasoning_summary_text.delta": "reasoning", + "response.reasoning_text.delta": "reasoning", + "response.function_call_arguments.delta": "tool_call", + }.get(_field(event, "type")) From d9ef697b82fde94373737aae518148df348e60b0 Mon Sep 17 00:00:00 2001 From: "wujiaming.ai" Date: Wed, 28 Oct 2026 15:57:02 +0800 Subject: [PATCH 2/2] test: stabilize concurrent startup prefetch and isolate sidecar environment --- .../2026-10-28-stabilize-concurrent-tests.md | 85 +++++++++++++++++++ ...026-10-28-stabilize-concurrent-tests.zh.md | 85 +++++++++++++++++++ .../2026-10-28-first-token-timestamps.md | 4 +- .../2026-10-28-first-token-timestamps.zh.md | 4 +- tests/cli/test_studio_rbac.py | 33 +++++-- .../harness/sidecar_runtime/test_sidecar.py | 8 +- 6 files changed, 208 insertions(+), 11 deletions(-) create mode 100644 prd-spec/bugfixes/pr1190-ci/2026-10-28-stabilize-concurrent-tests.md create mode 100644 prd-spec/bugfixes/pr1190-ci/2026-10-28-stabilize-concurrent-tests.zh.md diff --git a/prd-spec/bugfixes/pr1190-ci/2026-10-28-stabilize-concurrent-tests.md b/prd-spec/bugfixes/pr1190-ci/2026-10-28-stabilize-concurrent-tests.md new file mode 100644 index 000000000..3aa821005 --- /dev/null +++ b/prd-spec/bugfixes/pr1190-ci/2026-10-28-stabilize-concurrent-tests.md @@ -0,0 +1,85 @@ +# Stabilize PR #1190 CI tests + +[中文版](2026-10-28-stabilize-concurrent-tests.zh.md) + +Change ID: `pr1190-ci`; created/revised: 2026-10-28; status: approved. +Comparison base: feature commit `7527a832`; upstream `23014771`. +Related design: [first-token timestamps](../../features/first-token-timestamps/2026-10-28-first-token-timestamps.md). + +## Evidence and scope + +Unit Tests run `40649026439`, Python 3.10 job `122684493009`, reports 1 failed, 7,937 passed, 46 skipped, 4 xfailed. The failing startup-prefetch test assumes `tool-codex` is inspected before `tool-codex-snapshot`. Production intentionally uses `asyncio.gather` and `asyncio.to_thread`, so either order is valid. Python 3.12 was cancelled by matrix fail-fast. + +The earlier local full regression also exposed a baseline Sidecar test that inherits developer model upstream settings, activating a stable relay while asserting a fixed fake discovery URL. `run_with_harness_sidecar` already supports an explicit `env` mapping. + +Goal: deterministic tests for the actual contracts. Non-goals: runtime changes, serializing production concurrency, weakening assertions, retries/skips, CI configuration changes, frontend changes, dependencies, additional cloud deployments. + +## Requirements and design + +- FR-1: Both configured Sandbox tools are inspected exactly once, regardless of scheduling order, and their final cached states appear in the system-info response. +- FR-2: Explicitly exercise both inspection orders using a bounded thread Event; compare exact call multiplicities without requiring a production order. +- FR-3: Poll the returned state within a bounded deadline instead of treating entry into the inspection stub as cache publication. +- FR-4: The fake Sidecar child receives exact discovery endpoints in an explicitly isolated environment, even if the parent has model/MCP upstream variables. + +Change only `tests/cli/test_studio_rbac.py` and `tests/extensions/harness/sidecar_runtime/test_sidecar.py`, plus bilingual design/evidence. Preserve every existing response and endpoint assertion. No component spec changes: all public/runtime/permission/data/observability contracts are unchanged. Rollback reverts this test-only commit. + +## Tasks and acceptance + +| Requirement | Task | Acceptance | Evidence | +| --- | --- | --- | --- | +| FR-1/2 | T-1: force both thread orders, then correct the ordered assertion | AC-1: reverse scheduling fails before the correction; both schedules pass after it | Targeted pytest, original CI failure | +| FR-3 | T-2: wait for both published response states | AC-2: exact existing model-state assertions pass | Same parameterized TestClient test | +| FR-4 | T-3: give the Sidecar fake an explicit empty environment | AC-3: exact endpoint assertions pass with parent upstream variables present | Fake runtime and real child subprocess test | +| All | T-4: required gates, publish, inspect CI | AC-4: affected suites, full default regression, pre-commit and remote CI green; production diff coverage stays >=95% | Repository commands below | + +Commands: `uv run --no-sync --extra dev pytest tests/cli/test_studio_rbac.py tests/extensions/harness/sidecar_runtime/test_sidecar.py -n 2 -q`; `uv run --no-sync --extra dev pytest -n 2 -m "not codex_smoke and not piagent_smoke" -q`; `uv run --no-sync --extra dev pre-commit run --all-files`. Reuse feature coverage and real cloud evidence because production code is unchanged; verify its source hashes. + +## Review and authorization + +2026-10-28: User authorized repairing CI on PR #1190 after completing the Feishu design document. Earlier commit/push authorization persists for this branch. Direct design review (`review-spec` unavailable) confirms test-only scope, exact multiplicity/state/URL oracles, bounded synchronization, unchanged production concurrency and isolation from real credentials. English/Chinese identifiers and acceptance criteria agree. No design blocker remains. + +Risks: thread entry does not imply cache publication; response-state polling addresses this. Local Python 3.12 differs from CI's 3.10/3.12 matrix; remote results are required for delivery. Cloud revalidation is not applicable to a test-only repair; preserve the previously executed 6/6 real cloud conclusion. + +## Verification record + +2026-10-28, `7527a832` plus this test-only diff, Python 3.12.13: + +- Original ordered assertion with controlled schedules: 1 passed, 1 failed; reverse order reproduces the exact CI failure. +- Parent-upstream injection before explicit `env={}`: 1 failed with fixed discovery URL versus relay URL. Original isolated single-case execution passed, confirming the environment-dependent mechanism. +- After corrections: 3 targeted cases passed (two schedules and child environment). +- Combined affected run: 185 passed, 4 failed. All 174 Studio cases passed; four unrelated loopback requests used the local machine proxy and received HTTP 403. Rerunning the complete 15-case Sidecar file with command-scoped `NO_PROXY=localhost,127.0.0.1,::1 no_proxy=localhost,127.0.0.1,::1` passed. Thus all 189 affected cases passed in the appropriate local environment; no network test assertion or proxy policy was changed. +- All-files pre-commit passed after formatting; bilingual identifiers, literal commands, relative links and whitespace checks passed. +- All nine production file hashes match the original real-cloud build. Production changed-line coverage remains 100% (208/208); real cloud E2E remains 6/6 pass. This repair adds no production executable lines. +- Full default regression with the same command-scoped loopback bypass and `--junitxml` outside Git: **7,942 passed, 0 failed, 42 skipped, 4 xfailed; 6 subtests passed; 472.66s**. The 7,988 selected cases comprise 7,946 executed cases (including four expected failures) and 42 skipped cases. Final remote 3.10/3.12 results are tracked on the published PR. The user subsequently limited additional test execution to failed cases; no additional broad test run is planned. + + +### Gate status + +| Gate | Status | Evidence | +| --- | --- | --- | +| Change Basis | PASS | Complete test/design diff against `7527a832`; upstream `23014771` unchanged; no unrelated work. | +| Contract | PASS | Exact tool multiplicities, published states and child discovery endpoints; production contracts unchanged. | +| Impact | PASS | R-CI-1/2 cover the affected test oracles and environment boundary. | +| Portfolio | PASS | Two controlled thread schedules, TestClient state publication, real fake-runtime/child processes. | +| Baseline | PASS | Reverse-order assertion and parent-upstream injection both observed RED before correction. | +| Execution | PASS | Targeted 3 cases; all 189 affected cases; full default regression; all-files pre-commit. | +| Evaluation | PASS | Original state/URL assertions retained; duplicate calls still fail; source hashes and 208/208 production coverage unchanged. | +| Handoff | PASS | Bilingual design/evidence reconciled; final published CI evidence is linked from PR #1190. | + +### Change Impact + +| Change ID | Changed contract | Affected surface | Repository evidence | Failure mode | Risk | +| --- | --- | --- | --- | --- | --- | +| FR-1/2/3 | Test scheduling and completion oracle | Startup-prefetch test and system-info cache response | `asyncio.gather`, `asyncio.to_thread`, `test_system_info_prefetches_codex_model_env_state_on_startup` | False CI failure on valid scheduling, or checking states before publication | R-CI-1 Medium | +| FR-4 | Test subprocess environment | Fake Sidecar and child execution | `run_with_harness_sidecar(env=...)`, child-environment test | Parent upstream variables change expected discovery endpoints | R-CI-2 Medium | + +Both risks cross owned integration boundaries; inherent impact/exposure/hazard/detection are Medium/Medium/Medium/Low. No material unknown remains in the changed test scope. Residual uncertainty is bounded to untested host loads; no load/performance claim is made. + +### Test Portfolio + +| Risk ID | Boundary | Method | Cases | Oracle | Evidence | Status | +| --- | --- | --- | --- | --- | --- | --- | +| R-CI-1 | Threaded provider stub and real ASGI response | Controlled schedules and state transitions | Codex first; snapshot first; exact per-kind final state | Both tool IDs exactly once; controlled first entry; all original state/error assertions | RED: 1 failed/1 passed; GREEN: both cases and affected/full suites | VERIFIED | +| R-CI-2 | Real fake-runtime and child subprocesses | Parent-environment injection and isolation | Model/MCP parent upstreams; explicit empty child base environment | Exit 0 and exact original model/MCP URLs | RED: 1 failed; GREEN: targeted case and all 15 Sidecar tests | VERIFIED | + +Production, frontend, dependencies, CI policies, credentials and cloud behavior are not changed. Existing source-hash evidence discriminates accidental production drift. No new tests of formatting or duplicated implementation were added. Opt-in Codex/Pi smoke checks are outside this test-only repair; remote CI still runs its existing Codex smoke job. Final remote results are authoritative at https://github.com/volcengine/veadk-python/pull/1190/checks. diff --git a/prd-spec/bugfixes/pr1190-ci/2026-10-28-stabilize-concurrent-tests.zh.md b/prd-spec/bugfixes/pr1190-ci/2026-10-28-stabilize-concurrent-tests.zh.md new file mode 100644 index 000000000..a525de1d3 --- /dev/null +++ b/prd-spec/bugfixes/pr1190-ci/2026-10-28-stabilize-concurrent-tests.zh.md @@ -0,0 +1,85 @@ +# 稳定 PR #1190 CI 测试 + +[English](2026-10-28-stabilize-concurrent-tests.md) + +Change ID：`pr1190-ci`；创建/修订：2026-10-28；状态：approved。 +比较基线：功能提交 `7527a832`；上游 `23014771`。 +关联设计：[首 Token 时间戳](../../features/first-token-timestamps/2026-10-28-first-token-timestamps.zh.md)。 + +## 证据与范围 + +Unit Tests 运行 `40649026439` 的 Python 3.10 作业 `122684493009`：1 failed、7,937 passed、46 skipped、4 xfailed。启动预取测试要求先检查 `tool-codex`，再检查 `tool-codex-snapshot`。生产代码使用 `asyncio.gather` 与 `asyncio.to_thread` 并发执行，两种顺序都有效。Python 3.12 被矩阵 fail-fast 取消。 + +此前本地全量回归还暴露了一条 Sidecar 基线测试：继承开发机模型上游配置后启用稳定中继，却断言固定的伪运行时发现地址。`run_with_harness_sidecar` 已支持显式 `env` 映射。 + +目标：按真实契约稳定测试。非目标：修改运行时、串行化生产并发、放宽断言、添加重试或跳过、修改 CI 配置、前端、依赖和再次部署云端。 + +## 需求与设计 + +- FR-1：两个已配置 Sandbox 工具均恰好检查一次，不依赖调度顺序；最终缓存状态出现在 system-info 响应中。 +- FR-2：通过有时限的线程 Event 明确覆盖两种检查顺序；比较精确调用次数，不要求生产顺序。 +- FR-3：有界轮询响应状态，不能把进入检查桩当作缓存已发布。 +- FR-4：伪 Sidecar 的子进程在显式隔离环境中收到精确发现地址，即使父进程存在模型/MCP 上游变量。 + +仅修改 `tests/cli/test_studio_rbac.py`、`tests/extensions/harness/sidecar_runtime/test_sidecar.py` 和双语设计/证据。保留全部现有响应及地址断言。不修改组件规格：公共接口、运行时、权限、数据与可观测契约均不变。回滚只需撤回本次测试提交。 + +## 任务与验收 + +| 需求 | 任务 | 验收 | 证据 | +| --- | --- | --- | --- | +| FR-1/2 | T-1：控制两种线程顺序，再修正顺序断言 | AC-1：修正前反向调度失败，修正后两种调度均通过 | 定向 pytest 与原始 CI 失败 | +| FR-3 | T-2：等待两种响应状态均发布 | AC-2：全部现有模型状态精确断言通过 | 同一个参数化 TestClient 测试 | +| FR-4 | T-3：为伪 Sidecar 显式提供空环境 | AC-3：父进程存在上游变量时，精确地址断言仍通过 | 伪运行时与真实子进程测试 | +| 全部 | T-4：必需检查、推送并确认 CI | AC-4：受影响套件、默认全量回归、pre-commit 与远端 CI 通过；生产变更行覆盖率保持 >=95% | 下述仓库命令 | + +命令：`uv run --no-sync --extra dev pytest tests/cli/test_studio_rbac.py tests/extensions/harness/sidecar_runtime/test_sidecar.py -n 2 -q`；`uv run --no-sync --extra dev pytest -n 2 -m "not codex_smoke and not piagent_smoke" -q`；`uv run --no-sync --extra dev pre-commit run --all-files`。生产代码不变，复用功能覆盖率与真实云端证据，并确认源码哈希。 + +## 审查与授权 + +2026-10-28：用户授权完成飞书方案后修复 PR #1190 CI；此前对此分支的提交、推送授权继续有效。直接设计审查（`review-spec` 不可用）确认仅修改测试、精确次数/状态/地址断言、有界同步、生产并发不变，且隔离真实凭证。双语标识与验收一致，无设计阻塞。 + +风险：进入检查线程不代表缓存已发布，通过响应状态轮询解决。本地 Python 3.12 与 CI 的 3.10/3.12 矩阵不同，交付需远端结果。仅测试修复不适用再次云端验证；保留此前真实云端 6/6 通过的结论。 + +## 验证记录 + +2026-10-28，`7527a832` 加本次仅测试变更，Python 3.12.13： + +- 原有顺序断言加受控调度：1 passed、1 failed；反向顺序复现原 CI 失败。 +- 显式 `env={}` 前注入父进程上游变量：1 failed，固定发现地址与中继地址不符。原始单用例隔离运行通过，确认问题依赖环境。 +- 修正后：3 项定向用例通过（两种调度与子进程环境）。 +- 合并受影响套件：185 passed、4 failed。Studio 的全部 174 项通过；另四项回环请求走本机代理并收到 HTTP 403。对完整 15 项 Sidecar 文件使用命令级 `NO_PROXY=localhost,127.0.0.1,::1 no_proxy=localhost,127.0.0.1,::1` 后全部通过。因此 189 项受影响用例在正确本地环境中均通过；未修改网络断言或代理策略。 +- 格式修正后 all-files pre-commit 通过；双语标识、字面命令、相对链接与空白检查通过。 +- 九个生产文件哈希与原真实云端构建一致。生产变更行覆盖率仍为 100%（208/208），真实云端 E2E 仍为 6/6 通过。本次修复未新增生产可执行行。 +- 默认全量回归使用同一命令级回环直连配置，`--junitxml` 保存于 Git 外：**7,942 passed、0 failed、42 skipped、4 xfailed；6 subtests passed；472.66s**。选中 7,988 项，包括已执行 7,946 项（含四项预期失败）及跳过 42 项。最终远端 3.10/3.12 结果由推送后的 PR 记录。用户随后要求额外测试仅运行失败项，不再安排广泛测试。 + + +### Gate 状态 + +| Gate | Status | 证据 | +| --- | --- | --- | +| Change Basis | PASS | 对 `7527a832` 检查完整测试/设计差异；上游 `23014771` 未变；无无关变更。 | +| Contract | PASS | 工具精确调用次数、已发布状态及子进程发现地址；生产契约不变。 | +| Impact | PASS | R-CI-1/2 覆盖受影响测试断言与环境边界。 | +| Portfolio | PASS | 两种受控线程调度、TestClient 状态发布、真实伪运行时/子进程。 | +| Baseline | PASS | 修正前反向顺序断言与父进程上游注入均已观察到 RED。 | +| Execution | PASS | 3 项定向用例、全部 189 项受影响用例、默认全量回归与 all-files pre-commit。 | +| Evaluation | PASS | 保留原状态/地址断言；重复调用仍失败；源码哈希及 208/208 生产覆盖率不变。 | +| Handoff | PASS | 双语设计/证据同步；最终推送版本 CI 证据由 PR #1190 链接提供。 | + +### Change Impact + +| Change ID | Changed contract | Affected surface | Repository evidence | Failure mode | Risk | +| --- | --- | --- | --- | --- | --- | +| FR-1/2/3 | 测试调度与完成判定 | 启动预取测试与 system-info 缓存响应 | `asyncio.gather`、`asyncio.to_thread`、`test_system_info_prefetches_codex_model_env_state_on_startup` | 有效调度导致 CI 误报，或状态发布前检查响应 | R-CI-1 Medium | +| FR-4 | 测试子进程环境 | 伪 Sidecar 与子进程执行 | `run_with_harness_sidecar(env=...)`、子进程环境测试 | 父进程上游变量改变预期发现地址 | R-CI-2 Medium | + +两项风险均跨越自有集成边界；固有影响/暴露/变更危险/检测困难为 Medium/Medium/Medium/Low。本次测试范围没有重大未知项。剩余不确定性限于未测的主机负载；不作负载或性能结论。 + +### Test Portfolio + +| Risk ID | Boundary | Method | Cases | Oracle | Evidence | Status | +| --- | --- | --- | --- | --- | --- | --- | +| R-CI-1 | 线程化提供方桩与真实 ASGI 响应 | 受控调度与状态转换 | Codex 先执行、snapshot 先执行、各类型精确最终状态 | 两个工具 ID 各一次；受控首个入口；原有全部状态/错误断言 | RED：1 failed/1 passed；GREEN：两用例与受影响/全量套件 | VERIFIED | +| R-CI-2 | 真实伪运行时与子进程 | 父环境注入与隔离 | 父模型/MCP 上游；显式空子进程基环境 | 退出 0，原有模型/MCP 地址完全一致 | RED：1 failed；GREEN：定向用例及全部 15 项 Sidecar 测试 | VERIFIED | + +生产、前端、依赖、CI 策略、凭证与云端行为均不变。现有源码哈希证据可以识别意外生产漂移。没有新增格式测试或重复实现测试。需显式启用的 Codex/Pi smoke 不属于仅测试修复；远端 CI 仍执行既有 Codex smoke 作业。最终远端结果以 https://github.com/volcengine/veadk-python/pull/1190/checks 为准。 diff --git a/prd-spec/features/first-token-timestamps/2026-10-28-first-token-timestamps.md b/prd-spec/features/first-token-timestamps/2026-10-28-first-token-timestamps.md index a22609b66..6f33deb28 100644 --- a/prd-spec/features/first-token-timestamps/2026-10-28-first-token-timestamps.md +++ b/prd-spec/features/first-token-timestamps/2026-10-28-first-token-timestamps.md @@ -2,7 +2,7 @@ [中文版](2026-10-28-first-token-timestamps.zh.md) -Change ID: `first-token-timestamps`; created/revised: 2026-10-28; status: approved (implementation and feature verification complete; repository default regression is not fully green). +Change ID: `first-token-timestamps`; created/revised: 2026-10-28; status: approved (implementation and local verification complete; current publication checks are tracked in PR #1190). Component: [model-timing](../../../specs/model-timing/README.md). Base: `23014771`. ## Background, scope, and approval @@ -112,3 +112,5 @@ Sanitized evidence is retained outside Git in `veadk-first-token-20261028`; conc Request start is server ASGI ingress or direct Runner execution, not browser submission/render. Model start is adapter SDK request, not provider inference. Opaque LiteLLM/provider retries remain within one SDK request; external/custom engines have explicit unavailable status. No new TTFT/TPOT histogram or cross-host duration guarantee is added. Cloud proof covers one Beijing Runtime and the two supported adapters/model modes; no broad provider matrix or load benchmark is claimed. The existing Sidecar regression above remains outside this change. Commit and push are subsequently authorized; no PR is requested. Git history and the remote feature branch record publication. Pre-publication verification (2026-10-28): fetched and rebased against `origin/main` (`23014771`, unchanged); all applicable pre-commit hooks passed and all 244 affected tests passed again. Production SHA256 values still match the real cloud E2E build. + +CI follow-up (2026-10-28): [test stabilization](../../bugfixes/pr1190-ci/2026-10-28-stabilize-concurrent-tests.md) corrects the concurrent Sandbox test oracle and isolates the existing Sidecar fixture. The former baseline failure is resolved. Latest local default regression: **7,942 passed, 0 failed, 42 skipped, 4 xfailed; 6 subtests passed**. Historical failures above describe the original run. All nine production hashes, 100% (208/208) changed production-line coverage, and 6/6 real cloud E2E evidence remain unchanged. PR creation and the CI follow-up are now explicitly authorized. Published checks are available at [PR #1190](https://github.com/volcengine/veadk-python/pull/1190/checks). diff --git a/prd-spec/features/first-token-timestamps/2026-10-28-first-token-timestamps.zh.md b/prd-spec/features/first-token-timestamps/2026-10-28-first-token-timestamps.zh.md index e47c32f71..6d4676eac 100644 --- a/prd-spec/features/first-token-timestamps/2026-10-28-first-token-timestamps.zh.md +++ b/prd-spec/features/first-token-timestamps/2026-10-28-first-token-timestamps.zh.md @@ -2,7 +2,7 @@ [English](2026-10-28-first-token-timestamps.md) -变更 ID：`first-token-timestamps`；创建/修订：2026-10-28；状态：approved（实现与需求验证完成；仓库默认回归未全绿）。 +变更 ID：`first-token-timestamps`；创建/修订：2026-10-28；状态：approved（实现与本地验证已完成；当前发布检查由 PR #1190 记录）。 组件：[model-timing](../../../specs/model-timing/README.zh.md)。基线：`23014771`。 ## 背景、范围和授权 @@ -112,3 +112,5 @@ Chat 流式/非流式、Responses 流式/非流式及并发 Chat/Responses 流 请求起点是服务端 ASGI 入口或直接 Runner 执行，不是浏览器提交/渲染。模型起点是适配器 SDK 请求，不是 Provider 推理。不可见的 LiteLLM/Provider 内部重试计入一次 SDK 请求；外部/自定义引擎明确标记不可用。未新增 TTFT/TPOT 直方图或跨主机耗时保证。云端证据覆盖北京一个 Runtime、两个支持的适配器及模型调用模式，不宣称完整 Provider 矩阵或负载基准。上述既有 Sidecar 回归在本次变更范围之外。用户随后授权 commit 和 push，未要求创建 PR。发布记录见 Git 历史和远端功能分支。 发布前验证（2026-10-28）：已 fetch 并相对 `origin/main`（`23014771`，未变化）执行 rebase；全部适用的 pre-commit hook 通过，244 项受影响测试再次全部通过。生产文件 SHA256 仍与真实云端 E2E 构建一致。 + +CI 后续修复（2026-10-28）：[测试稳定性修复](../../bugfixes/pr1190-ci/2026-10-28-stabilize-concurrent-tests.zh.md)修正 Sandbox 并发测试断言，并隔离既有 Sidecar 测试环境。原基线失败已解决。最新本地默认全量回归：**7,942 passed、0 failed、42 skipped、4 xfailed；6 subtests passed**。上文失败属于原始运行历史。九个生产源码哈希、100%（208/208）生产变更行覆盖率及真实云端 E2E 6/6 证据均不变。用户现已明确授权创建 PR 和修复 CI。推送后的检查结果见 [PR #1190](https://github.com/volcengine/veadk-python/pull/1190/checks)。 diff --git a/tests/cli/test_studio_rbac.py b/tests/cli/test_studio_rbac.py index 6d515479f..fc5211324 100644 --- a/tests/cli/test_studio_rbac.py +++ b/tests/cli/test_studio_rbac.py @@ -1328,9 +1328,11 @@ def test_system_info_lists_configured_sandbox_tool_ids( assert user_denied.status_code == 403 +@pytest.mark.parametrize("first_tool_id", ["tool-codex", "tool-codex-snapshot"]) def test_system_info_prefetches_codex_model_env_state_on_startup( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, + first_tool_id: str, ) -> None: from veadk.cli.studio_sandbox_tools import StudioCodexModelEnvironmentStatus @@ -1339,9 +1341,14 @@ def test_system_info_prefetches_codex_model_env_state_on_startup( monkeypatch.setenv("VEADK_STUDIO_TOS_BUCKET", "teststudio") monkeypatch.setenv("VEADK_STUDIO_TOS_REGION", "cn-beijing") inspect_calls: list[dict[str, Any]] = [] + first_inspected = Event() def _inspect_codex_env(**kwargs: Any) -> StudioCodexModelEnvironmentStatus: + if kwargs["tool_id"] != first_tool_id: + assert first_inspected.wait(timeout=5) inspect_calls.append(kwargs) + if kwargs["tool_id"] == first_tool_id: + first_inspected.set() if kwargs["tool_id"] == "tool-codex": return StudioCodexModelEnvironmentStatus( needs_model_env_update=True, @@ -1377,16 +1384,28 @@ def _inspect_codex_env(**kwargs: Any) -> StudioCodexModelEnvironmentStatus: ) with TestClient(app) as client: - for _ in range(100): - if len(inspect_calls) >= 2: + deadline = time.monotonic() + 5 + while True: + response = client.get( + "/web/system-info", + headers={"Authorization": f"Bearer {_unsigned_jwt({'sub': 'admin'})}"}, + ) + assert response.status_code == 200 + sandbox_tools = { + item["kind"]: item for item in response.json()["sandboxTools"] + } + if all( + sandbox_tools[kind]["needsModelEnvUpdate"] + for kind in ("codex", "codex_snapshot") + ): break + assert time.monotonic() < deadline, ( + "Startup prefetch state was not published" + ) time.sleep(0.01) - response = client.get( - "/web/system-info", - headers={"Authorization": f"Bearer {_unsigned_jwt({'sub': 'admin'})}"}, - ) - assert [call["tool_id"] for call in inspect_calls] == [ + assert inspect_calls[0]["tool_id"] == first_tool_id + assert sorted(call["tool_id"] for call in inspect_calls) == [ "tool-codex", "tool-codex-snapshot", ] diff --git a/tests/extensions/harness/sidecar_runtime/test_sidecar.py b/tests/extensions/harness/sidecar_runtime/test_sidecar.py index 2ceca446b..08975eda5 100644 --- a/tests/extensions/harness/sidecar_runtime/test_sidecar.py +++ b/tests/extensions/harness/sidecar_runtime/test_sidecar.py @@ -619,8 +619,12 @@ def do_POST(self) -> None: # noqa: N802 def test_run_wraps_child_with_sidecar_environment( - fake_runtime: Path, tmp_path: Path + fake_runtime: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: + monkeypatch.setenv( + "MODEL_AGENT_API_BASE", "https://parent-model.example.test/api/v3" + ) + monkeypatch.setenv("MCP_URLS", "https://parent-mcp.example.test/metrics") output = tmp_path / "child-env.json" child = tmp_path / "child.py" child.write_text( @@ -631,7 +635,7 @@ def test_run_wraps_child_with_sidecar_environment( ) exit_code = run_with_harness_sidecar( - _config(fake_runtime), [sys.executable, str(child), str(output)] + _config(fake_runtime), [sys.executable, str(child), str(output)], env={} ) assert exit_code == 0