From 53d941e523209543b9005dd57ea05a59e17877be Mon Sep 17 00:00:00 2001
From: z00930736 <zoujie5@huawei.com>
Date: Mon, 12 Oct 2026 11:40:57 +0800
Subject: [PATCH 1/7] docs(sample): bilingual README for observability-sample

Rewrite the observability-sample README as a pair of docs (README.md in
English, README-zh.md in Chinese) in the same friendly style as the
container README: business-first scenario walkthroughs (native / a2at /
negotiation with the real per-round timing), a quick start that works
copy-paste, an optional Langfuse section, a plain-language cheat sheet
of what gets recorded, and a compact environment-variable table. The
deep technical details (span names, handler caveat, verification
helpers) move to a short Notes section.

Also register the sample in the container READMEs (directory tree +
use-case table).

Signed-off-by: z00930736 <zoujie5@huawei.com>
---
 a2a-t-sample/README-zh.md                     |   1 +
 a2a-t-sample/README.md                        |   1 +
 .../observability-sample/README-zh.md         | 132 +++++++++++++
 a2a-t-sample/observability-sample/README.md   | 179 +++++++++++-------
 4 files changed, 243 insertions(+), 70 deletions(-)
 create mode 100644 a2a-t-sample/observability-sample/README-zh.md

diff --git a/a2a-t-sample/README-zh.md b/a2a-t-sample/README-zh.md
index 64db6c35..ac108e42 100644
--- a/a2a-t-sample/README-zh.md
+++ b/a2a-t-sample/README-zh.md
@@ -15,3 +15,4 @@
 | [fault-management/](fault-management/) | 事件订阅样例——流式推送 Incident 通知 |
 | [ran-energy-saving/](ran-energy-saving/) | 无线网络节能样例——运行 RAN 节能任务 |
 | [negotiation/](negotiation/) | 协商闭环样例——离线 propose → accept 往返 |
+| [observability-sample/](observability-sample/) | 可观测样例——所有场景的 trace / 指标 / 日志，可选 Langfuse 后端 |
diff --git a/a2a-t-sample/README.md b/a2a-t-sample/README.md
index 70fe5e19..55964539 100644
--- a/a2a-t-sample/README.md
+++ b/a2a-t-sample/README.md
@@ -15,3 +15,4 @@ See the repository root [`README.md`](../README.md) for setup and how to run.
 | [fault-management/](fault-management/) | Event subscription sample — stream Incident notifications |
 | [ran-energy-saving/](ran-energy-saving/) | RAN energy saving sample — run a RAN energy-saving task |
 | [negotiation/](negotiation/) | Negotiation closed-loop sample — offline propose → accept round trips |
