diff --git a/README.md b/README.md index a32484cec..29d1014bb 100644 --- a/README.md +++ b/README.md @@ -182,3 +182,11 @@ Join our discussion group by scanning the QR code below: ## License This project is licensed under the [Apache 2.0 License](./LICENSE). + +## Codex 性能诊断 + +The Codex path retains `call_llm` as a whole-turn compatibility record for session indexing and evaluation. Each Responses Shim backend `litellm.aresponses` call has a separate `codex.model.request` Span under the originating Agent context, including library-internal retries; the reasoning compatibility retry creates another Span. This is not a per-HTTP-attempt or TTFT measurement: the backend returns a complete non-streaming response. Model Spans do not duplicate whole-turn usage or record bodies, endpoints or raw exceptions. + +验证使用 `tests/runtime/codex/test_model_tracing.py` 与现有 `test_codex_tracing.py`、`test_codex_shim_rounds.py`；实际观测接收仍需按部署环境验证。 + +The remote CodeEnv Worker client traces readiness, session creation, turn start/status/cancel and each SSE connection as CLIENT Spans, propagating W3C context. Retries, backoff, authentication, idempotency keys and event cursors preserve their existing semantics. Stream Spans are detached from ambient context across generator yields. Spans exclude endpoint, routing IDs, bodies and credentials. Subscription time includes caller consumption and must not be interpreted as remote execution time. Remote queue/start timing and cross-turn recovery still require receiver instrumentation. diff --git a/docs/content/docs/framework/tracing.en.mdx b/docs/content/docs/framework/tracing.en.mdx index a5e300a9a..05a89d4a2 100644 --- a/docs/content/docs/framework/tracing.en.mdx +++ b/docs/content/docs/framework/tracing.en.mdx @@ -163,3 +163,7 @@ 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. + +### Session event persistence duration + +VeADK's built-in SQLite, MySQL and PostgreSQL session backends emit `veadk.session.append_event` for actual non-partial event append calls, helping identify storage delays within execution. Streaming partial events are not counted as database saves, and custom session services are not automatically instrumented. Existing export settings apply; event bodies, session state and database URLs are not recorded. The duration covers the ADK storage operation, not an individual SQL statement. diff --git a/docs/content/docs/framework/tracing.mdx b/docs/content/docs/framework/tracing.mdx index 696075724..24c96b469 100644 --- a/docs/content/docs/framework/tracing.mdx +++ b/docs/content/docs/framework/tracing.mdx @@ -161,3 +161,7 @@ Tracing 数据由一系列 **Span** 组成，每个 Span 记录一次操作（ VeADK 的字段命名遵循 OpenTelemetry 生成式 AI 规范，你可以直接将 Tracing 数据导入到 OpenTelemetry 兼容的系统中进行分析和可视化。 + +### 会话事件保存耗时 + +VeADK 内置 SQLite、MySQL、PostgreSQL 会话后端为实际保存事件的 `append_event` 调用生成 `veadk.session.append_event` Span，便于判断会话保存是否拖慢执行。流式 partial 事件不作为数据库保存计时；自定义会话后端不自动增加此埋点。复用现有上报配置，不记录事件正文、会话状态或数据库地址。该耗时覆盖 ADK 存储调用，不等于某条 SQL 的执行时间。 diff --git a/prd-spec/features/execution-diagnostics/2026-10-10-execution-diagnostics.md b/prd-spec/features/execution-diagnostics/2026-10-10-execution-diagnostics.md new file mode 100644 index 000000000..6fa0912f2 --- /dev/null +++ b/prd-spec/features/execution-diagnostics/2026-10-10-execution-diagnostics.md @@ -0,0 +1,28 @@ +# Execution diagnostics + +[中文](2026-10-10-execution-diagnostics.zh.md) + +Change ID: execution-diagnostics; date: 2026-10-10; status: in-review. +Related contract: [execution tracing](../../../specs/execution-tracing/README.md). + +## Evidence and scope +The existing Codex `call_llm` Span covers a whole execution. The Responses Shim performs non-streaming `litellm.aresponses` requests. At the design baseline, Worker HTTP requests were traced, while the SSE subscription was not. This design records the scope already authorized by the performance-diagnostics goal; it does not assert a new user approval or deployed capability. + +## Requirements and design +FR-1: Preserve whole-turn compatibility while recording each model backend call separately under the originating Agent context. No duplicate whole-turn usage or synthetic TTFT. +FR-2: Record Worker requests and each SSE connection, propagate only W3C tracing context, and preserve authentication, idempotency, retry limits, event validation and cursors. +FR-3: Stream Spans must not remain current across an async-generator yield. End on completion, failure, cancellation or generator close. Record no endpoint, routing IDs, bodies or raw errors. +FR-4: Missing optional telemetry must preserve execution. Reuse the existing Provider/exporter. + +## Tasks and acceptance +T-1 / AC-1: Model Span tests prove parentage, retries, failures and cancellation; existing Codex indexing remains compatible. +T-2 / AC-2: Worker HTTP tests prove W3C propagation, redaction and unchanged keys. +T-3 / AC-3: SSE tests prove per-connection parentage, cursor continuity, cancellation/close and no current-context leak. +T-4 / AC-4: Synchronize bilingual contracts and user documentation, run affected tests and pre-commit, and record actual results. + +## Review and risks +Direct review: no public signature, protocol, authentication, persistence, concurrency ownership or deployment configuration changes. Existing retry semantics remain authoritative. Async-generator context leakage is a blocking risk addressed by detached stream Spans. Client subscription time is not remote queue/start/execution time. Real Worker receiver propagation, APMPlus reception and deployed versions remain unverified. SSE implementation and affected-test acceptance are complete in draft PR #1161; HTTP/model evidence is also recorded there. This record reconciles existing work and precedes the SSE extension. + +Verification (2026-10-10): 10 Worker tracing tests pass; 17 existing remote Sandbox tests pass (the initial combined run before adding new tests had 22 passes). SSE completion, reconnect/cursor, early close, cancellation and HTTP failure are covered. Full repository regression, live Codex smoke and APMPlus reception are not_run: isolated full-smoke dependencies and deployed receiver evidence are unavailable. Process deviation: SSE tests were added after the first implementation edit instead of first run failing; direct behavior assertions are now present. No deployed capability is claimed. + +Final checks: 27 affected tests pass after formatting; all-files pre-commit (Ruff and both secret scans) pass. Requirement scope is authorized by the existing performance-diagnostics goal. Direct design review found no remaining blocker for this extension; server-side verification remains separate. diff --git a/prd-spec/features/execution-diagnostics/2026-10-10-execution-diagnostics.zh.md b/prd-spec/features/execution-diagnostics/2026-10-10-execution-diagnostics.zh.md new file mode 100644 index 000000000..ad4ee51c7 --- /dev/null +++ b/prd-spec/features/execution-diagnostics/2026-10-10-execution-diagnostics.zh.md @@ -0,0 +1,28 @@ +# 执行诊断 + +[English](2026-10-10-execution-diagnostics.md) + +变更 ID：execution-diagnostics；日期：2026-10-10；状态：in-review。 +相关契约：[执行链路](../../../specs/execution-tracing/README.zh.md)。 + +## 依据与范围 +现有 Codex call_llm Span 覆盖整个执行；Responses Shim 使用非流式 litellm.aresponses。设计基线时 Worker HTTP 请求已有记录，SSE 订阅尚未记录。本设计记录性能诊断 goal 已授权的范围，不宣称新增用户审批或已部署能力。 + +## 要求与设计 +FR-1：保留整轮兼容记录，单独记录各次模型后端请求并关联原 Agent；不重复记录整轮用量，不计算合成 TTFT。 +FR-2：记录 Worker 请求与每次 SSE 连接，仅传递 W3C 链路上下文；认证、幂等、重试上限、事件校验和游标保持原行为。 +FR-3：流 Span 不得在异步生成器 yield 期间保持为当前 Span；完成、失败、取消和关闭生成器时结束。不记录 Endpoint、路由标识、正文和原始异常。 +FR-4：缺少可选观测依赖时保持执行；复用既有 Provider 和导出器。 + +## 任务与验收 +T-1 / AC-1：模型测试证明父子关联、重试、失败与取消，Codex 索引兼容。 +T-2 / AC-2：Worker HTTP 测试证明 W3C 传递、脱敏与 Key 不变。 +T-3 / AC-3：SSE 测试证明各次连接父子关系、游标连续、取消与关闭，以及当前上下文不泄漏。 +T-4 / AC-4：同步双语契约及使用文档，运行受影响测试和 pre-commit，记录实际结果。 + +## 评审与风险 +直接评审：不改公开签名、协议、认证、持久化、并发职责或部署配置。重试遵循现有语义。异步生成器上下文泄漏是必须解决的问题，使用不设为当前上下文的流 Span。客户端订阅时间不是远端排队、启动或执行耗时。Worker 接收端传递、APMPlus 接收与部署版本尚未验证。SSE 实现及受影响测试验收已在草稿 PR #1161 中完成，HTTP / 模型测试证据也记录在该 PR。本记录补齐已有改动依据，并先于 SSE 扩展建立。 + +验证（2026-10-10）：10 项 Worker 链路测试通过，17 项现有远端 Sandbox 测试通过（新增用例前的初次合并运行共 22 项通过）。覆盖 SSE 完成、重连游标、提前关闭、取消和 HTTP 失败。全仓回归、真实 Codex smoke 和 APMPlus 接收为 not_run：缺少隔离的完整 smoke 依赖与部署接收证据。流程偏差：SSE 测试在首次实现编辑后添加，没有先执行失败测试；现已补充直接行为断言。不宣称已部署能力。 + +最终检查：格式化后 27 项受影响测试通过；全文件 pre-commit（Ruff 和两项敏感信息扫描）通过。需求范围由既有性能诊断 goal 授权。直接设计评审未发现本扩展剩余阻断项；服务端验证仍单独进行。 diff --git a/prd-spec/features/retrieval-diagnostics/2026-10-10-retrieval-diagnostics.md b/prd-spec/features/retrieval-diagnostics/2026-10-10-retrieval-diagnostics.md new file mode 100644 index 000000000..67b0ea579 --- /dev/null +++ b/prd-spec/features/retrieval-diagnostics/2026-10-10-retrieval-diagnostics.md @@ -0,0 +1,64 @@ +# Retrieval diagnostics + +[中文](2026-10-10-retrieval-diagnostics.zh.md) + +Change ID: retrieval-diagnostics; revision: 2026-10-10; status: approved. + +## Background and scope + +`veadk/knowledgebase/knowledgebase.py:Knowledgebase.search` calls backend search and normalizes results. `veadk/memory/long_term_memory.py:LongTermMemory.search_memory` catches backend errors and returns empty memories; OpenViking calls run through `asyncio.to_thread`. `add_session_to_memory` filters events and saves them. These boundaries currently have no independent retrieval/write Span. The existing tool Span cannot distinguish retrieval from surrounding work. + +The user authorized the overall performance-diagnostics implementation, local verification and draft PR delivery. This change implements those already agreed retrieval boundaries. It does not authorize production deployment. + +## Goals and non-goals + +Measure actual knowledge search, memory search and memory save boundaries, inheriting the existing OTel context and Provider. Preserve signatures, results, error propagation and cancellation. Do not introduce exporters, sampling configuration, storage, dashboards or backend-internal embedding/reranking stages. Existing observability owns receiving, export, storage and queries; diagnostics owns the operation boundaries and evidence used to identify slow operations. + +## Requirements and scenarios + +- FR-1: Knowledge search creates `veadk.knowledge.search`, ending after result normalization; records backend type, effective top_k and result count. Backend failure retains its exception and records ERROR. +- FR-2: Memory search creates `veadk.memory.search`; backend failure remains an empty response under the existing API, but Span status is ERROR with safe error type. Empty successful results remain successful. +- FR-3: Memory save creates `veadk.memory.save`, covering filtering and backend save; records backend type, session ID and filtered event count. Errors still propagate. +- FR-4: All operations inherit parent context, finish on success/error/cancellation, and restore the caller context. No raw exception events, query, memory content, credentials, user ID, index, URL or arbitrary kwargs are exported. A telemetry failure must not block or repeat business execution. + +Given a tool calling knowledge search through `asyncio.to_thread`, the retrieval Span is a child of the existing tool Span. Given two concurrent memory requests, each remains under its own parent. Given a cancelled OpenViking await, the caller Span ends with cancellation; the underlying synchronous thread may continue, so the Span must not claim backend completion. + +## Design and contract impact + +Use a small internal helper under `veadk/tracing/` for fail-open fixed-name spans with explicit safe outcome handling and no exporter setup. Reuse `trace.get_tracer`; do not change global Provider. The helper activates a span for the operation and restores context in finally, disables automatic exception recording, and records only error class/status. Avoid catching business exceptions in the helper as success. Backend attributes come from configured backend type, never full configuration. Apply result counts only after successful normalization. Memory search marks its handled backend exception before returning the established empty response. `asyncio.to_thread` already copies context; add no extra worker propagation protocol. + +Public signatures, result types, backend arguments, auth, retry behavior and memory isolation remain unchanged. There is no new persistent state or deployment configuration. Extend the bilingual execution-tracing spec with the three boundaries and their failure semantics. + +## Tasks and acceptance + +| Requirement | Task | Acceptance | Verification | Status | +| --- | --- | --- | --- | --- | +| FR-1 | T-1 knowledge boundary | AC-1 parent/name/effective top_k/count and unchanged errors | Targeted knowledge tests with local SpanExporter | pass (local) | +| FR-2 | T-2 memory search | AC-2 handled failure distinguished from successful empty response | Memory tests: success, empty, failure, concurrent parents, OpenViking cancellation | pass (local) | +| FR-3 | T-3 memory save | AC-3 event count/session, propagation and unchanged save arguments | Memory save tests: filtering, backend failure and cancellation | pass (local) | +| FR-4 | T-4 protections | AC-4 no sensitive content; telemetry failures do not alter business calls/context | Helper tests with failing tracer/span plus integration parent tests | pass (local) | +| FR-1–FR-4 | T-5 delivery | AC-5 bilingual docs and affected regression gates, actual PR recorded | Repository-local targeted pytest; all-files pre-commit; synchronize base before commit | pass (draft PR) | + +Test entry points: `tests/test_knowledgebase.py`, `tests/test_long_term_memory.py`, `tests/test_openviking_long_term_memory.py`, `tests/tools/builtin_tools/test_load_knowledgebase.py`; add colocated tracing-helper tests where existing tests organize tracing. Execute through repository `.venv` using `uv run --extra dev pytest`. No real services or credentials in fixtures. Use a controlled blocking backend for thread-cancellation semantics and clean it up before test teardown. + +## Risks and review + +Backend internals and deployed APMPlus reception remain unverified. A cancelled thread-backed call may still run remotely; measure caller boundary only and mark cancelled, never backend success. Global telemetry setup can be absent: no-op tracing must leave execution unchanged. Do not add duplicate tokens/cost or infer unknown internal stages. + +Design review: pass (2026-10-10). Verified actual public boundaries, context propagation through to_thread, existing handled-error semantics, bounded attributes and matched bilingual requirements. No blockers; thread completion remains explicitly outside caller-cancellation evidence. Production code/tests begin with TDD after this review. Approval scope comes from the user's existing active goal; no new deployment permission is inferred. + +## Local implementation evidence (2026-10-10) + +| Check | Result and scope | +| --- | --- | +| TDD | Helper test initially failed because the module did not exist. Initial boundary failures were fixture setup errors, not valid missing-Span evidence; corrected isolated backend fixtures before asserting the public methods. | +| Retrieval/Worker subset | 76 passed, 1 optional-dependency skip; helper 100% lines and 98% combined branch coverage. Includes concurrent parents, thread cancellation with explicit drain, telemetry mutation/end failures, attribute whitelist and actual tool-to-retrieval context. | +| Initial broad regression | fail: 6607 passed, 16 failed, 59 skipped, 4 xfailed, 2 collection errors (422.59s). All failures/errors reported missing llama_index, openai_codex or anthropic. No tests removed or reclassified. | +| Environment recovery | Installed declared dev/extensions/codex/sandbox extras. Codex binary download first timed out at 30s; retry with command-scoped UV_HTTP_TIMEOUT=180 exited 0. No global configuration changed. | +| Affected regression after recovery | pass: 199 passed, no skips/errors, seven existing dependency deprecation warnings (46.95s). Covers the failed Harness/Codex suites, both Sandbox suites, retrieval/memory/tool context and Worker requests. | +| Final source/security review | pass: backend arguments, filtering, successful empty responses, errors and cancellation preserved. No new exporter/configuration or sensitive content. | +| Base and hooks | Fresh origin/main remains 171d8d86, already rebased and an ancestor of HEAD. All-files pre-commit passed Ruff and both secret scanners with complete extras. | +| Full regression | pass: 6698 passed, 50 skipped, 4 xfailed, 88 dependency warnings, 427.41s. Skips do not prove behavior. First interrupted run had no result; restarted only after confirming absence. Local metrics export reported localhost:8000 unavailable, so remote observability reception is not proved. | +| Delivery/deployment | draft PR delivered: [#1161](https://github.com/volcengine/veadk-python/pull/1161), code commit e2c1e887. Remote PR head verified after push. No merge or deployment; APMPlus reception and real performance remain unverified. | + +Evidence: `/tmp/codex-retrieval-save-tests.log`, `/tmp/codex-veadk-retrieval-broad.log`, `/tmp/codex-veadk-complete-env-retry.log`, `/tmp/codex-veadk-complete-env-retest.log`, `/tmp/codex-retrieval-extras-final-hooks.log`, `/tmp/codex-veadk-complete-broad-resumed.log`. diff --git a/prd-spec/features/retrieval-diagnostics/2026-10-10-retrieval-diagnostics.zh.md b/prd-spec/features/retrieval-diagnostics/2026-10-10-retrieval-diagnostics.zh.md new file mode 100644 index 000000000..4ee526c36 --- /dev/null +++ b/prd-spec/features/retrieval-diagnostics/2026-10-10-retrieval-diagnostics.zh.md @@ -0,0 +1,64 @@ +# 检索与记忆性能诊断 + +[English](2026-10-10-retrieval-diagnostics.md) + +Change ID：retrieval-diagnostics；修订：2026-10-10；状态：approved。 + +## 背景与范围 + +`veadk/knowledgebase/knowledgebase.py:Knowledgebase.search` 调用后端检索并归一化结果。`veadk/memory/long_term_memory.py:LongTermMemory.search_memory` 捕获后端错误后返回空记忆；OpenViking 调用经过 `asyncio.to_thread`。`add_session_to_memory` 过滤事件并保存。这些边界目前没有独立检索 / 写入 Span，外层工具 Span 无法区分检索与其他工作。 + +用户已授权整体性能诊断实现、本地验证和草稿 PR 交付。本变更补齐已约定的检索边界，不包含生产部署授权。 + +## 目标与非目标 + +记录真实知识检索、记忆检索及保存边界，继承已有 OTel 上下文和 Provider，保持接口、返回值、异常与取消行为。不新增导出器、采样配置、存储、看板，也不拆造后端内部向量生成 / 重排阶段。可观测负责接收、导出、存储和查询；性能诊断负责调用边界及定位慢调用的证据。 + +## 需求与场景 + +- FR-1：知识检索生成 `veadk.knowledge.search`，结果归一化后结束；记录后端类型、实际 top_k 和结果数量。后端失败保持抛出原异常，并记录 ERROR。 +- FR-2：记忆检索生成 `veadk.memory.search`；后端失败保持现有空返回，但 Span 标为 ERROR 并记录安全错误类型。正常空结果仍是成功。 +- FR-3：记忆保存生成 `veadk.memory.save`，覆盖事件过滤与后端保存；记录后端类型、会话 ID 及过滤后的事件数量。错误继续抛出。 +- FR-4：继承父上下文，在成功、错误、取消时结束并恢复调用方上下文。不导出原始异常事件、查询、记忆正文、凭证、用户 ID、索引、URL 或任意 kwargs。观测失败不得阻断或重复业务执行。 + +工具通过 `asyncio.to_thread` 调用知识检索时，检索 Span 应是已有工具 Span 的子调用。两个并发记忆请求分别归属各自父调用。取消等待 OpenViking 时，调用方 Span 以取消结束；底层同步线程可能继续，不得声称后端已完成。 + +## 设计与契约影响 + +在 `veadk/tracing/` 增加小型内部 helper，固定操作名、安全记录结果，观测失败不影响业务，不配置导出器。复用 `trace.get_tracer`，不改变全局 Provider。helper 在调用期间激活 Span，finally 恢复上下文；关闭自动异常记录，只保留错误类型与状态，不把业务异常吞成成功。后端属性取配置的类型，不导出完整配置。结果归一化成功后才写数量。记忆检索捕获后端异常时先标失败，再保留既有空返回。`asyncio.to_thread` 已复制上下文，不增加并行传播协议。 + +公共签名、结果类型、后端参数、鉴权、重试和记忆隔离不变。没有新增持久化状态和部署配置。扩展双语 execution-tracing spec，明确三个边界及错误语义。 + +## 任务与验收 + +| 需求 | 任务 | 验收 | 验证 | 状态 | +| --- | --- | --- | --- | --- | +| FR-1 | T-1 知识检索 | AC-1 父关系、名称、实际 top_k、数量与原异常不变 | 本地 SpanExporter 的知识检索专项测试 | pass (local) | +| FR-2 | T-2 记忆检索 | AC-2 区分已处理失败与正常空结果 | 成功、空结果、失败、并发父关系、OpenViking 取消测试 | pass (local) | +| FR-3 | T-3 记忆保存 | AC-3 事件数量 / 会话、关联和保存参数不变 | 过滤、后端失败及取消测试 | pass (local) | +| FR-4 | T-4 采集保护 | AC-4 不含敏感正文；采集失败不改变业务调用与上下文 | tracer / span 故障 helper 测试及父关系集成测试 | pass (local) | +| FR-1–FR-4 | T-5 交付 | AC-5 双语文档与受影响回归门禁、记录实际 PR | 仓库本地专项 pytest、全文件 pre-commit；提交前同步基线 | pass (draft PR) | + +测试入口：`tests/test_knowledgebase.py`、`tests/test_long_term_memory.py`、`tests/test_openviking_long_term_memory.py`、`tests/tools/builtin_tools/test_load_knowledgebase.py`；按现有 tracing 测试组织新增 helper 测试。使用仓库 `.venv`，命令为 `uv run --extra dev pytest`。fixture 不接真实服务，不使用凭证。线程取消使用受控阻塞后端，退出测试前完成清理。 + +## 风险与评审 + +后端内部阶段及部署环境 APMPlus 接收尚未验证。取消线程调用后远端可能继续，只测调用方边界并标取消，不标后端成功。全局观测未初始化时，无操作采集不得改变执行。不重复记录 Token / 成本，不推断未知内部阶段。 + +设计评审：pass（2026-10-10）。已核对公共边界、to_thread 上下文传播、既有捕获异常语义、属性白名单及双语需求一致性。无阻塞项；取消调用方不证明线程完成，验收保留此限制。评审后通过 TDD 进入代码和测试。授权来自用户现有 active goal，不推导新的部署权限。 + +## 本地实现证据（2026-10-10） + +| 检查 | 结果与范围 | +| --- | --- | +| TDD | helper 测试最初因模块不存在失败。首轮边界失败是 fixture 初始化问题，不作为缺失 Span 的有效证据；修正隔离后端 fixture 后才验证公开方法。 | +| 检索 / Worker 专项 | 76 项通过，1 项可选依赖跳过；helper 行覆盖率 100%，综合分支覆盖率 98%。覆盖并发父关系、线程取消与显式清理、采集修改 / 结束故障、属性白名单及实际工具到检索的关联。 | +| 首次广回归 | fail：6607 项通过、16 项失败、59 项跳过、4 项预期失败、2 项收集错误（422.59 秒）。失败与错误均报告缺少 llama_index、openai_codex 或 anthropic；未删除测试或修改分类。 | +| 环境修复 | 安装仓库声明的 dev / extensions / codex / sandbox extras。Codex 二进制首次下载因 30 秒超时失败；仅对重试命令设置 UV_HTTP_TIMEOUT=180，退出 0。未修改全局配置。 | +| 修复后受影响回归 | pass：199 项通过，无跳过或错误，7 条现有依赖弃用警告（46.95 秒）。覆盖此前失败的 Harness / Codex 测试、两组 Sandbox 测试、检索 / 记忆 / 工具关联与 Worker 请求。 | +| 最终源码 / 安全审查 | pass：保留后端参数、过滤、正常空结果、异常与取消语义；没有新增 exporter / 配置或敏感正文采集。 | +| 基线与 hooks | 重新 fetch 后 origin/main 仍为 171d8d86，已完成 rebase 且为 HEAD 的祖先。保留完整 extras 的全文件 pre-commit 已通过 Ruff 和两项密钥扫描。 | +| 全量回归 | pass：6698 项通过、50 项跳过、4 项预期失败、88 条依赖警告，耗时 427.41 秒。跳过不证明相关行为。首轮中断无结果，仅确认进程不存在后重跑。本地 metrics 导出报告 localhost:8000 不可用，不证明远端可观测接收。 | +| 交付 / 部署 | 草稿 PR 已交付：[#1161](https://github.com/volcengine/veadk-python/pull/1161)，代码提交 e2c1e887。推送后已核验远端 PR 提交；没有合并或部署，APMPlus 接收与真实性能仍未验证。 | + +证据：`/tmp/codex-retrieval-save-tests.log`、`/tmp/codex-veadk-retrieval-broad.log`、`/tmp/codex-veadk-complete-env-retry.log`、`/tmp/codex-veadk-complete-env-retest.log`、`/tmp/codex-retrieval-extras-final-hooks.log`、`/tmp/codex-veadk-complete-broad-resumed.log`。 diff --git a/prd-spec/features/session-persistence-diagnostics/2026-10-10-session-persistence.md b/prd-spec/features/session-persistence-diagnostics/2026-10-10-session-persistence.md new file mode 100644 index 000000000..e4cfcd70f --- /dev/null +++ b/prd-spec/features/session-persistence-diagnostics/2026-10-10-session-persistence.md @@ -0,0 +1,18 @@ +# Session persistence diagnostics + +## Authorization and evidence +The user authorized completing instrumentation against the AgentKit diagnostic scheme, including Runtime persistence phases. ADK Runner calls SessionService.append_event during execution; AgentkitAgentServerApp preserves caller-provided SessionService identity. VeADK constructs DatabaseSessionService in its SQLite, MySQL and PostgreSQL factories. + +## Design and scope +Use an internal DatabaseSessionService subclass in these three factories. Time actual append_event calls as veadk.session.append_event, preserving argument, return, state, failure and cancellation behavior. Partial events bypass timing because ADK does not persist them. Reuse the existing OTel provider; telemetry failure cannot block the database operation. Record no database URL, credentials, event body, session state or raw exception. Custom SessionService and the SDK identity contract remain unchanged. No new exporter, storage or schema. + +## Review +Reviewed against existing factories and SDK identity tests. No method monkeypatching, protocol or configuration changes. Bilingual maintained tracing spec must describe scope. SQLite verifies real persistence; patched base-method tests cover cancellation and telemetry failure. MySQL/PostgreSQL factories are checked without connecting to shared databases. Remote deployment and APMPlus receipt remain unverified. + +## Tasks and acceptance +- Add internal traced database service and wire all three owned factories. +- Verify real SQLite event/state persistence, partial events, parent context, cancellation, failure redaction and fail-open telemetry. +- Update tracing contract and record exact tests before commit. + +## Verification +Affected tests: `uv run --extra dev --extra extensions --extra codex --extra sandbox pytest tests/memory/test_session_persistence_tracing.py tests/memory/test_postgres_schema.py tests/test_adk_compat.py tests/test_short_term_memory.py -q`: 32 passed. Real ADK Runner persisted events/state through local SQLite; parent Trace, identifiers, partial-event bypass, error redaction, cancellation and telemetry fail-open verified. Full regression completed: 6704 passed, 50 skipped, 4 xfailed (484.07 s). Skips do not prove environment-dependent behavior. diff --git a/prd-spec/features/session-persistence-diagnostics/2026-10-10-session-persistence.zh.md b/prd-spec/features/session-persistence-diagnostics/2026-10-10-session-persistence.zh.md new file mode 100644 index 000000000..e6a9b8cda --- /dev/null +++ b/prd-spec/features/session-persistence-diagnostics/2026-10-10-session-persistence.zh.md @@ -0,0 +1,18 @@ +# 会话持久化诊断 + +## 授权与依据 +用户已授权按 AgentKit 诊断方案完成埋点，包括 Runtime 持久化阶段。ADK Runner 在执行中调用 SessionService.append_event；AgentkitAgentServerApp 保留调用方提供的 SessionService 实例。VeADK 在 SQLite、MySQL 和 PostgreSQL 工厂中创建 DatabaseSessionService。 + +## 设计与范围 +三个内置工厂使用内部 DatabaseSessionService 子类，为真实 append_event 调用生成 veadk.session.append_event Span；参数、返回值、状态更新、异常和取消行为保持一致。partial 事件不计时，因为 ADK 不持久化它们。复用已有 OTel Provider，采集故障不阻断数据库调用。不记录数据库地址、凭证、事件正文、会话状态或异常正文。不修改自定义 SessionService 与 SDK 的实例身份约定，不新增导出通道、存储或表结构。 + +## 评审 +已对照现有工厂和 SDK 实例身份测试评审。不修改对象方法，不改协议或配置。同步双语 tracing 规范说明范围。SQLite 验证真实持久化；替换基类方法的测试验证取消和采集故障。MySQL / PostgreSQL 工厂在不连接共享数据库的条件下检查。远端部署与 APMPlus 接收仍未验证。 + +## 任务与验收 +- 增加内部数据库计时服务，接入三个内置工厂。 +- 验证 SQLite 的事件与状态保存、partial 事件、父上下文、取消、异常脱敏和采集故障。 +- 更新 tracing 规范，提交前记录实际测试结果。 + +## 验证 +受影响测试：`uv run --extra dev --extra extensions --extra codex --extra sandbox pytest tests/memory/test_session_persistence_tracing.py tests/memory/test_postgres_schema.py tests/test_adk_compat.py tests/test_short_term_memory.py -q`：32 项通过。真实 ADK Runner 经本地 SQLite 保存事件／状态；父 Trace、关联标识、partial 事件不计时、异常脱敏、取消和采集故障不阻断均已验证。完整回归：6704 项通过、50 项跳过、4 项预期失败（484.07 秒）。跳过项不计作环境验证。 diff --git a/specs/execution-tracing/README.md b/specs/execution-tracing/README.md new file mode 100644 index 000000000..180492e3a --- /dev/null +++ b/specs/execution-tracing/README.md @@ -0,0 +1,21 @@ +# Execution tracing + +[中文](README.zh.md) + +Component: execution-tracing; status: draft; revision: 2026-10-10. +Owns `veadk/runtime/codex/model_tracing.py` and `_remote_sandbox/request_tracing.py`; integrates the Responses Shim and Worker client. Design: [execution diagnostics](../../prd-spec/features/execution-diagnostics/2026-10-10-execution-diagnostics.md). + +CON-1: `codex.model.request` covers one backend `aresponses` call including internal retries; `call_llm` remains whole-turn. No model TTFT or duplicate usage is inferred. +CON-2: Worker request Spans preserve existing retries, auth, keys and errors, and propagate W3C context without telemetry baggage. +CON-3: each SSE connection has a detached CLIENT Span, with no ambient context held across yield. Event cursor, binding checks, retry bounds and execution ownership remain unchanged. +CON-4: Fixed operation names and error types only; no credentials, URL, body or raw exception. Existing Provider owns sampling and export; no persistence or new configuration. + +Tests: `tests/agents/test_worker_request_tracing.py`, model tracing tests and existing remote Sandbox regression. CON-3 completion, reconnect/cursor, early close, cancellation and HTTP failure tests pass in the draft PR. Real receiving/export/deployment evidence is required separately; this draft makes no server-side instrumentation guarantee. + +## Retrieval boundaries + +CON-5 (implemented locally; not deployed): `veadk.knowledge.search`, `veadk.memory.search` and `veadk.memory.save` inherit the existing context and Provider. Record backend type, effective top_k/result count or session/event count only; never query/content/user ID/index/URL. Handled memory-search backend failures must remain empty responses but mark Span ERROR; normal empty results are successful. Cancellation ends the caller boundary without claiming the synchronous thread/backend stopped. No exporter or persistent state is added. Design and pending acceptance: [retrieval diagnostics](../../prd-spec/features/retrieval-diagnostics/2026-10-10-retrieval-diagnostics.md). + +## Session persistence + +CON-6 (implemented locally; not deployed): VeADK-owned SQLite, MySQL and PostgreSQL factories use an internal DatabaseSessionService subclass to record non-partial append_event calls as `veadk.session.append_event`. The Span covers the ADK storage operation, including serialization, state updates and database work; it does not split individual SQL statements or prove remote durable commit. Only actual session/invocation identifiers and error type are recorded. Existing Provider controls sampling/export. Arguments, returned events, errors, cancellation, caching and custom caller-owned SessionService remain unchanged. Design: [session persistence diagnostics](../../prd-spec/features/session-persistence-diagnostics/2026-10-10-session-persistence.md). diff --git a/specs/execution-tracing/README.zh.md b/specs/execution-tracing/README.zh.md new file mode 100644 index 000000000..a437dd9b3 --- /dev/null +++ b/specs/execution-tracing/README.zh.md @@ -0,0 +1,21 @@ +# 执行链路 + +[English](README.md) + +组件：execution-tracing；状态：draft；更新：2026-10-10。 +负责 veadk/runtime/codex/model_tracing.py 与 _remote_sandbox/request_tracing.py，对接 Responses Shim 和 Worker 客户端。设计：[执行诊断](../../prd-spec/features/execution-diagnostics/2026-10-10-execution-diagnostics.zh.md)。 + +CON-1：codex.model.request 覆盖一次后端 aresponses 及库内重试；call_llm 仍覆盖整轮。不据此推算模型 TTFT，不重复记录用量。 +CON-2：Worker 请求 Span 保留既有重试、认证、Key 和错误，仅传递 W3C 上下文，不附加观测 baggage。 +CON-3：每次 SSE 连接记录独立 CLIENT Span，不在 yield 期间持有当前上下文。事件游标、绑定校验、重试上限与执行职责不变。 +CON-4：仅记录固定操作名与错误类型，不记录凭证、URL、正文或原始异常。既有 Provider 负责采样和导出，不新增持久化或配置。 + +测试：tests/agents/test_worker_request_tracing.py、模型计时测试和现有远端 Sandbox 回归。草稿 PR 中 CON-3 完成、重连游标、提前关闭、取消与 HTTP 失败测试已通过。真实接收、导出和部署证据单独验证；本草案不保证服务端已埋点。 + +## 检索边界 + +CON-5（本地已实现，尚未部署）：`veadk.knowledge.search`、`veadk.memory.search`、`veadk.memory.save` 继承既有上下文与 Provider。只记录后端类型、实际 top_k / 结果数量或会话 / 事件数量，不记录查询、正文、用户 ID、索引和 URL。记忆检索的已处理后端错误保留空返回，但 Span 标 ERROR；正常空结果是成功。取消时结束调用方边界，不声称同步线程或后端已停止。不新增导出器及持久化状态。设计及待执行验收见[检索诊断](../../prd-spec/features/retrieval-diagnostics/2026-10-10-retrieval-diagnostics.zh.md)。 + +## 会话持久化 + +CON-6（本地已实现，未部署）：VeADK 内置 SQLite、MySQL、PostgreSQL 工厂通过内部 DatabaseSessionService 子类，将非 partial 的 append_event 调用记录为 `veadk.session.append_event`。Span 包含 ADK 存储操作中的序列化、状态更新和数据库调用，不拆分各条 SQL，也不证明远端持久化提交。仅记录实际会话／执行标识与错误类型。采样和上报由现有 Provider 管理。参数、返回事件、错误、取消、缓存及调用方自定义 SessionService 保持不变。设计：[会话持久化诊断](../../prd-spec/features/session-persistence-diagnostics/2026-10-10-session-persistence.zh.md)。 diff --git a/tests/agents/test_worker_request_tracing.py b/tests/agents/test_worker_request_tracing.py new file mode 100644 index 000000000..2a070f911 --- /dev/null +++ b/tests/agents/test_worker_request_tracing.py @@ -0,0 +1,196 @@ +# 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. + + +"""验证真实 HTTPX 请求携带当前 Span，并保留重试及异常边界。""" + +import asyncio + +import httpx +import pytest +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.agents._remote_sandbox.codex_worker_client import ( + CodexWorkerClient, + CodexWorkerError, +) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("outcome", ["success", "retry", "failure", "cancel"]) +async def test_worker_request_propagation_and_redaction(monkeypatch, outcome): + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + monkeypatch.setattr(trace, "get_tracer", provider.get_tracer) + requests = [] + + async def respond(request): + requests.append(request) + if outcome == "cancel": + raise asyncio.CancelledError("secret") + if outcome == "retry" and len(requests) == 1: + return httpx.Response(503) + if outcome == "failure": + return httpx.Response(400) + return httpx.Response(200, json={"status": "running"}) + + client = CodexWorkerClient( + "https://worker.invalid?signature=secret", api_key="secret" + ) + client._http = httpx.AsyncClient( + transport=httpx.MockTransport(respond), headers=client._headers + ) + try: + with provider.get_tracer("test").start_as_current_span("agent") as parent: + if outcome == "cancel": + with pytest.raises(asyncio.CancelledError): + await client.start_turn( + "private-session", "secret task", "same-key" + ) + elif outcome == "failure": + with pytest.raises(CodexWorkerError): + await client.start_turn( + "private-session", "secret task", "same-key" + ) + else: + assert ( + await client.start_turn( + "private-session", "secret task", "same-key" + ) + )["status"] == "running" + span = next( + s for s in exporter.get_finished_spans() if s.name == "worker.turn.start" + ) + assert span.parent.span_id == parent.get_span_context().span_id + for request in requests: + parts = request.headers["traceparent"].split("-") + assert parts[1] == f"{span.context.trace_id:032x}" + assert parts[2] == f"{span.context.span_id:016x}" + assert request.headers["idempotency-key"] == "same-key" + assert request.headers["x-api-key"] == "secret" + assert "baggage" not in request.headers + assert "secret" not in span.to_json() + assert "private-session" not in span.to_json() + if outcome in {"failure", "cancel"}: + assert span.status.status_code.name == "ERROR" + assert "error.type" in span.attributes + if outcome == "retry": + assert len(requests) == 2 + assert len(span.events) == 2 + assert span.attributes["http.response.status_code"] == 200 + finally: + await client._http.aclose() + provider.shutdown() + + +@pytest.mark.asyncio +async def test_worker_request_without_observation(monkeypatch): + from veadk.agents._remote_sandbox import request_tracing + + monkeypatch.setattr(request_tracing, "trace", None) + + async def respond(request): + assert "traceparent" not in request.headers + return httpx.Response(200, json={"status": "ready"}) + + client = CodexWorkerClient("https://worker.invalid") + client._http = httpx.AsyncClient(transport=httpx.MockTransport(respond)) + try: + assert (await client.request("GET", "/readyz"))["status"] == "ready" + finally: + await client._http.aclose() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "outcome", ["complete", "reconnect", "close", "cancel", "failure"] +) +async def test_worker_stream_context_and_cleanup(monkeypatch, outcome): + import json + + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + monkeypatch.setattr(trace, "get_tracer", provider.get_tracer) + requests = [] + + async def respond(request): + requests.append(request) + if outcome == "cancel": + raise asyncio.CancelledError("secret") + if outcome == "failure": + return httpx.Response(403) + sequence = len(requests) + terminal = outcome == "complete" or sequence == 2 + event = { + "schemaVersion": 1, + "sessionId": "private-session", + "turnId": "private-turn", + "eventId": sequence, + "type": "turn.completed" if terminal else "tool.delta", + "payload": "secret", + } + return httpx.Response(200, content="data: " + json.dumps(event) + "\n\n") + + client = CodexWorkerClient( + "https://worker.invalid?signature=secret", api_key="secret" + ) + client._http = httpx.AsyncClient( + transport=httpx.MockTransport(respond), headers=client._headers + ) + try: + with provider.get_tracer("test").start_as_current_span("agent") as parent: + stream = client.events("private-session", "private-turn") + if outcome == "cancel": + with pytest.raises(asyncio.CancelledError): + await anext(stream) + elif outcome == "failure": + with pytest.raises(CodexWorkerError): + await anext(stream) + else: + seen = [] + async for event in stream: + seen.append(event["eventId"]) + assert ( + trace.get_current_span().get_span_context() + == parent.get_span_context() + ) + if outcome == "close": + await stream.aclose() + break + assert seen == ([1, 2] if outcome == "reconnect" else [1]) + spans = [ + s for s in exporter.get_finished_spans() if s.name == "worker.turn.events" + ] + assert len(spans) == len(requests) == (2 if outcome == "reconnect" else 1) + for request, span in zip(requests, spans): + assert span.parent.span_id == parent.get_span_context().span_id + parts = request.headers["traceparent"].split("-") + assert parts[2] == f"{span.context.span_id:016x}" + assert request.headers["x-api-key"] == "secret" + assert "baggage" not in request.headers + assert "secret" not in span.to_json() + assert "private-session" not in span.to_json() + assert requests[0].headers["last-event-id"] == "0" + if outcome == "reconnect": + assert requests[1].headers["last-event-id"] == "1" + if outcome in {"cancel", "failure"}: + assert spans[0].status.status_code.name == "ERROR" + finally: + await client._http.aclose() + provider.shutdown() diff --git a/tests/memory/test_session_persistence_tracing.py b/tests/memory/test_session_persistence_tracing.py new file mode 100644 index 000000000..7a7a39ed2 --- /dev/null +++ b/tests/memory/test_session_persistence_tracing.py @@ -0,0 +1,158 @@ +import asyncio + +import pytest +from google.adk.events import Event +from google.adk.events.event_actions import EventActions +from google.adk.sessions import DatabaseSessionService +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.memory.short_term_memory_backends.sqlite_backend import SQLiteSTMBackend + + +@pytest.fixture +def tracing(monkeypatch): + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + tracer = provider.get_tracer("test.session") + monkeypatch.setattr(trace, "get_tracer", lambda *args, **kwargs: tracer) + yield tracer, exporter + provider.shutdown() + + +@pytest.mark.asyncio +async def test_real_sqlite_persists_event_state_and_parent(tracing, tmp_path): + tracer, exporter = tracing + service = SQLiteSTMBackend(local_path=str(tmp_path / "session.db")).session_service + session = await service.create_session(app_name="app", user_id="user") + event = Event( + author="agent", actions=EventActions(state_delta={"answer": "private"}) + ) + with tracer.start_as_current_span("execution") as parent: + result = await service.append_event(session, event) + reloaded = await service.get_session( + app_name="app", user_id="user", session_id=session.id + ) + assert result.id == event.id + assert reloaded.state["answer"] == "private" + assert reloaded.events[-1].id == event.id + spans = [ + s + for s in exporter.get_finished_spans() + if s.name == "veadk.session.append_event" + ] + assert len(spans) == 1 + assert spans[0].parent.span_id == parent.get_span_context().span_id + assert spans[0].end_time >= spans[0].start_time + assert "private" not in spans[0].to_json() + await service.db_engine.dispose() + + +@pytest.mark.asyncio +async def test_partial_event_is_not_persistence(tracing, tmp_path): + _, exporter = tracing + service = SQLiteSTMBackend(local_path=str(tmp_path / "partial.db")).session_service + session = await service.create_session(app_name="app", user_id="user") + await service.append_event(session, Event(author="agent", partial=True)) + assert not exporter.get_finished_spans() + await service.db_engine.dispose() + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "error", [ValueError("private database credentials"), asyncio.CancelledError()] +) +async def test_failure_and_cancel_preserve_original_error( + tracing, tmp_path, monkeypatch, error +): + _, exporter = tracing + service = SQLiteSTMBackend(local_path=str(tmp_path / "error.db")).session_service + session = await service.create_session(app_name="app", user_id="user") + + async def fail(self, session, event): + raise error + + monkeypatch.setattr(DatabaseSessionService, "append_event", fail) + with pytest.raises(type(error)) as caught: + await service.append_event(session, Event(author="agent")) + assert caught.value is error + span = exporter.get_finished_spans()[0] + assert span.attributes["error.type"] == type(error).__name__ + assert "private database credentials" not in span.to_json() + assert span.status.status_code == trace.StatusCode.ERROR + await service.db_engine.dispose() + + +@pytest.mark.asyncio +async def test_telemetry_failure_does_not_skip_or_repeat_persistence( + tmp_path, monkeypatch +): + service = SQLiteSTMBackend( + local_path=str(tmp_path / "fail-open.db") + ).session_service + session = await service.create_session(app_name="app", user_id="user") + + def fail(*args, **kwargs): + raise RuntimeError("telemetry unavailable") + + monkeypatch.setattr(trace, "get_tracer", fail) + event = Event(author="agent") + await service.append_event(session, event) + stored = await service.get_session( + app_name="app", user_id="user", session_id=session.id + ) + assert [e.id for e in stored.events].count(event.id) == 1 + await service.db_engine.dispose() + + +@pytest.mark.asyncio +async def test_real_runner_persists_events_within_execution(tracing, tmp_path): + from google.adk.agents import BaseAgent + from google.adk.runners import Runner + from google.genai import types + + class ProbeAgent(BaseAgent): + async def _run_async_impl(self, ctx): + yield Event( + author=self.name, + invocation_id=ctx.invocation_id, + content=types.Content(role="model", parts=[types.Part(text="probe")]), + actions=EventActions(state_delta={"saved": True}), + ) + + tracer, exporter = tracing + service = SQLiteSTMBackend(local_path=str(tmp_path / "runner.db")).session_service + session = await service.create_session(app_name="app", user_id="user") + runner = Runner( + agent=ProbeAgent(name="probe"), app_name="app", session_service=service + ) + with tracer.start_as_current_span("runtime.execution") as parent: + events = [ + event + async for event in runner.run_async( + user_id="user", + session_id=session.id, + new_message=types.Content(role="user", parts=[types.Part(text="test")]), + ) + ] + reloaded = await service.get_session( + app_name="app", user_id="user", session_id=session.id + ) + assert reloaded.state["saved"] is True + assert events[-1].id in [event.id for event in reloaded.events] + saves = [ + s + for s in exporter.get_finished_spans() + if s.name == "veadk.session.append_event" + ] + assert len(saves) >= 2 + assert all(s.context.trace_id == parent.get_span_context().trace_id for s in saves) + assert all(s.attributes["gen_ai.session.id"] == session.id for s in saves) + assert any( + s.attributes.get("gen_ai.invocation.id") == events[-1].invocation_id + for s in saves + ) + await service.db_engine.dispose() diff --git a/tests/runtime/codex/test_codex_tracing.py b/tests/runtime/codex/test_codex_tracing.py index 17a5b8e3c..09c3a1924 100644 --- a/tests/runtime/codex/test_codex_tracing.py +++ b/tests/runtime/codex/test_codex_tracing.py @@ -113,6 +113,12 @@ async def fake_get_shim(api_base, api_key): assert call_llm_spans, ( f"no call_llm span: {sorted({s.name for s in exporter._spans})}" ) + # 模型请求来自独立的 Shim 服务任务，必须接回本轮 Agent 的调用链。 + model_spans = [s for s in exporter._spans if s.name == "codex.model.request"] + assert len(model_spans) == 1 + assert model_spans[0].context.trace_id == call_llm_spans[0].context.trace_id + assert model_spans[0].parent is not None + assert "backend-key" not in model_spans[0].to_json() attributes = dict(call_llm_spans[0].attributes or {}) assert attributes.get("gen_ai.session.id") == session_id, attributes # Operational metadata for dashboards and incident triage: which path diff --git a/tests/runtime/codex/test_model_tracing.py b/tests/runtime/codex/test_model_tracing.py new file mode 100644 index 000000000..baefaaa34 --- /dev/null +++ b/tests/runtime/codex/test_model_tracing.py @@ -0,0 +1,72 @@ +# 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. + +"""验证模型请求的真实父子关系、取消和异常脱敏。""" + +import asyncio + +import pytest +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.runtime.codex.model_tracing import trace_model_request + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure", [None, ValueError, asyncio.CancelledError]) +async def test_model_request_parent_and_safe_failure(monkeypatch, failure): + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + monkeypatch.setattr(trace, "get_tracer", provider.get_tracer) + + async def call(): + await asyncio.sleep(0) + if failure: + raise failure("private prompt and credential") + return "response" + + with provider.get_tracer("test").start_as_current_span("invoke_agent") as parent: + if failure: + with pytest.raises(failure): + await trace_model_request(call) + else: + assert await trace_model_request(call) == "response" + spans = exporter.get_finished_spans() + model = next(span for span in spans if span.name == "codex.model.request") + assert model.parent.span_id == parent.get_span_context().span_id + assert model.context.trace_id == parent.get_span_context().trace_id + assert model.end_time >= model.start_time + assert "private" not in model.to_json() + assert not model.events + if failure: + assert model.attributes["error.type"] == failure.__name__ + assert model.status.status_code.name == "ERROR" + else: + assert model.status.status_code.name == "UNSET" + provider.shutdown() + + +@pytest.mark.asyncio +async def test_model_request_without_tracing(monkeypatch): + from veadk.runtime.codex import model_tracing + + monkeypatch.setattr(model_tracing, "trace", None) + + async def call(): + return "response" + + assert await model_tracing.trace_model_request(call) == "response" diff --git a/tests/tracing/test_retrieval_tracing.py b/tests/tracing/test_retrieval_tracing.py new file mode 100644 index 000000000..47db9914c --- /dev/null +++ b/tests/tracing/test_retrieval_tracing.py @@ -0,0 +1,297 @@ +import asyncio + +import pytest +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 opentelemetry.trace import StatusCode + +from veadk.tracing.retrieval_tracing import retrieval_span + + +@pytest.fixture +def telemetry(monkeypatch): + exporter = InMemorySpanExporter() + provider = TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + monkeypatch.setattr(trace, "get_tracer", provider.get_tracer) + yield provider, exporter + provider.shutdown() + + +@pytest.mark.parametrize("outcome", ["success", "failure", "cancel"]) +def test_parent_cleanup_and_redaction(telemetry, outcome): + provider, exporter = telemetry + with provider.get_tracer("test").start_as_current_span("parent") as parent: + try: + with retrieval_span( + "veadk.memory.search", backend="openviking" + ) as operation: + operation.set_attribute("veadk.retrieval.result_count", 0) + if outcome == "failure": + raise ValueError("sensitive query and token") + if outcome == "cancel": + raise asyncio.CancelledError("sensitive query and token") + except (ValueError, asyncio.CancelledError): + pass + assert trace.get_current_span() is parent + span = exporter.get_finished_spans()[0] + assert span.parent.span_id == parent.get_span_context().span_id + assert span.attributes["veadk.retrieval.backend"] == "openviking" + assert not span.events + assert "sensitive" not in str(span.attributes) + assert (span.status.status_code == StatusCode.ERROR) == (outcome != "success") + + +def test_telemetry_failure_does_not_repeat_business(monkeypatch): + def fail(*args, **kwargs): + raise RuntimeError("telemetry unavailable") + + monkeypatch.setattr(trace, "get_tracer", fail) + calls = [] + with retrieval_span("veadk.memory.save", backend="local") as operation: + calls.append("save") + operation.set_attribute("veadk.memory.event_count", 1) + assert calls == ["save"] + + +@pytest.mark.parametrize("error", [False, True]) +def test_knowledge_boundary(telemetry, error, monkeypatch): + from types import SimpleNamespace + from veadk.knowledgebase.knowledgebase import KnowledgeBase + + def search(**kwargs): + assert kwargs == {"query": "private-query", "top_k": 3} + if error: + raise ValueError("private-query") + return ["private-result"] + + monkeypatch.setattr( + KnowledgeBase, + "model_post_init", + lambda instance, *args: object.__setattr__( + instance, "_backend", SimpleNamespace(search=search) + ), + ) + kb = KnowledgeBase.model_construct(backend="local", top_k=3) + kb._backend = SimpleNamespace(search=search) + if error: + with pytest.raises(ValueError): + kb.search("private-query") + else: + assert kb.search("private-query")[0].content == "private-result" + spans = telemetry[1].get_finished_spans() + assert len(spans) == 1 + assert spans[0].name == "veadk.knowledge.search" + assert spans[0].attributes["veadk.retrieval.top_k"] == 3 + assert "private" not in str(spans[0].attributes) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("error", [False, True]) +async def test_memory_empty_result_distinguishes_failure(telemetry, error, monkeypatch): + from types import SimpleNamespace + from veadk.memory.long_term_memory import LongTermMemory + + def search(**kwargs): + if error: + raise ValueError("private-query") + return [] + + monkeypatch.setattr(LongTermMemory, "model_post_init", lambda *args: None) + memory = LongTermMemory.model_construct(backend="local", top_k=3) + memory._backend = SimpleNamespace(search_memory=search) + response = await memory.search_memory( + app_name="test", user_id="private-user", query="private-query" + ) + assert response.memories == [] + spans = telemetry[1].get_finished_spans() + assert len(spans) == 1 + assert spans[0].name == "veadk.memory.search" + assert (spans[0].status.status_code == StatusCode.ERROR) == error + assert "private" not in str(spans[0].attributes) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("error", [False, True]) +async def test_memory_save_boundary(telemetry, error, monkeypatch): + from types import SimpleNamespace + from google.adk.sessions import Session + from veadk.memory.long_term_memory import LongTermMemory + + calls = [] + + def save(**kwargs): + calls.append(kwargs) + if error: + raise ValueError("private-memory") + + monkeypatch.setattr(LongTermMemory, "model_post_init", lambda *args: None) + memory = LongTermMemory.model_construct(backend="local", top_k=3) + memory._backend = SimpleNamespace(save_memory=save) + session = Session( + id="session-test", app_name="test", user_id="private-user", state={}, events=[] + ) + if error: + with pytest.raises(ValueError): + await memory.add_session_to_memory(session) + else: + await memory.add_session_to_memory(session) + assert len(calls) == 1 + assert calls[0]["user_id"] == "private-user" + span = telemetry[1].get_finished_spans()[0] + assert span.name == "veadk.memory.save" + assert span.attributes["session.id"] == "session-test" + assert span.attributes["veadk.memory.event_count"] == 0 + assert (span.status.status_code == StatusCode.ERROR) == error + assert "private" not in str(span.attributes) + + +@pytest.mark.asyncio +async def test_threaded_search_keeps_concurrent_parent_context(telemetry, monkeypatch): + import threading + from types import SimpleNamespace + from veadk.memory.long_term_memory import LongTermMemory + + barrier = threading.Barrier(2) + parent_by_query = {} + observed = {} + monkeypatch.setattr(LongTermMemory, "model_post_init", lambda *args: None) + memory = LongTermMemory.model_construct(backend="openviking", top_k=3) + + def search(**kwargs): + observed[kwargs["query"]] = trace.get_current_span().get_span_context() + barrier.wait(timeout=3) + return [] + + memory._backend = SimpleNamespace(search_memory=search) + + async def request(query): + with telemetry[0].get_tracer("test").start_as_current_span(query) as parent: + parent_by_query[query] = parent.get_span_context() + await memory.search_memory(app_name="test", user_id="user", query=query) + assert trace.get_current_span() is parent + + await asyncio.gather(request("first"), request("second")) + spans = {span.context.span_id: span for span in telemetry[1].get_finished_spans()} + for query, child in observed.items(): + assert spans[child.span_id].parent.span_id == parent_by_query[query].span_id + assert child.trace_id == parent_by_query[query].trace_id + + +@pytest.mark.asyncio +@pytest.mark.parametrize("operation_name", ["search", "save"]) +async def test_thread_cancellation_finishes_caller_without_claiming_backend_end( + telemetry, monkeypatch, operation_name +): + import threading + from types import SimpleNamespace + from veadk.memory.long_term_memory import LongTermMemory + + started = threading.Event() + release = threading.Event() + finished = threading.Event() + monkeypatch.setattr(LongTermMemory, "model_post_init", lambda *args: None) + memory = LongTermMemory.model_construct(backend="openviking", top_k=3) + + def search(**kwargs): + started.set() + try: + assert release.wait(timeout=3) + return [] + finally: + finished.set() + + memory._backend = SimpleNamespace(search_memory=search, save_memory=search) + if operation_name == "search": + call = memory.search_memory(app_name="test", user_id="user", query="query") + else: + from google.adk.sessions import Session + + call = memory.add_session_to_memory( + Session( + id="cancel-session", + app_name="test", + user_id="user", + state={}, + events=[], + ) + ) + task = asyncio.create_task(call) + try: + assert await asyncio.to_thread(started.wait, 2) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert not finished.is_set() + span = telemetry[1].get_finished_spans()[0] + assert span.attributes["error.type"] == "CancelledError" + assert span.status.status_code == StatusCode.ERROR + assert "veadk.retrieval.result_count" not in span.attributes + finally: + release.set() + assert await asyncio.to_thread(finished.wait, 2) + + +def test_span_mutation_and_end_failure_preserve_business_error(monkeypatch): + class BrokenSpan: + def set_attribute(self, *args): + raise RuntimeError("telemetry attribute failed") + + def set_status(self, *args): + raise RuntimeError("telemetry status failed") + + def end(self): + raise RuntimeError("telemetry end failed") + + def get_span_context(self): + return trace.INVALID_SPAN_CONTEXT + + from types import SimpleNamespace + + monkeypatch.setattr( + trace, + "get_tracer", + lambda *args: SimpleNamespace(start_span=lambda *a: BrokenSpan()), + ) + original = ValueError("business failure") + with pytest.raises(ValueError) as raised: + with retrieval_span("veadk.knowledge.search", backend="local") as operation: + operation.set_attribute("veadk.retrieval.top_k", 3) + raise original + assert raised.value is original + + +@pytest.mark.asyncio +async def test_real_knowledge_tool_thread_inherits_tool_parent(telemetry, monkeypatch): + from types import SimpleNamespace + from veadk.knowledgebase.knowledgebase import KnowledgeBase + from veadk.tools.builtin_tools.load_knowledgebase import LoadKnowledgebaseTool + + monkeypatch.setattr( + KnowledgeBase, + "model_post_init", + lambda instance, *args: object.__setattr__( + instance, "_backend", SimpleNamespace(search=lambda **kwargs: ["result"]) + ), + ) + kb = KnowledgeBase.model_construct(backend="local", top_k=3) + tool = LoadKnowledgebaseTool(knowledgebase=kb) + with telemetry[0].get_tracer("test").start_as_current_span("tool") as parent: + response = await tool.load_knowledgebase("private-query", tool_context=None) + assert trace.get_current_span() is parent + assert response.knowledges[0].content == "result" + child = telemetry[1].get_finished_spans()[0] + assert child.name == "veadk.knowledge.search" + assert child.parent.span_id == parent.get_span_context().span_id + + +def test_sensitive_attributes_are_not_recorded(telemetry): + with retrieval_span("veadk.memory.search", backend="local") as operation: + operation.set_attribute("query", "private-query") + operation.set_attribute("user.id", "private-user") + operation.set_attribute("endpoint", "private-endpoint") + assert dict(telemetry[1].get_finished_spans()[0].attributes) == { + "veadk.retrieval.backend": "local" + } diff --git a/veadk/agents/_remote_sandbox/codex_worker_client.py b/veadk/agents/_remote_sandbox/codex_worker_client.py index 83061dc1b..90e288c65 100644 --- a/veadk/agents/_remote_sandbox/codex_worker_client.py +++ b/veadk/agents/_remote_sandbox/codex_worker_client.py @@ -23,6 +23,8 @@ import httpx +from veadk.agents._remote_sandbox.request_tracing import worker_request_span + class CodexWorkerError(RuntimeError): """A sanitized worker failure; never includes credential URLs or wire data.""" @@ -75,21 +77,43 @@ def _endpoint(self, path): return self._url.copy_with(path="/v1/codex-worker" + path) async def request(self, method, path, *, body=None, key=None): + with worker_request_span(method, path) as (trace_headers, span): + return await self._request( + method, path, body=body, key=key, trace_headers=trace_headers, span=span + ) + + async def _request(self, method, path, *, body, key, trace_headers, span): for attempt in range(3): try: response = await self._http.request( method, self._endpoint(path), json=body, - headers={"Idempotency-Key": key} if key else None, + headers={ + **trace_headers, + **({"Idempotency-Key": key} if key else {}), + }, ) except httpx.TransportError: + if span: + span.add_event( + "worker.request.transport_error", {"attempt": attempt + 1} + ) if attempt == 2: raise CodexWorkerError( "Worker request unavailable; execution may have started" ) from None await asyncio.sleep(0.1 * (attempt + 1)) continue + if span: + span.set_attribute("http.response.status_code", response.status_code) + span.add_event( + "worker.request.response", + { + "attempt": attempt + 1, + "http.response.status_code": response.status_code, + }, + ) if response.status_code >= 400: # POST retries only use a stable idempotency key. A 5xx can # represent an accepted request, never generate a replacement key. @@ -145,50 +169,63 @@ async def events(self, sid, tid, *, after_event_id=0): cursor = after_event_id for attempt in range(4): try: - async with self._http.stream( - "GET", - self._endpoint(self.turn_path(sid, tid) + "/events"), - headers={"Last-Event-ID": str(cursor)}, - ) as response: - if response.status_code != 200: - raise CodexWorkerError( - f"Worker event stream returned HTTP {response.status_code}" - ) - data = [] - size = 0 - async for line in response.aiter_lines(): - size += len(line) - if size > 4 * 1024 * 1024: - raise CodexWorkerError("Worker event too large") - if line.startswith("data:"): - data.append(line[5:].lstrip()) - elif not line: - size = 0 - if not data: - continue - try: - event = json.loads("\n".join(data)) - except ValueError: - raise CodexWorkerError("Invalid worker event") from None - data = [] - if event.get("schemaVersion") != 1: - raise CodexWorkerError( - "Unsupported worker event or expired stream" - ) - if ( - event.get("sessionId") != sid - or event.get("turnId") != tid - ): - raise CodexWorkerError("Worker event binding mismatch") - sequence = event.get("eventId") - if not isinstance(sequence, int) or sequence < 1: - raise CodexWorkerError("Invalid worker event cursor") - if sequence <= cursor: - continue - cursor = sequence - yield event - if event["type"] == "turn.completed": - return + with worker_request_span( + "GET", self.turn_path(sid, tid) + "/events", activate=False + ) as (trace_headers, span): + async with self._http.stream( + "GET", + self._endpoint(self.turn_path(sid, tid) + "/events"), + headers={**trace_headers, "Last-Event-ID": str(cursor)}, + ) as response: + if span: + span.set_attribute( + "http.response.status_code", response.status_code + ) + if response.status_code != 200: + raise CodexWorkerError( + f"Worker event stream returned HTTP {response.status_code}" + ) + data = [] + size = 0 + async for line in response.aiter_lines(): + size += len(line) + if size > 4 * 1024 * 1024: + raise CodexWorkerError("Worker event too large") + if line.startswith("data:"): + data.append(line[5:].lstrip()) + elif not line: + size = 0 + if not data: + continue + try: + event = json.loads("\n".join(data)) + except ValueError: + raise CodexWorkerError( + "Invalid worker event" + ) from None + data = [] + if event.get("schemaVersion") != 1: + raise CodexWorkerError( + "Unsupported worker event or expired stream" + ) + if ( + event.get("sessionId") != sid + or event.get("turnId") != tid + ): + raise CodexWorkerError( + "Worker event binding mismatch" + ) + sequence = event.get("eventId") + if not isinstance(sequence, int) or sequence < 1: + raise CodexWorkerError( + "Invalid worker event cursor" + ) + if sequence <= cursor: + continue + cursor = sequence + yield event + if event["type"] == "turn.completed": + return except httpx.TransportError: pass if attempt < 3: diff --git a/veadk/agents/_remote_sandbox/request_tracing.py b/veadk/agents/_remote_sandbox/request_tracing.py new file mode 100644 index 000000000..9424fd1d0 --- /dev/null +++ b/veadk/agents/_remote_sandbox/request_tracing.py @@ -0,0 +1,72 @@ +# 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. + + +"""远端 Worker 调用的脱敏计时和 W3C 上下文透传。""" + +from contextlib import contextmanager + +try: + from opentelemetry import trace + from opentelemetry.trace import SpanKind, Status, StatusCode + from opentelemetry.trace.propagation.tracecontext import ( + TraceContextTextMapPropagator, + ) +except ImportError: + trace = None + + +def operation_name(method, path): + # 路由只保留操作类别；Session / Turn ID、查询参数和 Endpoint 不进入 Span 名称。 + if path == "/readyz": + return "ready" + if path == "/sessions": + return "session.create" + if path.endswith("/events"): + return "turn.events" + if path.endswith("/cancel"): + return "turn.cancel" + if path.endswith("/turns"): + return "turn.start" + if method == "GET" and "/turns/" in path: + return "turn.status" + return "request" + + +@contextmanager +def worker_request_span(method, path, *, activate=True): + if trace is None: + yield {}, None + return + tracer = trace.get_tracer("veadk.worker.client") + # Streaming generators must not hold ambient context while yielding to callers. + factory = tracer.start_as_current_span if activate else tracer.start_span + with factory( + "worker." + operation_name(method, path), + kind=SpanKind.CLIENT, + attributes={"http.request.method": method}, + record_exception=False, + set_status_on_exception=False, + ) as span: + headers = {} + # 只透传 W3C Trace 上下文，避免全局 propagator 自动带出 baggage 或业务正文。 + TraceContextTextMapPropagator().inject( + headers, context=trace.set_span_in_context(span) + ) + try: + yield headers, span + except BaseException as error: + span.set_attribute("error.type", type(error).__name__) + span.set_status(Status(StatusCode.ERROR)) + raise diff --git a/veadk/knowledgebase/knowledgebase.py b/veadk/knowledgebase/knowledgebase.py index 656f1ec20..57473e75e 100644 --- a/veadk/knowledgebase/knowledgebase.py +++ b/veadk/knowledgebase/knowledgebase.py @@ -22,6 +22,7 @@ from veadk.knowledgebase.backends.base_backend import BaseKnowledgebaseBackend from veadk.knowledgebase.entry import KnowledgebaseEntry from veadk.knowledgebase.types import KnowledgebaseProfile +from veadk.tracing.retrieval_tracing import retrieval_span from veadk.utils.logger import get_logger logger = get_logger(__name__) @@ -264,22 +265,30 @@ def add_from_text(self, text: str | list[str], **kwargs) -> bool: def search(self, query: str, top_k: int = 0, **kwargs) -> list[KnowledgebaseEntry]: """Search knowledge from knowledgebase""" - top_k = top_k if top_k != 0 else self.top_k - - _entries = self._backend.search(query=query, top_k=top_k, **kwargs) - - entries = [] - for entry in _entries: - if isinstance(entry, KnowledgebaseEntry): - entries.append(entry) - elif isinstance(entry, str): - entries.append(KnowledgebaseEntry(content=entry)) - else: - logger.error( - f"Unsupported entry type from backend search method: {type(entry)} with {entry}. Expected `KnowledgebaseEntry` or `str`. Skip for this entry." - ) - - return entries + backend = ( + self.backend + if isinstance(self.backend, str) + else type(self.backend).__name__ + ) + with retrieval_span("veadk.knowledge.search", backend=backend) as operation: + top_k = top_k if top_k != 0 else self.top_k + + operation.set_attribute("veadk.retrieval.top_k", top_k) + _entries = self._backend.search(query=query, top_k=top_k, **kwargs) + + entries = [] + for entry in _entries: + if isinstance(entry, KnowledgebaseEntry): + entries.append(entry) + elif isinstance(entry, str): + entries.append(KnowledgebaseEntry(content=entry)) + else: + logger.error( + f"Unsupported entry type from backend search method: {type(entry)} with {entry}. Expected `KnowledgebaseEntry` or `str`. Skip for this entry." + ) + + operation.set_attribute("veadk.retrieval.result_count", len(entries)) + return entries def close(self) -> None: """Release backend resources when the backend exposes a close hook.""" diff --git a/veadk/memory/long_term_memory.py b/veadk/memory/long_term_memory.py index d371588fb..299923596 100644 --- a/veadk/memory/long_term_memory.py +++ b/veadk/memory/long_term_memory.py @@ -34,6 +34,7 @@ from veadk.memory.long_term_memory_backends.base_backend import ( BaseLongTermMemoryBackend, ) +from veadk.tracing.retrieval_tracing import retrieval_span from veadk.utils.logger import get_logger logger = get_logger(__name__) @@ -551,40 +552,48 @@ async def add_session_to_memory( # Added 2 events to long term memory: index=main, user_id=user_123 ``` """ - user_id = session.user_id - save_kwargs = dict(kwargs) - nested_kwargs = save_kwargs.pop("kwargs", None) - if isinstance(nested_kwargs, dict): - save_kwargs.update(nested_kwargs) - auto_save_memory_policy = save_kwargs.pop("auto_save_memory_policy", None) - app_name = self.app_name or getattr(session, "app_name", "") - include_assistant = ( - self.backend == "openviking" - or self._backend.__class__.__name__ == "OpenVikingLTMBackend" - ) - event_strings = self._filter_and_convert_events( - session.events, - include_assistant=include_assistant, - auto_save_memory_policy=auto_save_memory_policy, + backend = ( + self.backend + if isinstance(self.backend, str) + else type(self.backend).__name__ ) + with retrieval_span("veadk.memory.save", backend=backend) as operation: + operation.set_attribute("session.id", session.id) + user_id = session.user_id + save_kwargs = dict(kwargs) + nested_kwargs = save_kwargs.pop("kwargs", None) + if isinstance(nested_kwargs, dict): + save_kwargs.update(nested_kwargs) + auto_save_memory_policy = save_kwargs.pop("auto_save_memory_policy", None) + app_name = self.app_name or getattr(session, "app_name", "") + include_assistant = ( + self.backend == "openviking" + or self._backend.__class__.__name__ == "OpenVikingLTMBackend" + ) + event_strings = self._filter_and_convert_events( + session.events, + include_assistant=include_assistant, + auto_save_memory_policy=auto_save_memory_policy, + ) - logger.info( - f"Adding {len(event_strings)} events to long term memory: index={self.index}" - ) - save_call = { - "user_id": user_id, - "event_strings": event_strings, - "session_id": session.id, - "app_name": app_name, - **save_kwargs, - } - if self._uses_openviking_backend(): - await asyncio.to_thread(self._backend.save_memory, **save_call) - else: - self._backend.save_memory(**save_call) - logger.info( - f"Added {len(event_strings)} events to long term memory: index={self.index}, user_id={user_id}" - ) + operation.set_attribute("veadk.memory.event_count", len(event_strings)) + logger.info( + f"Adding {len(event_strings)} events to long term memory: index={self.index}" + ) + save_call = { + "user_id": user_id, + "event_strings": event_strings, + "session_id": session.id, + "app_name": app_name, + **save_kwargs, + } + if self._uses_openviking_backend(): + await asyncio.to_thread(self._backend.save_memory, **save_call) + else: + self._backend.save_memory(**save_call) + logger.info( + f"Added {len(event_strings)} events to long term memory: index={self.index}, user_id={user_id}" + ) @override async def search_memory( @@ -606,37 +615,46 @@ async def search_memory( An object containing a list of `MemoryEntry` items representing the retrieved memory snippets relevant to the query. """ - logger.info(f"Search memory with query={query}") + backend = ( + self.backend + if isinstance(self.backend, str) + else type(self.backend).__name__ + ) + with retrieval_span("veadk.memory.search", backend=backend) as operation: + operation.set_attribute("veadk.retrieval.top_k", self.top_k) + logger.info(f"Search memory with query={query}") - memory_chunks = [] - try: - search_kwargs = {"app_name": app_name} - search_call = { - "query": query, - "top_k": self.top_k, - "user_id": user_id, - **search_kwargs, - } - if self._uses_openviking_backend(): - memory_chunks = await asyncio.to_thread( - self._backend.search_memory, - **search_call, + memory_chunks = [] + try: + search_kwargs = {"app_name": app_name} + search_call = { + "query": query, + "top_k": self.top_k, + "user_id": user_id, + **search_kwargs, + } + if self._uses_openviking_backend(): + memory_chunks = await asyncio.to_thread( + self._backend.search_memory, + **search_call, + ) + else: + memory_chunks = self._backend.search_memory(**search_call) + except Exception as e: + operation.fail(e) + logger.error( + f"Exception orrcus during memory search: {e}. Return empty memory chunks" ) - else: - memory_chunks = self._backend.search_memory(**search_call) - except Exception as e: - logger.error( - f"Exception orrcus during memory search: {e}. Return empty memory chunks" - ) - memory_events = [] - for memory in memory_chunks: - memory_events.extend(self._convert_memory_chunk_to_entries(memory)) + memory_events = [] + for memory in memory_chunks: + memory_events.extend(self._convert_memory_chunk_to_entries(memory)) - logger.info( - f"Return {len(memory_events)} memory events for query: {query} index={self.index} user_id={user_id}" - ) - return SearchMemoryResponse(memories=memory_events) + logger.info( + f"Return {len(memory_events)} memory events for query: {query} index={self.index} user_id={user_id}" + ) + operation.set_attribute("veadk.retrieval.result_count", len(memory_events)) + return SearchMemoryResponse(memories=memory_events) def _uses_openviking_backend(self) -> bool: return ( diff --git a/veadk/memory/short_term_memory_backends/mysql_backend.py b/veadk/memory/short_term_memory_backends/mysql_backend.py index e3e92d91b..18efdde43 100644 --- a/veadk/memory/short_term_memory_backends/mysql_backend.py +++ b/veadk/memory/short_term_memory_backends/mysql_backend.py @@ -17,7 +17,6 @@ from google.adk.sessions import ( BaseSessionService, - DatabaseSessionService, ) from pydantic import Field from typing_extensions import override @@ -25,6 +24,9 @@ import veadk.config # noqa E401 from veadk.configs.database_configs import MysqlConfig +from veadk.memory.short_term_memory_backends.traced_database_service import ( + TracedDatabaseSessionService as DatabaseSessionService, +) from veadk.memory.short_term_memory_backends.base_backend import ( BaseShortTermMemoryBackend, ) diff --git a/veadk/memory/short_term_memory_backends/postgresql_backend.py b/veadk/memory/short_term_memory_backends/postgresql_backend.py index afc385108..107f6cd9e 100644 --- a/veadk/memory/short_term_memory_backends/postgresql_backend.py +++ b/veadk/memory/short_term_memory_backends/postgresql_backend.py @@ -21,7 +21,6 @@ from google.adk.sessions import ( BaseSessionService, - DatabaseSessionService, ) from pydantic import Field from sqlalchemy import text @@ -30,6 +29,9 @@ import veadk.config # noqa E401 from veadk.configs.database_configs import PostgreSqlConfig +from veadk.memory.short_term_memory_backends.traced_database_service import ( + TracedDatabaseSessionService as DatabaseSessionService, +) from veadk.memory.short_term_memory_backends.base_backend import ( BaseShortTermMemoryBackend, ) diff --git a/veadk/memory/short_term_memory_backends/sqlite_backend.py b/veadk/memory/short_term_memory_backends/sqlite_backend.py index 34e13bd7b..5ab4d1756 100644 --- a/veadk/memory/short_term_memory_backends/sqlite_backend.py +++ b/veadk/memory/short_term_memory_backends/sqlite_backend.py @@ -19,10 +19,12 @@ from google.adk.sessions import ( BaseSessionService, - DatabaseSessionService, ) from typing_extensions import override +from veadk.memory.short_term_memory_backends.traced_database_service import ( + TracedDatabaseSessionService as DatabaseSessionService, +) from veadk.memory.short_term_memory_backends.base_backend import ( BaseShortTermMemoryBackend, ) diff --git a/veadk/memory/short_term_memory_backends/traced_database_service.py b/veadk/memory/short_term_memory_backends/traced_database_service.py new file mode 100644 index 000000000..f78d84290 --- /dev/null +++ b/veadk/memory/short_term_memory_backends/traced_database_service.py @@ -0,0 +1,44 @@ +"""Observe owned database persistence without changing caller-owned services.""" + +from contextlib import suppress + +from google.adk.events import Event +from google.adk.sessions import DatabaseSessionService, Session +from opentelemetry import context, trace +from opentelemetry.trace import Status, StatusCode + + +class TracedDatabaseSessionService(DatabaseSessionService): + async def append_event(self, session: Session, event: Event) -> Event: + # ADK does not persist partial events; do not count them as database writes. + if event.partial: + return await super().append_event(session, event) + span = None + token = None + try: + span = trace.get_tracer("veadk.session").start_span( + "veadk.session.append_event" + ) + span.set_attribute("gen_ai.session.id", session.id) + if event.invocation_id: + span.set_attribute("gen_ai.invocation.id", event.invocation_id) + token = context.attach(trace.set_span_in_context(span)) + except Exception: + # Observability failure must not skip or repeat the storage operation. + pass + try: + return await super().append_event(session, event) + except BaseException as error: + if span is not None: + # Database exceptions may contain credentials or event bodies. + with suppress(Exception): + span.set_attribute("error.type", type(error).__name__) + span.set_status(Status(StatusCode.ERROR)) + raise + finally: + if token is not None: + with suppress(Exception): + context.detach(token) + if span is not None: + with suppress(Exception): + span.end() diff --git a/veadk/runtime/codex/model_tracing.py b/veadk/runtime/codex/model_tracing.py new file mode 100644 index 000000000..136ab5e1e --- /dev/null +++ b/veadk/runtime/codex/model_tracing.py @@ -0,0 +1,53 @@ +# 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. + +"""Codex Shim 的模型请求计时，不收集请求或响应正文。""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from typing import TypeVar + +try: + from opentelemetry import trace + from opentelemetry.trace import Status, StatusCode +except ImportError: + # 与 Shim 的可选观测依赖保持一致，未安装时仍可执行模型请求。 + trace = None + +T = TypeVar("T") + + +async def trace_model_request(call: Callable[[], Awaitable[T]]) -> T: + if trace is None: + return await call() + # 每次库调用独立计时；库内部重试仍包含在该调用中，不冒充底层 HTTP 尝试。 + tracer = trace.get_tracer("veadk.codex.model") + with tracer.start_as_current_span( + "codex.model.request", + attributes={ + "gen_ai.operation.name": "chat", + "veadk.runtime": "codex", + "veadk.model.request.scope": "litellm.aresponses", + }, + record_exception=False, + set_status_on_exception=False, + ) as span: + try: + return await call() + except BaseException as error: + # 异常正文可能含 Endpoint、提示词或凭证；只记录类型，保留取消语义。 + span.set_attribute("error.type", type(error).__name__) + span.set_status(Status(StatusCode.ERROR)) + raise diff --git a/veadk/runtime/codex/proxy.py b/veadk/runtime/codex/proxy.py index b3d3e4e0c..9b85b32b6 100644 --- a/veadk/runtime/codex/proxy.py +++ b/veadk/runtime/codex/proxy.py @@ -55,6 +55,7 @@ TurnRequests, ) from veadk.utils.logger import get_logger +from veadk.runtime.codex.model_tracing import trace_model_request try: # OpenTelemetry is optional; the shim must import without it. from opentelemetry import context as otel_context_api @@ -1045,7 +1046,9 @@ def _fail( resp or {"model": model}, usage_acc ), ) - result = await _call_backend_tolerating_reasoning(call_kwargs) + # Shim 服务协程独立于 Agent；恢复调用上下文，避免模型 Span 成为孤立链路。 + with _otel_scope(turn_context.otel_context): + result = await _call_backend_tolerating_reasoning(call_kwargs) turn_context.requests.check_active() resp = _to_dict(result) _accumulate_usage(usage_acc, resp.get("usage")) @@ -1427,7 +1430,7 @@ async def _call_backend_tolerating_reasoning(call_kwargs: dict[str, Any]) -> Any hard failure on a few models for quieter, worse answers on the rest. """ try: - return await litellm.aresponses(**call_kwargs) + return await trace_model_request(lambda: litellm.aresponses(**call_kwargs)) except Exception as e: # noqa: BLE001 - re-raised unless it is this one case conversation = call_kwargs.get("input") if not _looks_like_reasoning_rejection(e) or not isinstance(conversation, list): @@ -1443,7 +1446,9 @@ async def _call_backend_tolerating_reasoning(call_kwargs: dict[str, Any]) -> Any "codex_backend_reasoning_items_dropped removed=%d", len(conversation) - len(kept), ) - return await litellm.aresponses(**{**call_kwargs, "input": kept}) + return await trace_model_request( + lambda: litellm.aresponses(**{**call_kwargs, "input": kept}) + ) #: ``extra_body`` keys that the Responses transport cannot carry, so they are diff --git a/veadk/tracing/retrieval_tracing.py b/veadk/tracing/retrieval_tracing.py new file mode 100644 index 000000000..815143215 --- /dev/null +++ b/veadk/tracing/retrieval_tracing.py @@ -0,0 +1,74 @@ +# 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. + +"""Record retrieval boundaries through the existing telemetry provider.""" + +from collections.abc import Iterator +from contextlib import contextmanager, suppress + +from opentelemetry import context, trace +from opentelemetry.trace import Status, StatusCode + +_ALLOWED_ATTRIBUTES = frozenset( + { + "veadk.retrieval.backend", + "veadk.retrieval.top_k", + "veadk.retrieval.result_count", + "veadk.memory.event_count", + "session.id", + } +) + + +class RetrievalOperation: + def __init__(self, span: trace.Span | None): + self._span = span + + def set_attribute(self, key: str, value: str | int) -> None: + if self._span is not None and key in _ALLOWED_ATTRIBUTES: + with suppress(Exception): + self._span.set_attribute(key, value) + + def fail(self, error: BaseException) -> None: + if self._span is not None: + # Error messages can contain queries or credentials; retain only type. + with suppress(Exception): + self._span.set_attribute("error.type", type(error).__name__) + self._span.set_status(Status(StatusCode.ERROR)) + + +@contextmanager +def retrieval_span(name: str, *, backend: str) -> Iterator[RetrievalOperation]: + span = None + token = None + try: + span = trace.get_tracer("veadk.retrieval").start_span(name) + token = context.attach(trace.set_span_in_context(span)) + except Exception: + # Telemetry setup must never rerun or prevent the backend operation. + pass + operation = RetrievalOperation(span) + operation.set_attribute("veadk.retrieval.backend", backend) + try: + yield operation + except BaseException as error: + operation.fail(error) + raise + finally: + if token is not None: + with suppress(Exception): + context.detach(token) + if span is not None: + with suppress(Exception): + span.end()