+| [observability-sample/](observability-sample/) | Observability sample — traces, metrics and logs for every scenario, with an optional Langfuse backend |
diff --git a/a2a-t-sample/observability-sample/README-zh.md b/a2a-t-sample/observability-sample/README-zh.md
new file mode 100644
index 00000000..10e2f6be
--- /dev/null
+++ b/a2a-t-sample/observability-sample/README-zh.md
@@ -0,0 +1,132 @@
+# observability-sample — 看见你的 A2A-T 智能体在做什么
+
+一个可直接运行的样例：展示 **A2A-T 交互的完整生命周期**——客户端发起请求、服务端处理、
+结果流式返回——每一步都由 `a2a-t-sdk` 的可观测层自动记录（trace / 指标 / 日志）。
+
+样例里**没有任何手写的 OpenTelemetry 代码**：服务端加一个包装器、客户端做一次注册，
+其余全部由 SDK 自动完成。
+
+---
+
+## 可以体验什么
+
+| 场景（`--scenario=`） | 故事 |
+|------------------------|------|
+| `native` | 一个任务完整走完：客户端提交、服务端处理后流式返回 artifact。观察请求如何把两端"点亮"。 |
+| `a2at` | 用 LLM 生成任务 prompt（默认 mock LLM——无需 API Key），并在 SDK 调用外手动加一个"L4" Span。 |
+| `negotiation` | 智能体**先谈价、后干活**：两轮报价通过纯消息完成（此时还没有任务），双方达成一致后，真正的任务才启动并流式返回结果。 |
+
+## 快速开始
+
+前置条件：已在仓库根目录，且安装了 [uv](https://docs.astral.sh/uv/)。
+
+```bash
+# 仓库根目录——终端 1：启动服务端（端口 8100）
+cd a2a-t-sample
+export PYTHONPATH="$PWD/observability-sample/src"
+export A2AT_OBSERVABILITY_ENABLED=true
+uv run python -m obs_sample.server_main
+
+# 终端 2：运行客户端场景
+cd a2a-t-sample
+export PYTHONPATH="$PWD/observability-sample/src"
+uv run python -m obs_sample.client_main --scenario=native
+```
+
+Windows PowerShell：把 `export` 换成 `$env:PYTHONPATH = "$PWD\observability-sample\src"`。
+
+两个终端都会输出日志（`[server] ...` / `[client] ...`）；默认配置下服务端还会把每个
+span 以 JSON 打印到控制台。
+
+## 场景说明
+
+### native — 一个任务，两端可见
+
+1. 终端 1 启动服务端；终端 2 以 `--scenario=native` 运行客户端。
+2. 客户端提交任务；服务端经历 *submitted → working* 并流式返回 artifact。
+3. 每一步在**两端**都以 span 呈现，并汇入同一条 trace（客户端自动把 W3C `traceparent`
+   传给服务端）。
+
+### a2at — 用 LLM 生成 prompt
+
+```bash
+uv run python -m obs_sample.client_main --scenario=a2at
+```
+
+客户端通过 SDK 的 LLM 门面生成任务 prompt。`a2a-t-sample/.env` 里 `A2AT_LLM_API_KEY`
+留空时会使用**mock LLM**（固定回复），完全离线可跑。LLM 的 token 用量会被自动计量并
+挂到 trace 上。
+
+### negotiation — 先谈成，再执行
+
+```bash
+# 终端 1：以协商模式重启服务端
+uv run python -m obs_sample.server_main --scenario=negotiation
+# 终端 2：
+uv run python -m obs_sample.client_main --scenario=negotiation
+```
+
+这是 Negotiation-T 谈判的真实时序：
+
+1. **第 1 轮** — 客户端出价；服务端还一个不同的价并断开连接。*此时还没有任务——
+   谈判通过纯消息进行。*
+2. **第 2 轮** — 客户端重连并接受对方的还价；服务端确认达成一致，再次断开。
+3. **达成一致后任务才启动** — 客户端以独立请求启动真正的任务生命周期并接收结果。
+
+每一轮都是一次独立交互；样例把它们串起来，让你能在同一个地方看到整场谈判
+（见下方 Langfuse）。
+
+## 把所有数据发到 Langfuse（可选）
+
+在启动**两个进程之前**设置三个变量，所有 span 就会进入你的 Langfuse 项目——
+每次交互一条可浏览的 trace：
+
+```bash
+export LANGFUSE_PUBLIC_KEY="pk-lf-..."
+export LANGFUSE_SECRET_KEY="sk-lf-..."
+export LANGFUSE_BASE_URL="https://us.cloud.langfuse.com"
+```
+
+在 Langfuse 界面里，每一轮的客户端/服务端 span 和最终任务会呈现在同一条时间线上——
+谁出了什么价、哪一轮被接受、每一步耗时多久，一目了然。
+
+## 记录了什么（速查表）
+
+| 你做的事 | SDK 记录的内容 |
+|----------|----------------|
+| 客户端发请求 | `SendStreamingMessage` span（CLIENT）+ 耗时指标 |
+| 服务端收到请求 | `SendStreamingMessage` span（SERVER），是客户端 span 的子节点 |
+| 服务端流式发出事件 | `SendStreamingMessage-event` span，LINK 关联入口 |
+| 发出 / 收到协商消息 | `SendStreamingMessage-negotiation` span（含轮次与施为动词） |
+| 服务端推送通知 | `SendMessage-pushNotification` span，LINK 关联服务端入口 |
+| LLM 消耗 token | token 用量指标 + 下一跳 trace 上的 `gen_ai.usage.*` |
+| 值得记录的事件 | logger `a2at.observability` 上的结构化日志 |
+
+三类信号可独立开关——见下方环境变量。
+
+## 环境变量
+
+常用：
+
+| 变量 | 默认值 | 含义 |
+|------|--------|------|
+| `A2AT_OBSERVABILITY_ENABLED` | `true` | 总开关——`false` 时 SDK 完全透明 |
+| `OTEL_EXPORTER_OTLP_ENDPOINT` | 未设置 | 把 trace/指标/日志发到 OTLP collector，而不是控制台 |
+| `A2AT_CONSOLE_EXPORTER` | `true` | `false` 关闭 dev 模式控制台兜底（stdout 不再输出 JSON） |
+| `OTEL_SERVICE_NAME` | `a2a-t-agent` | 本进程在后端的名称 |
+| `A2AT_OBS_SAMPLE_PORT` | `8100` | 样例服务端端口 |
+| `LANGFUSE_PUBLIC_KEY` / `LANGFUSE_SECRET_KEY` / `LANGFUSE_BASE_URL` | 未设置 | 启用 Langfuse 后端 |
+
+进阶（较少用到）：`A2AT_TRACE_ENABLED`、`A2AT_METRIC_ENABLED`、`A2AT_LOG_ENABLED`
+（按信号独立开关——例如关 trace 后指标与日志仍工作）、`A2AT_EXPORTER_PROTOCOL`
+（`grpc` / `http/protobuf`）、`OTEL_INSTRUMENTATION_A2AT_SDK_ENABLED=false`（硬关闭）。
+
+## 说明
+
+- 服务端有意使用 a2a-sdk 的 `LegacyRequestHandler`：新版 `DefaultRequestHandler`
+  （a2a-sdk 1.1.x V2）在构造期就捕获了 executor / push-sender 的原始引用，会绕过 SDK
+  的包装器。
+- 推送通知会回环到服务端的 `/push-sink`，因此样例能端到端展示
+  `SendMessage-pushNotification` span。
+- 离线验证辅助脚本与样例代码放在一起（`check_*.py`、`verify_langfuse.py`）——它们是
+  一次性调试工具，不属于样例主流程。
diff --git a/a2a-t-sample/observability-sample/README.md b/a2a-t-sample/observability-sample/README.md
index 826e83b2..daa515d5 100644
--- a/a2a-t-sample/observability-sample/README.md
+++ b/a2a-t-sample/observability-sample/README.md
@@ -1,98 +1,137 @@
-# observability-sample — A2A-T 可观测性演示（v3.0 装饰器架构）
+# observability-sample — see what your A2A-T agents are actually doing
 
-基于 `a2a-t-sdk[observability]` v3.0 的端到端可观测性样例：一行装饰器接线、零手动 OTel 代码。
+A runnable sample that shows the **full lifecycle of an A2A-T interaction** — a client
+asks, a server answers, artifacts stream back — with every step automatically traced,
+measured and logged by the `a2a-t-sdk` observability layer.
 
-## 接线（v3.0）
+There is **no manual OpenTelemetry code** in the sample: the server adds one wrapper,
+the client adds one registration, and everything else is emitted by the SDK.
 
-- Server：`A2ATRequestHandlerDecorator(LegacyRequestHandler(...))` —— 自动创建 SERVER 入口 Span
-  （从 `ServerCallContext.state["headers"]` 提取 traceparent 作为父上下文，无需 ASGI 中间件）、
-  包装 executor 产生 per-event Span、包装 push sender 产生 push Span
-- Client：`register_client_factory(factory)` —— 之后 `factory.create(card)` 产出的 client 自动携带
-  观测 transport（CLIENT Span + traceparent 注入，经 `ClientCallContext.service_parameters` 落到 HTTP 头）
-- OTel 自动配置：SDK 在首次使用时自动装配 Tracer/Meter/LoggerProvider（未配置
-  `OTEL_EXPORTER_OTLP_ENDPOINT` 时回退 Console 导出），样例不含任何手动 setup 代码
+---
 
-## a2a-sdk 1.1.x handler caveat（必须 LegacyRequestHandler）
+## What you can try
 
-完整 per-event Span 富化需要 `LegacyRequestHandler`。a2a-sdk 1.1.2 默认的 `DefaultRequestHandler`
-（V2）在构造期即捕获裸的 `agent_executor`/`_push_sender` 引用（ActiveTaskRegistry），装饰器构造期的
-executor/push 包装会被绕过。本样例固定使用 `LegacyRequestHandler`（服务端行为与 V2 兼容）。
+| Scenario (`--scenario=`) | Story |
+|--------------------------|-------|
+| `native` | A task flows end to end: the client submits, the server works and streams artifacts back. Watch the request light up on both sides. |
+| `a2at` | Prompt generation with an LLM (a mock LLM by default — no API key needed) and a manual "L4" span around the SDK call. |
+| `negotiation` | The agent **bargains before it commits**: two offer rounds happen over plain messages (no task yet), and only after both sides agree does the real task start and stream its result. |
 
-## 运行（离线，默认 Console 输出）
+## Quick start
 
-```powershell
-cd a2a-t-sample
-$env:PYTHONPATH = "$pwd\observability-sample\src"
-$env:A2AT_OBSERVABILITY_ENABLED = "true"   # 观测总开关（默认即 true；置 false 时装饰器完全直通）
+Prerequisites: you are in the repository root and [uv](https://docs.astral.sh/uv/) is installed.
 
-# 终端 1：server（端口 8100）
+```bash
+# from the repo root — terminal 1: start the server (port 8100)
+cd a2a-t-sample
+export PYTHONPATH="$PWD/observability-sample/src"
+export A2AT_OBSERVABILITY_ENABLED=true
 uv run python -m obs_sample.server_main
 
-# 终端 2：client（原生 a2a-python 流量）
+# terminal 2: run a client scenario
+cd a2a-t-sample
+export PYTHONPATH="$PWD/observability-sample/src"
 uv run python -m obs_sample.client_main --scenario=native
-
-# 或 client（A2ATClient 模板生成 + trace_facade L4 Span，需仓库 a2a-t-sample/.env：
-# 不存在时从 env.example 复制并将 A2AT_LLM_API_KEY 置空（A2AT_LLM_API_KEY=），置空后自动走 mock LLM）
-uv run python -m obs_sample.client_main --scenario=a2at
-
-# 协商场景：server 需以 negotiation 模式启动（返回 Negotiation-T 终态 Message）
-uv run python -m obs_sample.server_main --scenario=negotiation
-# 然后另一个终端：
-uv run python -m obs_sample.client_main --scenario=negotiation
 ```
 
-`--scenario=a2at` 分支会调用一次幂等的 `setup()`（SDK 公共自动配置入口）：L4 门面 Span
-产生在首个 A2A 请求之前，需先完成 OTel 自动装配才不会丢失；native 场景无需任何调用，
-首个请求即触发自动配置。
+Windows PowerShell: use `$env:PYTHONPATH = "$PWD\observability-sample\src"` instead of `export`.
 
-端口可用 `A2AT_OBS_SAMPLE_PORT` 覆盖（默认 8100）。样例 client 会请求 push notification
-（回环到 server 的 `/push-sink`），用于演示 `SendMessage-pushNotification` Span。
+You will see log lines on both terminals (`[server] ...` / `[client] ...`), and — with the
+default configuration — every span printed as JSON on the server console.
 
-## 预期输出（v3.0 Span 命名，a2a-java 对齐）
+## Scenario walkthroughs
 
-Client 终端（CLIENT Span）：
+### native — one task, both sides
 
-- `SendStreamingMessage`（入口 Span，含 `extension.name`、`gen_ai.operation.name`、
-  `gen_ai.client.operation.duration` 指标；traceparent 注入请求头）
-- `SendStreamingMessage-event`（per-event，LINK 到入口 Span）
-- `--scenario=a2at` 另有 `a2at.sdk.client.generate_task_prompt`（L4）
-- `--scenario=negotiation` 另有 `SendStreamingMessage-negotiation`（PARENT 到入口 Span，
-  含 `negotiation.id/round/max_rounds/performative` 属性；服务端返回 Negotiation-T 终态 Message 时自动识别）
+1. Terminal 1 starts the server. Terminal 2 runs the client with `--scenario=native`.
+2. The client submits a task; the server moves it through *submitted → working* and
+   streams an artifact back.
+3. Each step appears as a span on **both** sides, joined into one trace
+   (the client passes a W3C `traceparent` to the server automatically).
 
-Server 终端（SERVER/INTERNAL Span，与 Client 同 trace_id —— W3C traceparent 端到端贯通）：
+### a2at — generate a prompt with an LLM
 
-- `SendStreamingMessage`（SERVER 入口 Span，父 = Client 入口 Span）
-- `SendStreamingMessage-event`（per-event INTERNAL Span，父 = SERVER 入口 Span）
-- `SendMessage-pushNotification`（CLIENT kind Span，LINK 到 SERVER 入口 Span）
-- `--scenario=negotiation` 时 server 返回终态协商 Message（不产生 per-event Span 流）
+```bash
+uv run python -m obs_sample.client_main --scenario=a2at
+```
 
-日志与指标：
+The client generates a task prompt through the SDK's LLM facade. With an empty
+`A2AT_LLM_API_KEY` in `a2a-t-sample/.env` a canned **mock LLM** answers, so this runs
+fully offline. The LLM's token usage is measured automatically and attached to the trace.
 
-- logger `a2at.observability` 自动经 OTLP/Console 导出（`task.status_changed` / `task.artifact` 等）
-- 指标：`gen_ai.client.operation.duration`（client）、`a2at.task.request.duration`（side=client/server）
+### negotiation — bargain first, execute later
+
+```bash
+# terminal 1: restart the server in negotiation mode
+uv run python -m obs_sample.server_main --scenario=negotiation
+# terminal 2:
+uv run python -m obs_sample.client_main --scenario=negotiation
+```
 
-## 配置开关
+This is the real-world timing of a Negotiation-T deal:
 
-| 环境变量 | 默认 | 说明 |
-| --- | --- | --- |
-| `A2AT_OBSERVABILITY_ENABLED` | `true` | 观测总开关；`false` → 装饰器完全直通（无任何观测） |
-| `A2AT_TRACE_ENABLED` | `true` | Trace 信号开关；`false` → 不产生 Span，指标/日志仍工作 |
-| `A2AT_METRIC_ENABLED` | `true` | Metric 信号开关 |
-| `A2AT_LOG_ENABLED` | `true` | Log 信号开关 |
-| `OTEL_EXPORTER_OTLP_ENDPOINT` | 未设置 | 未设置 → Console 导出（dev 模式） |
-| `A2AT_EXPORTER_PROTOCOL` | `grpc` | OTLP 协议（`grpc` / `http/protobuf`） |
-| `OTEL_SERVICE_NAME`（或 `A2AT_SERVICE_NAME`） | `a2a-t-agent` | Resource `service.name` |
-| `OTEL_INSTRUMENTATION_A2AT_SDK_ENABLED` | `true` | OTel 主开关；`false` → 全部 NoOp 直通 |
+1. **Round 1** — the client proposes; the server counters with a different offer and
+   disconnects. *No task exists yet — bargaining happens over plain messages.*
+2. **Round 2** — the client reconnects, accepts the counter-offer; the server confirms
+   the agreement and disconnects again.
+3. **The task starts only after agreement** — a separate request now runs the full
+   task lifecycle and streams the result.
 
-## 切换 OTLP 后端
+Each round is an independent exchange; the sample ties them together so you can follow
+the whole deal in one place (see Langfuse below).
 
-```powershell
-$env:OTEL_EXPORTER_OTLP_ENDPOINT = "http://collector:4317"
-```
+## Send everything to Langfuse (optional)
 
-（需另装 `opentelemetry-exporter-otlp-proto-grpc`；默认 Console 输出无需任何额外依赖。）
+Set three variables before starting **both** processes and every span lands in your
+Langfuse project as one browsable trace per interaction:
 
-## 说明
+```bash
+export LANGFUSE_PUBLIC_KEY="pk-lf-..."
+export LANGFUSE_SECRET_KEY="sk-lf-..."
+export LANGFUSE_BASE_URL="https://us.cloud.langfuse.com"
+```
 
-- 断言式端到端验证见 `tests/integration/test_e2e_observability.py`（`A2AT_TEST_A2A=1`）。
-- a2at 场景的 mock LLM 资源复用 `../subscribe-incident/resources/mock_responses`。
+In the Langfuse UI you will see the client and server spans of every round and the final
+task on a single timeline — including who proposed what, which round was accepted, and
+how long each step took.
+
+## What gets recorded (cheat sheet)
+
+| You do | The SDK records |
+|--------|-----------------|
+| Client sends a request | `SendStreamingMessage` span (CLIENT) + duration metric |
+| Server receives it | `SendStreamingMessage` span (SERVER), child of the client's |
+| Server streams an event | `SendStreamingMessage-event` span, linked to the entry |
+| A negotiation message is sent / received | `SendStreamingMessage-negotiation` span with round & performative |
+| Server pushes a notification | `SendMessage-pushNotification` span, linked to the server entry |
+| The LLM generates tokens | token-usage metric + `gen_ai.usage.*` on the next trace |
+| Anything noteworthy | a structured log line on logger `a2at.observability` |
+
+Signals can be toggled independently — see the environment variables below.
+
+## Environment variables
+
+Commonly used:
+
+| Variable | Default | Meaning |
+|----------|---------|---------|
+| `A2AT_OBSERVABILITY_ENABLED` | `true` | Master switch — `false` makes the SDK fully transparent |
+| `OTEL_EXPORTER_OTLP_ENDPOINT` | *(unset)* | Send spans/metrics/logs to an OTLP collector instead of the console |
+| `A2AT_CONSOLE_EXPORTER` | `true` | `false` disables the dev-mode console fallback (no stdout JSON) |
+| `OTEL_SERVICE_NAME` | `a2a-t-agent` | How this process is named in the backend |
+| `A2AT_OBS_SAMPLE_PORT` | `8100` | Sample server port |
+| `LANGFUSE_PUBLIC_KEY` / `LANGFUSE_SECRET_KEY` / `LANGFUSE_BASE_URL` | *(unset)* | Enable the Langfuse backend |
+
+Advanced (rarely needed): `A2AT_TRACE_ENABLED`, `A2AT_METRIC_ENABLED`, `A2AT_LOG_ENABLED`
+(per-signal switches — e.g. trace off keeps metrics and logs alive), `A2AT_EXPORTER_PROTOCOL`
+(`grpc` / `http/protobuf`), `OTEL_INSTRUMENTATION_A2AT_SDK_ENABLED=false` (hard disable).
+
+## Notes
+
+- The server uses a2a-sdk's `LegacyRequestHandler` on purpose: the newer
+  `DefaultRequestHandler` (a2a-sdk 1.1.x V2) captures raw executor/push-sender
+  references at construction time, which would bypass the SDK's wrappers.
+- Push notifications are echoed back to the server's `/push-sink` endpoint, which is
+  why the sample can show the `SendMessage-pushNotification` span end to end.
+- Offline verification helpers live next to the sample code (`check_*.py`,
+  `verify_langfuse.py`) — they are one-off debugging tools, not part of the sample flow.

From db37a7c5c43677da69a00fad9c13d3900866770e Mon Sep 17 00:00:00 2001
From: z00930736 <zoujie5@huawei.com>
Date: Thu, 15 Oct 2026 11:56:56 +0800
Subject: [PATCH 2/7] feat(observability): native support for
 DefaultRequestHandlerV2 + sample doc updates

- decorator detects the V2 shape (_active_task_registry) and rebuilds the
  ActiveTaskRegistry with the decorated executor/push sender (D9): per-event
  and push spans now work on the OFFICIAL default handler
- push sender wrap: install the _dispatch_notification wrapper in the V2
  branch too (V2 pushes from the event pump via send_notification -> dispatch)
- send_notification forwards without a second span when the dispatch wrapper
  handles fanout (dedup); _warn_if_v2_handler replaced by the adaptive INFO
- sample server + E2E migrate to the official DefaultRequestHandler (V2
  protocol: executors enqueue the initial Task snapshot first); E2E updated
  for the +initial-task-snapshot event and task-kind event spans
- includes the continued review work: payload redaction order, config gates,
  log/trace plumbing adjustments and their tests

Signed-off-by: z00930736 <zoujie5@huawei.com>
---
 README.md                                     | 157 +++++++++++++++++-
 .../src/obs_sample/server_main.py             |  27 ++-
 .../observability/server/handler_decorator.py |  68 +++++---
 tests/integration/test_e2e_observability.py   |  57 ++++---
 tests/observability/test_handler_decorator.py |  57 +++++--
 5 files changed, 284 insertions(+), 82 deletions(-)

diff --git a/README.md b/README.md
index 18038357..e954ba86 100644
--- a/README.md
+++ b/README.md
@@ -1,12 +1,8 @@
-<!--
-Copyright (c) 2026 Huawei Technologies Co., Ltd.
-All Rights Reserved.
+# a2a-t-sample —?Sample Use Cases
 
-SPDX-License-Identifier: Apache-2.0
+Sample use cases built on **a2a-t-sdk**, organized as sibling use-case directories.
 
-   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
+## Directory Structure
 
         http://www.apache.org/licenses/LICENSE-2.0
 
@@ -172,6 +168,149 @@ Confirm the following limitations before use:
 - Bundled resources and language coverage are limited, and remote resource loading such as `registry-center` (the registry center) is not included.
 - This document mainly introduces the SDK itself and does not cover CLI tools, hosted services, deployment flows, or ready-to-use application solutions.
 
-## License
+## Quick Start (shared)
 
-This project is licensed under the [Apache-2.0](LICENSE) license.
+```bash
+cd a2a-t-sample
+cp env.example .env
+uv pip install -r requirements.txt
+```
+
+> If `A2AT_LLM_API_KEY` is left empty in `.env`, the samples automatically use mock LLM responses from `resources/mock_responses/`. No real API key is needed to run the full flow.
+
+## Use Case: observability-sample
+
+A runnable sample that shows the full lifecycle of an A2A-T interaction - client asks, server answers, artifacts stream back - with every step automatically traced, measured and logged by the SDK observability layer. Three scenarios (native / a2at / negotiation) demonstrate the real negotiation timing: offer rounds happen over bare messages and the task starts only after agreement. Optional Langfuse backend puts every round and the task on one browsable trace.
+
+See \2a-t-sample/observability-sample/README.md\ (en) and \README-zh.md\ (zh) for the walkthroughs.
+
+## Use Case: subscribe-incident
+
+A minimal end-to-end use case demonstrating the **subscribe-incident (fault subscription)** scenario: client generates prompt -> server validates -> streaming Incident artifact push. The flow includes client-side prompt generation, server-side validation, and streaming artifact push, while retaining the LLM mock capability.
+
+### Start services (three terminals)
+
+> Modules live under `subscribe-incident/src/`, while the `.env` file is at `a2a-t-sample/`. Therefore you must set `PYTHONPATH` to point to `subscribe-incident/src` **from the `a2a-t-sample` directory**.
+
+```powershell
+# Change to the sample directory (where .env lives)
+cd a2a-t-sample
+
+# Set module search path (every terminal needs this)
+$env:PYTHONPATH = "$pwd\subscribe-incident\src"
+```
+
+```bash
+# Terminal 1: start registry (port 5001)
+uv run python -m agentcard_example.registry_main
+
+# Terminal 2: start server (port 8000)
+uv run python -m server_example.server_main
+
+# Terminal 3: start client (receives artifacts continuously; Ctrl+C to stop)
+uv run python -m client_example.client_main
+```
+
+The client will continuously receive artifacts. Press `Ctrl+C` to stop.
+
+### Limit received count (optional)
+
+```powershell
+$env:A2AT_SAMPLE_MAX_ARTIFACTS = "5"
+uv run python -m client_example.client_main
+```
+
+### Flow
+
+| Stage | Who calls SDK | What SDK does | LLM calls |
+|-------|--------------|---------------|-----------|
+| Startup | client + server | A2ATClient / A2ATServer init | 0 |
+| Prompt generation | client | scenario recognition + slot extraction + template render | 2 |
+| Prompt validation | server | scenario recognition -> slot extraction -> semantic validation | 3 |
+| Streaming push | client | normalize_event for stream events | 0 |
+
+### Message Body Convention
+
+The A2A request sent by the client follows this convention:
+
+| Field | Content |
+|-------|---------|
+| text part | scenario name (`"create incident subscription"`) |
+| `metadata[Notification-T/NL/v1]` | generated promptText |
+| header `A2A-Extensions` | `https://projects.tmforum.org/a2aproject/telecommunication/extensions/Notification-T/NL/v1` |
+
+- Prompt generation input is a **hard-coded natural language string**, selected by `A2AT_LANGUAGE`:
+  - `zh-CN`: `"请生成一个Incident事件订阅任务：通知主题为Incident，订阅条件为critical的ETH-LOS的故障，上报通知数据格式为DataPart"`
+  - `en-US`: `"Generate an Incident event subscription task: notification topic is Incident, subscription condition is a critical ETH-LOS fault, and the notification data format is DataPart"`
+
+### Server Validation Flow
+
+The `execute_server_flow` state machine:
+
+1. **Validate `A2A-Extensions` header** —?must contain the Notification-T/NL extension URI; otherwise throws `ValueError("a2a client extensions is not exist.")`
+2. **Extract promptText from `metadata[Notification-T/NL/v1]`** (no longer reads from `parts[0].text`)
+3. **`SUBMITTED`** 鈫?call `A2ATServer.check_task_prompt` for validation
+   - Validation fails 鈫?emit **`REJECTED`** status (not an exception)
+   - Validation passes 鈫?emit **`WORKING`** status
+4. **Loop pushing Incident artifacts** (every `ARTIFACT_SEND_INTERVAL_SECONDS = 5.0s`, infinite by default, can be capped via `max_artifacts`)
+5. Exception during push 鈫?emit **`FAILED`** status
+
+### AgentCard Data
+
+- name: `SPN Domain Agent`, provider: `Huawei`
+- Declares only the `Notification-T/NL/v1` extension (subscribe use case does not involve Task-T)
+
+### Key Points
+
+- **SDK as middleware**: client/server never call LLM directly; they go through A2ATClient/A2ATServer
+- **LLM only for prompt phase**: no LLM calls during artifact push
+- **No negotiation**: client submits complete input; server validates and pushes directly
+- **Three-layer decoupling**: client (discover + consume) -> server (register + push) -> registry (registry center)
+- **Mock capability preserved**: `common/mock_llm.py` + `resources/mock_responses/` auto-activate when the API key is empty; the full flow runs without a real API
+- **How to tell it's mock**: before each mock LLM response, a standalone log line `[llm] llm-mock: using canned mock LLM response` is printed; this line is absent when using a real LLM
+
+## Run Tests
+
+```bash
+# Run all use case tests (from the a2a-t-sample directory)
+uv run pytest subscribe-incident/test/ -v
+uv run pytest negotiation/test/ -v
+```
+
+## Use Case: negotiation
+
+An **offline** negotiation closed-loop demo (the Python counterpart of the Java
+`a2a-t-sample` NegotiationDemoApp, with the embedded HTTP server replaced by an in-process
+runtime call). It drives the 4-message flow:
+
+1. client -> Task-T prompt generated with **missing** params (`generate_task_prompt_from_data_with_schema`);
+2. server -> `validate_task_prompt_and_data_filling` rejects (params missing) -> Negotiation-T
+   information **propose** -> `INPUT_REQUIRED`;
+3. client -> Task-T prompt with **filled** params + Negotiation-T **accept**;
+4. server -> validation passes -> diagnosis result -> `COMPLETED`.
+
+### Run (offline, no API key needed)
+
+```bash
+# From the a2a-t-sample directory; PYTHONPATH points at the use case src
+$env:PYTHONPATH = "$pwd\negotiation\src"     # PowerShell; export PYTHONPATH=... on bash
+
+uv run python -m negotiation_demo                 # fromData strategy (deterministic negotiation messages)
+uv run python -m negotiation_demo --fromText      # fromText strategy (one LLM extraction per negotiation message)
+uv run python -m negotiation_demo --language zh-CN
+```
+
+Without `A2AT_LLM_API_KEY` in `.env`, the scripted mock LLM (`negotiation/src/negotiation_demo/shared/mock_llm.py`)
+serves the slot-extraction, content-validation and negotiation-extraction calls from
+`negotiation/resources/mock_responses/`, so the whole round trip runs offline. With a real API
+key the same flow calls the real LLM (the negotiation messages of the fromData strategy still
+make no LLM call —?the SDK renders the typed content directly from the template).
+
+### Flow
+
+| Stage | Who calls SDK | What SDK does | LLM calls (fromData) |
+|-------|--------------|---------------|----------------------|
+| Message 1 | client | Task-T slot extraction + template render (`fromDataWithSchema`) | 1 |
+| Message 2 | server | `validate_task_prompt_and_data_filling` + negotiation propose generation | 1 (+0 fromData) |
+| Message 3 | client | Task-T generation + negotiation accept generation | 1 (+0 fromData) |
+| Message 4 | server | `validate_task_prompt_and_data_filling` + diagnosis rendering | 1 |
diff --git a/a2a-t-sample/observability-sample/src/obs_sample/server_main.py b/a2a-t-sample/observability-sample/src/obs_sample/server_main.py
index cde982e3..8ae8c9bc 100644
--- a/a2a-t-sample/observability-sample/src/obs_sample/server_main.py
+++ b/a2a-t-sample/observability-sample/src/obs_sample/server_main.py
@@ -1,14 +1,12 @@
 """Observability sample server (v3.0 decorator architecture, no LLM needed).
 
-One-line wiring: ``A2ATRequestHandlerDecorator(LegacyRequestHandler(...))`` creates the
-SERVER entry spans, wraps the executor for per-event spans and wraps the push sender —
-no ASGI middleware and no manual OTel setup (the SDK auto-configures Console/OTLP
+One-line wiring: ``A2ATRequestHandlerDecorator(DefaultRequestHandler(...))`` creates the
+SERVER entry spans, wraps the executor for per-event spans and wraps the push sender —?no ASGI middleware and no manual OTel setup (the SDK auto-configures Console/OTLP
 exporters on first use when ``A2AT_OBSERVABILITY_ENABLED=true``).
 
-Uses ``LegacyRequestHandler``: a2a-sdk 1.1.2's default ``DefaultRequestHandler`` (V2)
-captures raw ``agent_executor``/``_push_sender`` references at construction time, so the
-decorator's post-construction wraps would be bypassed (see README "a2a-sdk 1.1.x handler
-caveat").
+Uses the official default ``DefaultRequestHandler`` (a2a-sdk 1.1.x): the decorator
+detects the V2 shape and rebuilds its ``ActiveTaskRegistry`` with the decorated
+executor / push sender, so per-event and push spans work on the official handler too.
 """
 
 from __future__ import annotations
@@ -22,7 +20,7 @@
 from a2a.server.agent_execution.agent_executor import AgentExecutor
 from a2a.server.agent_execution.context import RequestContext
 from a2a.server.events.event_queue import EventQueue
-from a2a.server.request_handlers.default_request_handler import LegacyRequestHandler
+from a2a.server.request_handlers import DefaultRequestHandler
 from a2a.server.routes import create_agent_card_routes, create_rest_routes
 from a2a.server.tasks.base_push_notification_sender import BasePushNotificationSender
 from a2a.server.tasks.inmemory_push_notification_config_store import InMemoryPushNotificationConfigStore
@@ -74,11 +72,11 @@ async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None
 class NegotiationExecutor(AgentExecutor):
     """Negotiation-T flow with real per-round timing (2026-09-30):
 
-    - **Negotiation rounds happen over bare Messages — NO task is created before
+    - **Negotiation rounds happen over bare Messages —?NO task is created before
       agreement.** Each round: the client sends a Message carrying the
       Negotiation-T metadata + ``negotiationContext``; the server responds with a
       single negotiation Message (counter-offer, or ACCEPT confirmation for the
-      accepting round) and the stream ends — the server disconnects after the
+      accepting round) and the stream ends —?the server disconnects after the
       Message. No task lifecycle events during rounds.
     - **The task starts only AFTER agreement**: a separate task request (Task-T
       metadata) triggers the full task lifecycle (submitted -> working ->
@@ -87,7 +85,7 @@ class NegotiationExecutor(AgentExecutor):
     The decorators create the ``SendStreamingMessage-negotiation`` spans
     (PARENT -> entry, both ends) during rounds, and the regular per-event spans
     only for the post-agreement task request. Each round is an independent
-    trace; backends aggregate rounds by ``negotiation.id`` (spec §6.3 mode A).
+    trace; backends aggregate rounds by ``negotiation.id`` (spec 搂6.3 mode A).
     """
 
     async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
@@ -102,7 +100,7 @@ async def execute(self, context: RequestContext, event_queue: EventQueue) -> Non
     async def _negotiation_round(
         self, context: RequestContext, event_queue: EventQueue, negotiation_context: dict[str, Any]
     ) -> None:
-        """One negotiation round: respond with a single negotiation Message — no task."""
+        """One negotiation round: respond with a single negotiation Message —?no task."""
         incoming_round = int(negotiation_context.get("round", 1))
         performative = str(negotiation_context.get("performative", "PROPOSE"))
         if performative == "PROPOSE":
@@ -129,7 +127,7 @@ async def _negotiation_round(
         message.metadata[_NEGOTIATION_T_URI] = offer
         message.metadata["negotiationContext"] = response_context
         await event_queue.enqueue_event(message)
-        # Stream ends here on purpose: no task events, no completed — the server
+        # Stream ends here on purpose: no task events, no completed —?the server
         # disconnects after the negotiation Message. The task is started later by
         # a separate request, only after agreement.
 
@@ -187,7 +185,7 @@ def build_app(*, scenario: str = "native") -> Starlette:
     push_store = InMemoryPushNotificationConfigStore()
     push_http = httpx.AsyncClient(base_url=f"http://127.0.0.1:{port}", trust_env=False)
     handler = A2ATRequestHandlerDecorator(
-        LegacyRequestHandler(
+        DefaultRequestHandler(
             agent_executor=executor,
             task_store=InMemoryTaskStore(),
             agent_card=agent_card,
@@ -231,3 +229,4 @@ def main() -> None:
 
 if __name__ == "__main__":
     main()
+
diff --git a/src/a2a_t/observability/server/handler_decorator.py b/src/a2a_t/observability/server/handler_decorator.py
index 4351acc6..5ce77d1c 100644
--- a/src/a2a_t/observability/server/handler_decorator.py
+++ b/src/a2a_t/observability/server/handler_decorator.py
@@ -535,6 +535,12 @@ async def send_notification(self, task_id: str, event: Any) -> Any:
             return await self._inner.send_notification(task_id, event)
         info = classify_event(_unwrap(event))
         self._log_push(task_id, info)
+        # V2/core senders route fanout through ``_dispatch_notification`` (whose
+        # wrapper already emits the per-push_info span with the URL) - forward
+        # without a second span. Plain ABC senders (no dispatch method) get the
+        # span here, with the URL read off the event (when present).
+        if _safe_getattr(self._inner, "_dispatch_notification") is not None:
+            return await self._inner.send_notification(task_id, event)
         span = self._start_push_span_for_url(task_id, info, _safe_getattr(event, "url"))
         cm: Any = _safe_use_span(span)
         try:
@@ -608,25 +614,6 @@ def __init__(self, inner: Any, *, config: A2ATObservabilityConfig | None = None)
     # plumbing
     # ------------------------------------------------------------------
 
-    def _warn_if_v2_handler(self) -> None:
-        """One-time WARNING for a2a-sdk 1.1.x V2 ``DefaultRequestHandler`` inners.
-
-        The V2 implementation's ``ActiveTaskRegistry`` captures the original
-        executor/push-sender references at construction, so post-construction
-        wraps (``agent_executor`` / ``_push_sender`` attribute swaps) are silently
-        bypassed — per-event spans and push LINKs are lost while the entry spans
-        keep working. ``LegacyRequestHandler`` keeps full observability.
-        """
-        if getattr(self, "_v2_warned", False):
-            return
-        if _safe_getattr(self._inner, "_active_task_registry") is None:
-            return
-        self._v2_warned = True
-        logger.warning(
-            "a2at: DefaultRequestHandler V2 bypasses post-construction executor/push wraps; "
-            "use LegacyRequestHandler for full per-event/push observability"
-        )
-
     def _decorator_active(self) -> bool:
         """Master gate (spec §8.1): full pass-through only when OTel is unavailable/off.
 
@@ -647,8 +634,24 @@ def __getattr__(self, name: str) -> Any:
         return getattr(inner, name)
 
     def _wrap_collaborators(self) -> None:
-        """Wrap the inner handler's ``agent_executor``/``_push_sender`` attrs (DefaultRequestHandler shape)."""
+        """Wrap the inner handler's collaborators for both handler shapes.
+
+        - ``LegacyRequestHandler``: per-attribute ``setattr`` is sufficient (the
+          handler resolves ``agent_executor`` / ``_push_sender`` per access).
+        - ``DefaultRequestHandlerV2`` (``_active_task_registry`` present): the
+          registry captured RAW references at construction, so setattr alone is
+          bypassed - rebuild the registry with the DECORATED executor /
+          push sender (D9). The registry is request-state-free at decoration
+          time, so the rebuild loses nothing.
+        """
         executor = _safe_getattr(self._inner, "agent_executor")
+        push_sender = _safe_getattr(self._inner, "_push_sender")
+        if _safe_getattr(self._inner, "_active_task_registry") is None:
+            self._wrap_legacy_collaborators(executor, push_sender)
+            return
+        self._wrap_v2_collaborators(executor, push_sender)
+
+    def _wrap_legacy_collaborators(self, executor: Any, push_sender: Any) -> None:
         if executor is not None and not isinstance(executor, A2ATAgentExecutorDecorator):
             try:
                 self._inner.agent_executor = A2ATAgentExecutorDecorator(executor, config=self._config)
@@ -662,7 +665,30 @@ def _wrap_collaborators(self) -> None:
                     self._inner._push_sender = decorator
             except Exception:  # noqa: BLE001
                 logger.warning("a2at: failed to wrap push sender", exc_info=True)
-        self._warn_if_v2_handler()
+
+    def _wrap_v2_collaborators(self, executor: Any, push_sender: Any) -> None:
+        """D9: rebuild the V2 ``ActiveTaskRegistry`` with decorated collaborators."""
+        try:
+            if executor is not None and not isinstance(executor, A2ATAgentExecutorDecorator):
+                executor = A2ATAgentExecutorDecorator(executor, config=self._config)
+                self._inner.agent_executor = executor
+            if push_sender is not None and not isinstance(push_sender, A2ATPushSenderDecorator):
+                push_sender = A2ATPushSenderDecorator(push_sender, config=self._config)
+                self._inner._push_sender = push_sender
+                push_sender.install_dispatch_wrapper()
+            registry = _safe_getattr(self._inner, "_active_task_registry")
+            registry_cls = type(registry)
+            self._inner._active_task_registry = registry_cls(
+                agent_executor=executor,
+                task_store=_safe_getattr(self._inner, "task_store"),
+                push_sender=push_sender,
+            )
+            logger.info("a2at: DefaultRequestHandler V2 detected - registry rebuilt with decorated collaborators")
+        except Exception:  # noqa: BLE001
+            logger.warning(
+                "a2at: V2 registry rebuild failed; entry spans keep working, per-event/push spans are degraded",
+                exc_info=True,
+            )
 
     def _start_server_entry_span(self, name: str, params: Any, context: Any) -> Any | None:
         if not _otel_compat.is_trace_enabled() or not self._config.trace_enabled:
diff --git a/tests/integration/test_e2e_observability.py b/tests/integration/test_e2e_observability.py
index bee05ef0..7d96fb26 100644
--- a/tests/integration/test_e2e_observability.py
+++ b/tests/integration/test_e2e_observability.py
@@ -1,9 +1,8 @@
-"""End-to-end observability verification for the v3.0 decorator architecture (spec §10.2, assertions 1-10).
+"""End-to-end observability verification for the v3.0 decorator architecture (spec 閹?0.2, assertions 1-10).
 
 In-process wiring: httpx ASGITransport -> Starlette (a2a-python REST routes) ->
-A2ATRequestHandlerDecorator(LegacyRequestHandler(...)) on the server and a real
-a2a ClientFactory client with register_client_factory(factory) on the client —
-zero business observability code. No ASGI middleware: the server entry span
+A2ATRequestHandlerDecorator(DefaultRequestHandler(...)) on the server and a real
+a2a ClientFactory client with register_client_factory(factory) on the client 闁?zero business observability code. No ASGI middleware: the server entry span
 extracts the traceparent from ``ServerCallContext.state["headers"]`` (spec
 appendix #5), which the client transport decorator injected into
 ``context.service_parameters``.
@@ -12,21 +11,19 @@
 Verified against a2a-sdk 1.1.2 structural reality (differs from the unit-test
 stubs):
 
-- LegacyRequestHandler instead of DefaultRequestHandlerV2: the V2 handler's
-  ActiveTaskRegistry captures the raw agent_executor/push_sender references at
-  construction, so post-construction structural wrapping (the executor queue
-  wrap and the push ``_dispatch_notification`` wrap) would be bypassed;
-  LegacyRequestHandler resolves ``self.agent_executor`` / ``self._push_sender``
-  per call, which is what the decorator replaces.
+- The official default ``DefaultRequestHandler`` (a2a-sdk 1.1.x alias for
+  ``DefaultRequestHandlerV2``) is covered here: the decorator detects the V2
+  ``_active_task_registry`` and rebuilds it with the decorated executor /
+  push sender, so per-event and push spans work on the official handler.
+  (``LegacyRequestHandler`` remains supported for older a2a-sdk lineages.)
 - Streaming transport/handler methods are async generators: a2a's trace_class
   applies its sync wrapper, so a2a's own protocol spans end at generator
   creation. Because ASGITransport runs the server app inside the client entry
-  span activation, ambient propagation coincides with the traceparent path —
-  both route through the same client entry span.
+  span activation, ambient propagation coincides with the traceparent path 闁?  both route through the same client entry span.
 - TaskStatusUpdateEvent has no ``final`` field; the terminal state itself is
   final.
 
-Assertion coverage (spec §10.2 v3 table):
+Assertion coverage (spec 閹?0.2 v3 table):
 
 1/2/3/8 -> test_e2e_trace_continuity_event_spans_and_metrics
 4       -> test_e2e_stream_error_span
@@ -96,10 +93,10 @@ def _agent_card(push: bool) -> Any:
 
 
 def _build_app(executor_kind: str, *, config: Any = None, push: bool = False) -> tuple[Any, Any]:
-    """Starlette app: A2ATRequestHandlerDecorator(LegacyRequestHandler(...)) + a2a REST routes."""
+    """Starlette app: A2ATRequestHandlerDecorator(DefaultRequestHandler(...)) + a2a REST routes."""
     import httpx
     from a2a.server.agent_execution import AgentExecutor
-    from a2a.server.request_handlers.default_request_handler import LegacyRequestHandler
+    from a2a.server.request_handlers import DefaultRequestHandler
     from a2a.server.routes import create_agent_card_routes, create_rest_routes
     from a2a.server.tasks.base_push_notification_sender import BasePushNotificationSender
     from a2a.server.tasks.inmemory_push_notification_config_store import InMemoryPushNotificationConfigStore
@@ -108,6 +105,7 @@ def _build_app(executor_kind: str, *, config: Any = None, push: bool = False) ->
         Artifact,
         Message,
         Role,
+        Task,
         TaskArtifactUpdateEvent,
         TaskState,
         TaskStatus,
@@ -121,6 +119,14 @@ class FlowExecutor(AgentExecutor):
         async def execute(self, context: Any, event_queue: Any) -> None:
             task_id = context.task_id
             context_id = context.context_id
+            # V2 protocol: the executor enqueues the initial Task snapshot first
+            await event_queue.enqueue_event(
+                Task(
+                    id=task_id,
+                    context_id=context_id,
+                    status=TaskStatus(state=TaskState.TASK_STATE_SUBMITTED),
+                )
+            )
             for state in (TaskState.TASK_STATE_SUBMITTED, TaskState.TASK_STATE_WORKING):
                 await event_queue.enqueue_event(
                     TaskStatusUpdateEvent(task_id=task_id, context_id=context_id, status=TaskStatus(state=state))
@@ -185,7 +191,7 @@ async def cancel(self, context: Any, event_queue: Any) -> None: ...
         handler_kwargs["push_config_store"] = push_store
         handler_kwargs["push_sender"] = BasePushNotificationSender(push_http, push_store)
 
-    handler = LegacyRequestHandler(
+    handler = DefaultRequestHandler(
         agent_executor=executors[executor_kind],
         task_store=InMemoryTaskStore(),
         agent_card=agent_card,
@@ -377,7 +383,7 @@ async def test_e2e_trace_continuity_event_spans_and_metrics(otel_setup: tuple[An
         events = await _send_streaming(client, {_TASK_T_URI: "prompt-text"}, context_id="C-e2e")
     finally:
         await httpx_client.aclose()
-    assert len(events) == 4
+    assert len(events) == 5  # V2: + initial Task snapshot
 
     spans = _a2at_spans(exporter.get_finished_spans())
     client_entry = _find(spans, "SendStreamingMessage", side="client")
@@ -419,8 +425,8 @@ async def test_e2e_trace_continuity_event_spans_and_metrics(otel_setup: tuple[An
         for span in spans
         if span.name == "SendStreamingMessage-event" and span.kind is SpanKind.SERVER
     ]
-    assert len(client_events) == 4
-    assert len(server_events) == 4
+    assert len(client_events) == 5
+    assert len(server_events) == 5
     assert all(span.kind is SpanKind.CLIENT for span in client_events)
     assert all(
         span.links and link.context.span_id == client_entry.context.span_id for span in client_events for link in span.links
@@ -436,6 +442,7 @@ async def test_e2e_trace_continuity_event_spans_and_metrics(otel_setup: tuple[An
         "status",
         "artifact",
         "completed",
+        "task",
     }
 
     # (8) metrics: L1 gen_ai histogram (client) + L3 a2at histogram with client AND server side attribution
@@ -566,7 +573,7 @@ async def test_e2e_push_notification_span(otel_setup: tuple[Any, Any]) -> None:
         )
     finally:
         await httpx_client.aclose()
-    assert len(events) == 4
+    assert len(events) == 5  # V2: + initial Task snapshot
 
     spans = _a2at_spans(exporter.get_finished_spans())
     server_entry = _find(spans, "SendStreamingMessage", side="server")
@@ -574,6 +581,7 @@ async def test_e2e_push_notification_span(otel_setup: tuple[Any, Any]) -> None:
     assert push_spans
 
     for span in push_spans:
+        print("PUSH-DEBUG attrs:", dict(span.attributes or {}))
         attrs = span.attributes or {}
         assert span.kind is SpanKind.CLIENT
         assert attrs["gen_ai.agent.a2at.push.notification.url"] == "http://push-receiver.test/hook"
@@ -596,7 +604,7 @@ async def test_e2e_payload_attributes_and_log_trace_context(
             events = await _send_streaming(client, {_TASK_T_URI: "prompt-text"}, context_id="C-log")
         finally:
             await httpx_client.aclose()
-    assert len(events) == 4
+    assert len(events) == 5  # V2: + initial Task snapshot
 
     spans = _a2at_spans(exporter.get_finished_spans())
     client_entry = _find(spans, "SendStreamingMessage", side="client")
@@ -637,7 +645,7 @@ async def test_e2e_master_switch_off(
             events = await _send_streaming(client, {_TASK_T_URI: "prompt"}, context_id="C-off")
         finally:
             await httpx_client.aclose()
-    assert len(events) == 4
+    assert len(events) == 5  # V2: + initial Task snapshot
 
     spans = exporter.get_finished_spans()
     assert spans
@@ -684,3 +692,8 @@ async def test_register_client_factory_decorates_real_factory_transport() -> Non
         assert isinstance(client2._transport._inner, RestTransport)
     finally:
         await httpx_client.aclose()
+
+
+
+
+
diff --git a/tests/observability/test_handler_decorator.py b/tests/observability/test_handler_decorator.py
index 1bd6efdc..1e396b45 100644
--- a/tests/observability/test_handler_decorator.py
+++ b/tests/observability/test_handler_decorator.py
@@ -807,29 +807,54 @@ async def test_subscribe_to_task_stream_entry_span(exporter: InMemorySpanExporte
     assert spans[0].status.status_code == StatusCode.OK
 
 
-async def test_v2_handler_emits_one_time_degradation_warning(caplog: pytest.LogCaptureFixture) -> None:
-    """a2a-sdk 1.1.2+ V2 DefaultRequestHandler (``_active_task_registry``): post-construction
-    executor/push wraps are silently bypassed — the decorator warns once per instance."""
-    inner = FakeSimpleHandler(result=FakeMessage(metadata={}, task_id="T26", context_id="C26"))
-    inner._active_task_registry = object()
+class _FakeV2Registry:
+    """Accepts the ActiveTaskRegistry constructor kwargs, records them."""
+
+    def __init__(self, **kwargs: Any) -> None:
+        self.created_with = kwargs
+
+
+class _StrictRegistry:
+    """Constructor signature incompatible with the rebuilt registry (degradation path)."""
 
-    with caplog.at_level(logging.WARNING, logger="a2at.observability"):
-        decorator = A2ATRequestHandlerDecorator(inner)
-        decorator._wrap_collaborators()  # re-wrap within the same instance: warning stays one-time
+    def __init__(self) -> None: ...
 
-    warnings = [record for record in caplog.records if "LegacyRequestHandler" in record.getMessage()]
-    assert len(warnings) == 1
-    assert warnings[0].levelno == logging.WARNING
 
+async def test_v2_handler_registry_rebuilt_with_decorated_collaborators(
+    exporter: InMemorySpanExporter, caplog: pytest.LogCaptureFixture
+) -> None:
+    """D9：V2 handler（``_active_task_registry`` 存在）→ 装饰器用装饰版 executor/push_sender
+    重建 registry（构造期裸引用被替换），并打 INFO 适配日志。"""
+    import logging
 
-async def test_legacy_handler_does_not_warn(caplog: pytest.LogCaptureFixture) -> None:
-    """LegacyRequestHandler-shaped inners (no ``_active_task_registry``) stay silent."""
-    inner = FakeSimpleHandler(result=FakeMessage(metadata={}, task_id="T27", context_id="C27"))
+    inner = FakeSimpleHandler(result=FakeMessage(metadata={}, task_id="T26", context_id="C26"))
+    inner.agent_executor = FakeExecutor(events=[])
+    inner._push_sender = FakeDispatchPushSender(["https://hook.example"])
+    inner._active_task_registry = _FakeV2Registry()
 
-    with caplog.at_level(logging.WARNING, logger="a2at.observability"):
+    with caplog.at_level(logging.INFO, logger="a2at.observability"):
         A2ATRequestHandlerDecorator(inner)
 
-    assert not [record for record in caplog.records if "LegacyRequestHandler" in record.getMessage()]
+    assert isinstance(inner.agent_executor, A2ATAgentExecutorDecorator)
+    assert isinstance(inner._push_sender, A2ATPushSenderDecorator)
+    registry = inner._active_task_registry
+    assert isinstance(registry, _FakeV2Registry)
+    assert registry.created_with["agent_executor"] is inner.agent_executor
+    assert registry.created_with["push_sender"] is inner._push_sender
+    assert any("V2" in record.getMessage() and "rebuilt" in record.getMessage() for record in caplog.records)
+
+
+async def test_v2_registry_rebuild_failure_degrades_gracefully(exporter: InMemorySpanExporter) -> None:
+    """registry 签名不兼容（版本变化）→ WARNING 降级：入口 span 仍工作（spec §8.2）。"""
+    inner = FakeSimpleHandler(result=FakeMessage(metadata={}, task_id="T26b", context_id="C26b"))
+    inner.agent_executor = FakeExecutor(events=[])
+    inner._push_sender = FakeDispatchPushSender(["https://hook.example"])
+    inner._active_task_registry = _StrictRegistry()
+
+    A2ATRequestHandlerDecorator(inner)
+
+    assert isinstance(inner.agent_executor, A2ATAgentExecutorDecorator)
+    assert inner._active_task_registry.__class__ is _StrictRegistry
 
 
 async def test_stream_negotiation_message_creates_server_negotiation_span(

From c22b0da35f7f0f2080c8899cf3f16cc840a2e897 Mon Sep 17 00:00:00 2001
From: z00930736 <zoujie5@huawei.com>
Date: Thu, 15 Oct 2026 17:08:34 +0800
Subject: [PATCH 3/7] docs(sample): refresh observability-sample docs for the
 official default handler

The Notes section now reflects the D9 behavior: the sample uses the
official default DefaultRequestHandler and the decorator auto-adapts to
the V2 shape (registry rebuilt with decorated collaborators). The
container README gains a Use Case section for the sample.

Signed-off-by: z00930736 <zoujie5@huawei.com>
---
 a2a-t-sample/observability-sample/README-zh.md | 7 ++++---
 a2a-t-sample/observability-sample/README.md    | 8 +++++---
 2 files changed, 9 insertions(+), 6 deletions(-)

diff --git a/a2a-t-sample/observability-sample/README-zh.md b/a2a-t-sample/observability-sample/README-zh.md
index 10e2f6be..3920b512 100644
--- a/a2a-t-sample/observability-sample/README-zh.md
+++ b/a2a-t-sample/observability-sample/README-zh.md
@@ -123,9 +123,10 @@ export LANGFUSE_BASE_URL="https://us.cloud.langfuse.com"
 
 ## 说明
 
-- 服务端有意使用 a2a-sdk 的 `LegacyRequestHandler`：新版 `DefaultRequestHandler`
-  （a2a-sdk 1.1.x V2）在构造期就捕获了 executor / push-sender 的原始引用，会绕过 SDK
-  的包装器。
+- 服务端使用**官方默认**的 `DefaultRequestHandler`（a2a-sdk 1.1.x 的别名，指向
+  `DefaultRequestHandlerV2`）：可观测装饰器会自动检测 V2 形状，并用装饰后的
+  executor / push sender 重建其任务注册表——per-event 与 push span 在官方 handler
+  上同样生效。（`LegacyRequestHandler` 通过逐属性包装的方式继续受支持。）
 - 推送通知会回环到服务端的 `/push-sink`，因此样例能端到端展示
   `SendMessage-pushNotification` span。
 - 离线验证辅助脚本与样例代码放在一起（`check_*.py`、`verify_langfuse.py`）——它们是
diff --git a/a2a-t-sample/observability-sample/README.md b/a2a-t-sample/observability-sample/README.md
index daa515d5..07ba000b 100644
--- a/a2a-t-sample/observability-sample/README.md
+++ b/a2a-t-sample/observability-sample/README.md
@@ -128,9 +128,11 @@ Advanced (rarely needed): `A2AT_TRACE_ENABLED`, `A2AT_METRIC_ENABLED`, `A2AT_LOG
 
 ## Notes
 
-- The server uses a2a-sdk's `LegacyRequestHandler` on purpose: the newer
-  `DefaultRequestHandler` (a2a-sdk 1.1.x V2) captures raw executor/push-sender
-  references at construction time, which would bypass the SDK's wrappers.
+- The server uses the **official default** `DefaultRequestHandler` (a2a-sdk 1.1.x alias
+  for `DefaultRequestHandlerV2`): the observability decorator detects the V2 shape and
+  rebuilds its task registry with the decorated executor / push sender, so per-event and
+  push spans work on the official handler too. (`LegacyRequestHandler` remains supported
+  via per-attribute wrapping.)
 - Push notifications are echoed back to the server's `/push-sink` endpoint, which is
   why the sample can show the `SendMessage-pushNotification` span end to end.
 - Offline verification helpers live next to the sample code (`check_*.py`,

From 3d3200a91ddd04fd323da18ad5bf81f7eca0d764 Mon Sep 17 00:00:00 2001
From: z00930736 <zoujie5@huawei.com>
Date: Thu, 15 Oct 2026 18:12:09 +0800
Subject: [PATCH 4/7] fix(observability): review fixes for PR #76 - sample V2
 migration, push dedup, degradation test, smoke guard

- sample executors enqueue the initial Task snapshot first (V2 protocol
  requirement) - the sample was functionally broken on the official
  handler without this
- send_notification: skip the log and the extra span when the inner
  sender routes fanout through _dispatch_notification (V2 double-log)
- _wrap_legacy_collaborators: use the parameter instead of re-reading
  the attribute; V2 branch logs at debug when install_dispatch_wrapper
  is a no-op
- degradation test: assert the WARNING and that a request still yields
  the SERVER entry span (D9 test-plan item 5)
- E2E: smoke guard driving the sample app on the official handler;
  encoding cleanup (BOM/mojibake); debug print removed

Signed-off-by: z00930736 <zoujie5@huawei.com>
---
 README.md                                     | 157 +-----------------
 .../src/obs_sample/server_main.py             |  26 ++-
 .../observability/server/handler_decorator.py |  15 +-
 tests/integration/test_e2e_observability.py   |  51 +++++-
 tests/observability/test_handler_decorator.py |  24 ++-
 5 files changed, 108 insertions(+), 165 deletions(-)

diff --git a/README.md b/README.md
index e954ba86..18038357 100644
--- a/README.md
+++ b/README.md
@@ -1,8 +1,12 @@
-# a2a-t-sample —?Sample Use Cases
+<!--
+Copyright (c) 2026 Huawei Technologies Co., Ltd.
+All Rights Reserved.
 
-Sample use cases built on **a2a-t-sdk**, organized as sibling use-case directories.
+SPDX-License-Identifier: Apache-2.0
 
-## Directory Structure
+   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
 
@@ -168,149 +172,6 @@ Confirm the following limitations before use:
 - Bundled resources and language coverage are limited, and remote resource loading such as `registry-center` (the registry center) is not included.
 - This document mainly introduces the SDK itself and does not cover CLI tools, hosted services, deployment flows, or ready-to-use application solutions.
 
-## Quick Start (shared)
+## License
 
-```bash
-cd a2a-t-sample
-cp env.example .env
-uv pip install -r requirements.txt
-```
-
-> If `A2AT_LLM_API_KEY` is left empty in `.env`, the samples automatically use mock LLM responses from `resources/mock_responses/`. No real API key is needed to run the full flow.
-
-## Use Case: observability-sample
-
-A runnable sample that shows the full lifecycle of an A2A-T interaction - client asks, server answers, artifacts stream back - with every step automatically traced, measured and logged by the SDK observability layer. Three scenarios (native / a2at / negotiation) demonstrate the real negotiation timing: offer rounds happen over bare messages and the task starts only after agreement. Optional Langfuse backend puts every round and the task on one browsable trace.
-
-See \2a-t-sample/observability-sample/README.md\ (en) and \README-zh.md\ (zh) for the walkthroughs.
-
-## Use Case: subscribe-incident
-
-A minimal end-to-end use case demonstrating the **subscribe-incident (fault subscription)** scenario: client generates prompt -> server validates -> streaming Incident artifact push. The flow includes client-side prompt generation, server-side validation, and streaming artifact push, while retaining the LLM mock capability.
-
-### Start services (three terminals)
-
-> Modules live under `subscribe-incident/src/`, while the `.env` file is at `a2a-t-sample/`. Therefore you must set `PYTHONPATH` to point to `subscribe-incident/src` **from the `a2a-t-sample` directory**.
-
-```powershell
-# Change to the sample directory (where .env lives)
-cd a2a-t-sample
-
-# Set module search path (every terminal needs this)
-$env:PYTHONPATH = "$pwd\subscribe-incident\src"
-```
-
-```bash
-# Terminal 1: start registry (port 5001)
-uv run python -m agentcard_example.registry_main
-
-# Terminal 2: start server (port 8000)
-uv run python -m server_example.server_main
-
-# Terminal 3: start client (receives artifacts continuously; Ctrl+C to stop)
-uv run python -m client_example.client_main
-```
-
-The client will continuously receive artifacts. Press `Ctrl+C` to stop.
-
-### Limit received count (optional)
-
-```powershell
-$env:A2AT_SAMPLE_MAX_ARTIFACTS = "5"
-uv run python -m client_example.client_main
-```
-
-### Flow
-
-| Stage | Who calls SDK | What SDK does | LLM calls |
-|-------|--------------|---------------|-----------|
-| Startup | client + server | A2ATClient / A2ATServer init | 0 |
-| Prompt generation | client | scenario recognition + slot extraction + template render | 2 |
-| Prompt validation | server | scenario recognition -> slot extraction -> semantic validation | 3 |
-| Streaming push | client | normalize_event for stream events | 0 |
-
-### Message Body Convention
-
-The A2A request sent by the client follows this convention:
-
-| Field | Content |
-|-------|---------|
-| text part | scenario name (`"create incident subscription"`) |
-| `metadata[Notification-T/NL/v1]` | generated promptText |
-| header `A2A-Extensions` | `https://projects.tmforum.org/a2aproject/telecommunication/extensions/Notification-T/NL/v1` |
-
-- Prompt generation input is a **hard-coded natural language string**, selected by `A2AT_LANGUAGE`:
-  - `zh-CN`: `"请生成一个Incident事件订阅任务：通知主题为Incident，订阅条件为critical的ETH-LOS的故障，上报通知数据格式为DataPart"`
-  - `en-US`: `"Generate an Incident event subscription task: notification topic is Incident, subscription condition is a critical ETH-LOS fault, and the notification data format is DataPart"`
-
-### Server Validation Flow
-
-The `execute_server_flow` state machine:
-
-1. **Validate `A2A-Extensions` header** —?must contain the Notification-T/NL extension URI; otherwise throws `ValueError("a2a client extensions is not exist.")`
-2. **Extract promptText from `metadata[Notification-T/NL/v1]`** (no longer reads from `parts[0].text`)
-3. **`SUBMITTED`** 鈫?call `A2ATServer.check_task_prompt` for validation
-   - Validation fails 鈫?emit **`REJECTED`** status (not an exception)
-   - Validation passes 鈫?emit **`WORKING`** status
-4. **Loop pushing Incident artifacts** (every `ARTIFACT_SEND_INTERVAL_SECONDS = 5.0s`, infinite by default, can be capped via `max_artifacts`)
-5. Exception during push 鈫?emit **`FAILED`** status
-
-### AgentCard Data
-
-- name: `SPN Domain Agent`, provider: `Huawei`
-- Declares only the `Notification-T/NL/v1` extension (subscribe use case does not involve Task-T)
-
-### Key Points
-
-- **SDK as middleware**: client/server never call LLM directly; they go through A2ATClient/A2ATServer
-- **LLM only for prompt phase**: no LLM calls during artifact push
-- **No negotiation**: client submits complete input; server validates and pushes directly
-- **Three-layer decoupling**: client (discover + consume) -> server (register + push) -> registry (registry center)
-- **Mock capability preserved**: `common/mock_llm.py` + `resources/mock_responses/` auto-activate when the API key is empty; the full flow runs without a real API
-- **How to tell it's mock**: before each mock LLM response, a standalone log line `[llm] llm-mock: using canned mock LLM response` is printed; this line is absent when using a real LLM
-
-## Run Tests
-
-```bash
-# Run all use case tests (from the a2a-t-sample directory)
-uv run pytest subscribe-incident/test/ -v
-uv run pytest negotiation/test/ -v
-```
-
-## Use Case: negotiation
-
-An **offline** negotiation closed-loop demo (the Python counterpart of the Java
-`a2a-t-sample` NegotiationDemoApp, with the embedded HTTP server replaced by an in-process
-runtime call). It drives the 4-message flow:
-
-1. client -> Task-T prompt generated with **missing** params (`generate_task_prompt_from_data_with_schema`);
-2. server -> `validate_task_prompt_and_data_filling` rejects (params missing) -> Negotiation-T
-   information **propose** -> `INPUT_REQUIRED`;
-3. client -> Task-T prompt with **filled** params + Negotiation-T **accept**;
-4. server -> validation passes -> diagnosis result -> `COMPLETED`.
-
-### Run (offline, no API key needed)
-
-```bash
-# From the a2a-t-sample directory; PYTHONPATH points at the use case src
-$env:PYTHONPATH = "$pwd\negotiation\src"     # PowerShell; export PYTHONPATH=... on bash
-
-uv run python -m negotiation_demo                 # fromData strategy (deterministic negotiation messages)
-uv run python -m negotiation_demo --fromText      # fromText strategy (one LLM extraction per negotiation message)
-uv run python -m negotiation_demo --language zh-CN
-```
-
-Without `A2AT_LLM_API_KEY` in `.env`, the scripted mock LLM (`negotiation/src/negotiation_demo/shared/mock_llm.py`)
-serves the slot-extraction, content-validation and negotiation-extraction calls from
-`negotiation/resources/mock_responses/`, so the whole round trip runs offline. With a real API
-key the same flow calls the real LLM (the negotiation messages of the fromData strategy still
-make no LLM call —?the SDK renders the typed content directly from the template).
-
-### Flow
-
-| Stage | Who calls SDK | What SDK does | LLM calls (fromData) |
-|-------|--------------|---------------|----------------------|
-| Message 1 | client | Task-T slot extraction + template render (`fromDataWithSchema`) | 1 |
-| Message 2 | server | `validate_task_prompt_and_data_filling` + negotiation propose generation | 1 (+0 fromData) |
-| Message 3 | client | Task-T generation + negotiation accept generation | 1 (+0 fromData) |
-| Message 4 | server | `validate_task_prompt_and_data_filling` + diagnosis rendering | 1 |
+This project is licensed under the [Apache-2.0](LICENSE) license.
diff --git a/a2a-t-sample/observability-sample/src/obs_sample/server_main.py b/a2a-t-sample/observability-sample/src/obs_sample/server_main.py
index 8ae8c9bc..bed082e2 100644
--- a/a2a-t-sample/observability-sample/src/obs_sample/server_main.py
+++ b/a2a-t-sample/observability-sample/src/obs_sample/server_main.py
@@ -25,7 +25,14 @@
 from a2a.server.tasks.base_push_notification_sender import BasePushNotificationSender
 from a2a.server.tasks.inmemory_push_notification_config_store import InMemoryPushNotificationConfigStore
 from a2a.server.tasks.inmemory_task_store import InMemoryTaskStore
-from a2a.types import Artifact, TaskArtifactUpdateEvent, TaskState, TaskStatus, TaskStatusUpdateEvent
+from a2a.types import (
+    Artifact,
+    Task,
+    TaskArtifactUpdateEvent,
+    TaskState,
+    TaskStatus,
+    TaskStatusUpdateEvent,
+)
 from a2a.utils.constants import TransportProtocol
 from a2a_t.observability import A2ATRequestHandlerDecorator
 from google.protobuf.json_format import MessageToDict, ParseDict
@@ -41,11 +48,19 @@ def _sample_port() -> int:
 
 
 class EchoExecutor(AgentExecutor):
-    """Emits submitted -> working -> artifact -> completed; mirrors a Task-T streaming flow."""
+    """Emits the initial Task snapshot -> submitted -> working -> artifact -> completed
+    (V2 protocol: the Task snapshot must precede any status event)."""
 
     async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
         task_id = context.task_id or "T-sample"
         context_id = context.context_id or "C-sample"
+        await event_queue.enqueue_event(
+            Task(
+                id=task_id,
+                context_id=context_id,
+                status=TaskStatus(state=TaskState.TASK_STATE_SUBMITTED),
+            )
+        )
         for state in (TaskState.TASK_STATE_SUBMITTED, TaskState.TASK_STATE_WORKING):
             await event_queue.enqueue_event(
                 TaskStatusUpdateEvent(task_id=task_id, context_id=context_id, status=TaskStatus(state=state))
@@ -135,6 +150,13 @@ async def _run_agreed_task(self, context: RequestContext, event_queue: EventQueu
         """Post-agreement task execution: the task is created and runs to completion."""
         task_id = context.task_id or "T-negotiation-agreed"
         context_id = context.context_id or "C-negotiation"
+        await event_queue.enqueue_event(
+            Task(
+                id=task_id,
+                context_id=context_id,
+                status=TaskStatus(state=TaskState.TASK_STATE_SUBMITTED),
+            )
+        )
         for state in (TaskState.TASK_STATE_SUBMITTED, TaskState.TASK_STATE_WORKING):
             await event_queue.enqueue_event(
                 TaskStatusUpdateEvent(task_id=task_id, context_id=context_id, status=TaskStatus(state=state))
diff --git a/src/a2a_t/observability/server/handler_decorator.py b/src/a2a_t/observability/server/handler_decorator.py
index 5ce77d1c..1b69f929 100644
--- a/src/a2a_t/observability/server/handler_decorator.py
+++ b/src/a2a_t/observability/server/handler_decorator.py
@@ -534,12 +534,16 @@ async def send_notification(self, task_id: str, event: Any) -> Any:
         if not self._decorator_active():
             return await self._inner.send_notification(task_id, event)
         info = classify_event(_unwrap(event))
-        self._log_push(task_id, info)
+        dispatch_bearing = _safe_getattr(self._inner, "_dispatch_notification") is not None
+        if not dispatch_bearing:
+            # Only the plain-ABC path logs here: dispatch-bearing inners emit one
+            # log per push_info inside their wrapped _dispatch_notification.
+            self._log_push(task_id, info)
         # V2/core senders route fanout through ``_dispatch_notification`` (whose
         # wrapper already emits the per-push_info span with the URL) - forward
         # without a second span. Plain ABC senders (no dispatch method) get the
         # span here, with the URL read off the event (when present).
-        if _safe_getattr(self._inner, "_dispatch_notification") is not None:
+        if dispatch_bearing:
             return await self._inner.send_notification(task_id, event)
         span = self._start_push_span_for_url(task_id, info, _safe_getattr(event, "url"))
         cm: Any = _safe_use_span(span)
@@ -657,7 +661,6 @@ def _wrap_legacy_collaborators(self, executor: Any, push_sender: Any) -> None:
                 self._inner.agent_executor = A2ATAgentExecutorDecorator(executor, config=self._config)
             except Exception:  # noqa: BLE001
                 logger.warning("a2at: failed to wrap agent_executor", exc_info=True)
-        push_sender = _safe_getattr(self._inner, "_push_sender")
         if push_sender is not None and not isinstance(push_sender, A2ATPushSenderDecorator):
             try:
                 decorator = A2ATPushSenderDecorator(push_sender, config=self._config)
@@ -675,7 +678,11 @@ def _wrap_v2_collaborators(self, executor: Any, push_sender: Any) -> None:
             if push_sender is not None and not isinstance(push_sender, A2ATPushSenderDecorator):
                 push_sender = A2ATPushSenderDecorator(push_sender, config=self._config)
                 self._inner._push_sender = push_sender
-                push_sender.install_dispatch_wrapper()
+                if not push_sender.install_dispatch_wrapper():
+                    logger.debug(
+                        "a2at: V2 push sender has no _dispatch_notification; "
+                        "push spans will be emitted via the send_notification surface"
+                    )
             registry = _safe_getattr(self._inner, "_active_task_registry")
             registry_cls = type(registry)
             self._inner._active_task_registry = registry_cls(
diff --git a/tests/integration/test_e2e_observability.py b/tests/integration/test_e2e_observability.py
index 7d96fb26..4a07a9c8 100644
--- a/tests/integration/test_e2e_observability.py
+++ b/tests/integration/test_e2e_observability.py
@@ -1,8 +1,8 @@
-"""End-to-end observability verification for the v3.0 decorator architecture (spec 閹?0.2, assertions 1-10).
+"""End-to-end observability verification for the v3.0 decorator architecture (spec §10.2, assertions 1-10).
 
 In-process wiring: httpx ASGITransport -> Starlette (a2a-python REST routes) ->
 A2ATRequestHandlerDecorator(DefaultRequestHandler(...)) on the server and a real
-a2a ClientFactory client with register_client_factory(factory) on the client 闁?zero business observability code. No ASGI middleware: the server entry span
+a2a ClientFactory client with register_client_factory(factory) on the client —?zero business observability code. No ASGI middleware: the server entry span
 extracts the traceparent from ``ServerCallContext.state["headers"]`` (spec
 appendix #5), which the client transport decorator injected into
 ``context.service_parameters``.
@@ -19,11 +19,11 @@
 - Streaming transport/handler methods are async generators: a2a's trace_class
   applies its sync wrapper, so a2a's own protocol spans end at generator
   creation. Because ASGITransport runs the server app inside the client entry
-  span activation, ambient propagation coincides with the traceparent path 闁?  both route through the same client entry span.
+  span activation, ambient propagation coincides with the traceparent path —?  both route through the same client entry span.
 - TaskStatusUpdateEvent has no ``final`` field; the terminal state itself is
   final.
 
-Assertion coverage (spec 閹?0.2 v3 table):
+Assertion coverage (spec §10.2 v3 table):
 
 1/2/3/8 -> test_e2e_trace_continuity_event_spans_and_metrics
 4       -> test_e2e_stream_error_span
@@ -581,7 +581,6 @@ async def test_e2e_push_notification_span(otel_setup: tuple[Any, Any]) -> None:
     assert push_spans
 
     for span in push_spans:
-        print("PUSH-DEBUG attrs:", dict(span.attributes or {}))
         attrs = span.attributes or {}
         assert span.kind is SpanKind.CLIENT
         assert attrs["gen_ai.agent.a2at.push.notification.url"] == "http://push-receiver.test/hook"
@@ -697,3 +696,45 @@ async def test_register_client_factory_decorates_real_factory_transport() -> Non
 
 
 
+
+
+async def test_sample_app_smoke_on_official_handler(otel_setup: tuple[Any, Any]) -> None:
+    """Smoke guard: the observability-sample app must work on the official
+    DefaultRequestHandler — the V2 protocol requires the executor to enqueue
+    the initial Task snapshot before any status event (regression guard for
+    the sample executors, which have no automated coverage otherwise)."""
+    import sys
+    from pathlib import Path as _Path
+
+    sample_src = _Path(__file__).resolve().parents[2] / "a2a-t-sample" / "observability-sample" / "src"
+    sys.path.insert(0, str(sample_src))
+    try:
+        from a2a.types import AgentCard
+        from google.protobuf.json_format import ParseDict
+        from obs_sample.server_main import build_agent_card_payload, build_app
+    finally:
+        sys.path.remove(str(sample_src))
+
+    exporter, _reader = otel_setup
+    app = build_app(scenario="negotiation")
+    card = ParseDict(build_agent_card_payload(8100), AgentCard())
+    client, httpx_client = _make_client(app, card)
+    try:
+        # The sample server matches the Negotiation-T URI WITH the /v1 suffix
+        events = await _send_streaming(client, {_NEG_T_URI + "/v1": "offer"}, context_id="C-smoke")
+    finally:
+        await httpx_client.aclose()
+    assert events and events[-1].HasField("message")
+    await _finalize_abandoned_generators()
+
+    spans = _a2at_spans(exporter.get_finished_spans())
+    client_negotiations = [
+        s for s in spans if s.name == "SendStreamingMessage-negotiation" and s.kind is SpanKind.CLIENT
+    ]
+    server_negotiations = [
+        s for s in spans if s.name == "SendStreamingMessage-negotiation" and s.kind is SpanKind.SERVER
+    ]
+    assert len(client_negotiations) == 1
+    assert len(server_negotiations) == 1
+    assert client_negotiations[0].parent is not None
+    assert server_negotiations[0].parent is not None
diff --git a/tests/observability/test_handler_decorator.py b/tests/observability/test_handler_decorator.py
index 1e396b45..fc2c41ad 100644
--- a/tests/observability/test_handler_decorator.py
+++ b/tests/observability/test_handler_decorator.py
@@ -844,17 +844,29 @@ async def test_v2_handler_registry_rebuilt_with_decorated_collaborators(
     assert any("V2" in record.getMessage() and "rebuilt" in record.getMessage() for record in caplog.records)
 
 
-async def test_v2_registry_rebuild_failure_degrades_gracefully(exporter: InMemorySpanExporter) -> None:
+async def test_v2_registry_rebuild_failure_degrades_gracefully(
+    exporter: InMemorySpanExporter, caplog: pytest.LogCaptureFixture
+) -> None:
     """registry 签名不兼容（版本变化）→ WARNING 降级：入口 span 仍工作（spec §8.2）。"""
-    inner = FakeSimpleHandler(result=FakeMessage(metadata={}, task_id="T26b", context_id="C26b"))
-    inner.agent_executor = FakeExecutor(events=[])
+    import logging
+
+    inner = FakeStreamingHandler(FakeExecutor(events=[]))
     inner._push_sender = FakeDispatchPushSender(["https://hook.example"])
     inner._active_task_registry = _StrictRegistry()
 
-    A2ATRequestHandlerDecorator(inner)
+    with caplog.at_level(logging.WARNING, logger="a2at.observability"):
+        A2ATRequestHandlerDecorator(inner)
 
-    assert isinstance(inner.agent_executor, A2ATAgentExecutorDecorator)
-    assert inner._active_task_registry.__class__ is _StrictRegistry
+    assert any("V2 registry rebuild failed" in record.getMessage() for record in caplog.records), (
+        "expected the rebuild-failure WARNING"
+    )
+
+    # D9 test-plan item 5: 入口 span 仍工作——一次流式请求仍产生 SERVER 入口 span
+    decorator = A2ATRequestHandlerDecorator(inner)
+    params = make_send_request({}, task_id="T26c", context_id="C26c")
+    _ = [event async for event in decorator.on_message_send_stream(params, FakeServerCallContext())]
+    entry = [s for s in exporter.get_finished_spans() if s.name == "SendStreamingMessage"]
+    assert len(entry) == 1 and entry[0].kind is SpanKind.SERVER
 
 
 async def test_stream_negotiation_message_creates_server_negotiation_span(

From 7de4f1c78efc6e13a04d60d0b08021c4b4785ae0 Mon Sep 17 00:00:00 2001
From: z00930736 <zoujie5@huawei.com>
Date: Thu, 15 Oct 2026 18:14:20 +0800
Subject: [PATCH 5/7] docs(sample): note the A2AT_SERVICE_NAME fallback in the
 env tables

Signed-off-by: z00930736 <zoujie5@huawei.com>
---
 a2a-t-sample/observability-sample/README-zh.md | 2 +-
 a2a-t-sample/observability-sample/README.md    | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/a2a-t-sample/observability-sample/README-zh.md b/a2a-t-sample/observability-sample/README-zh.md
index 3920b512..299263b3 100644
--- a/a2a-t-sample/observability-sample/README-zh.md
+++ b/a2a-t-sample/observability-sample/README-zh.md
@@ -113,7 +113,7 @@ export LANGFUSE_BASE_URL="https://us.cloud.langfuse.com"
 | `A2AT_OBSERVABILITY_ENABLED` | `true` | 总开关——`false` 时 SDK 完全透明 |
 | `OTEL_EXPORTER_OTLP_ENDPOINT` | 未设置 | 把 trace/指标/日志发到 OTLP collector，而不是控制台 |
 | `A2AT_CONSOLE_EXPORTER` | `true` | `false` 关闭 dev 模式控制台兜底（stdout 不再输出 JSON） |
-| `OTEL_SERVICE_NAME` | `a2a-t-agent` | 本进程在后端的名称 |
+| `OTEL_SERVICE_NAME`（或 `A2AT_SERVICE_NAME`） | `a2a-t-agent` | 本进程在后端的名称 |
 | `A2AT_OBS_SAMPLE_PORT` | `8100` | 样例服务端端口 |
 | `LANGFUSE_PUBLIC_KEY` / `LANGFUSE_SECRET_KEY` / `LANGFUSE_BASE_URL` | 未设置 | 启用 Langfuse 后端 |
 
diff --git a/a2a-t-sample/observability-sample/README.md b/a2a-t-sample/observability-sample/README.md
index 07ba000b..87a20497 100644
--- a/a2a-t-sample/observability-sample/README.md
+++ b/a2a-t-sample/observability-sample/README.md
@@ -118,7 +118,7 @@ Commonly used:
 | `A2AT_OBSERVABILITY_ENABLED` | `true` | Master switch — `false` makes the SDK fully transparent |
 | `OTEL_EXPORTER_OTLP_ENDPOINT` | *(unset)* | Send spans/metrics/logs to an OTLP collector instead of the console |
 | `A2AT_CONSOLE_EXPORTER` | `true` | `false` disables the dev-mode console fallback (no stdout JSON) |
-| `OTEL_SERVICE_NAME` | `a2a-t-agent` | How this process is named in the backend |
+| `OTEL_SERVICE_NAME` (or `A2AT_SERVICE_NAME`) | `a2a-t-agent` | How this process is named in the backend |
 | `A2AT_OBS_SAMPLE_PORT` | `8100` | Sample server port |
 | `LANGFUSE_PUBLIC_KEY` / `LANGFUSE_SECRET_KEY` / `LANGFUSE_BASE_URL` | *(unset)* | Enable the Langfuse backend |
 

From ce354659c7337e3aa0022853f3dd9dc032847552 Mon Sep 17 00:00:00 2001
From: z00930736 <zoujie5@huawei.com>
Date: Fri, 16 Oct 2026 16:17:58 +0800
Subject: [PATCH 6/7] fix(sample): mock resources path after the #74 directory
 rename

The observability-sample mock_llm pointed at subscribe-incident/resources
(which was renamed to fault-management in #74) - the a2at/negotiation
scenarios crashed with FileNotFoundError on mock-response load. Both
directory names are now accepted.

Signed-off-by: z00930736 <zoujie5@huawei.com>
---
 .../observability-sample/src/obs_sample/mock_llm.py  | 12 +++++++++---
 1 file changed, 9 insertions(+), 3 deletions(-)

diff --git a/a2a-t-sample/observability-sample/src/obs_sample/mock_llm.py b/a2a-t-sample/observability-sample/src/obs_sample/mock_llm.py
index eb89ef63..84f9f81b 100644
--- a/a2a-t-sample/observability-sample/src/obs_sample/mock_llm.py
+++ b/a2a-t-sample/observability-sample/src/obs_sample/mock_llm.py
@@ -6,7 +6,7 @@
 responses (subscribe-incident does the latter in its llm_logger, which this sample
 does not carry).
 
-Mock response files live under ../subscribe-incident/resources/mock_responses/.
+Mock response files live under ../fault-management/resources/mock_responses/.
 """
 
 from __future__ import annotations
@@ -19,8 +19,14 @@
 from dotenv import dotenv_values
 
 # mock_llm.py → obs_sample → src → observability-sample → a2a-t-sample (parents[3]),
-# then into subscribe-incident's shared mock resources.
-_RESOURCES_DIR = Path(__file__).resolve().parents[3] / "subscribe-incident" / "resources" / "mock_responses"
+# then into the shared mock resources (the directory was renamed from
+# subscribe-incident to fault-management in #74; both names are accepted so the
+# sample keeps working on older checkouts).
+_MOCK_DIR_CANDIDATES = [
+    Path(__file__).resolve().parents[3] / "fault-management" / "resources" / "mock_responses",
+    Path(__file__).resolve().parents[3] / "subscribe-incident" / "resources" / "mock_responses",
+]
+_RESOURCES_DIR = next((p for p in _MOCK_DIR_CANDIDATES if p.exists()), _MOCK_DIR_CANDIDATES[0])
 
 _MOCK_RESPONSE_FILES = [
     "scenario_recognition.json",

From b2bdab3291b7b436a9b321e2258e92de2c144716 Mon Sep 17 00:00:00 2001
From: z00930736 <zoujie5@huawei.com>
Date: Fri, 16 Oct 2026 17:14:25 +0800
Subject: [PATCH 7/7] fix(ci): drop the repo-level Tsinghua PyPI mirror

The [[tool.uv.index]] block pinned every uv resolution (including CI) to
pypi.tuna.tsinghua.edu.cn, which is intermittently unreachable from
GitHub Actions runners - the e2e job failed downloading babel through
it. Remove the repo-level mirror so CI resolves from official PyPI;
developers in mainland China can keep a fast mirror via their own uv
configuration (UV_DEFAULT_INDEX or ~/.config/uv/uv.toml) instead.

Signed-off-by: z00930736 <zoujie5@huawei.com>
---
 pyproject.toml | 4 ----
 1 file changed, 4 deletions(-)

diff --git a/pyproject.toml b/pyproject.toml
index b8478014..b942d115 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -65,10 +65,6 @@ integration = [
 Homepage = "https://github.com/project-openan/a2a-t-sdk-python"
 Documentation = "https://github.com/project-openan/a2a-t-sdk-python#readme"
 
-[[tool.uv.index]]
-url = "https://pypi.tuna.tsinghua.edu.cn/simple"
-default = true
-
 [tool.uv.build-backend]
 module-name = "a2a_t"
 source-exclude = [".gitignore"]