# claude-agent-sdk 0.1.76 升级与新特性适配 — 实施计划

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** 把 `claude-agent-sdk` 依赖下限提升到 0.1.76，并消费两个低风险新字段（`ResultMessage.api_error_status` 透传 SSE + 错误日志；`ToolPermissionContext.decision_reason` 拼到 `_can_use_tool` default-deny 提示）。

**Architecture:** 后端单进程改动。pyproject 改下限，`uv sync` 重锁。`server/agent_runtime/service.py` 在 `_build_status_event_payload` 和 result 错误日志路径消费 `api_error_status`；`server/agent_runtime/session_manager.py` 在 `_can_use_tool` default-deny 分支用 `getattr` 读 `decision_reason`。所有新字段读取都做 None 兜底，向后兼容。

**Tech Stack:** Python 3.x, claude-agent-sdk 0.1.76, FastAPI/SSE, pytest (asyncio_mode=auto), uv 包管理, ruff lint/format。

**参考 spec:** `docs/superpowers/specs/2026-05-07-sdk-0.1.76-upgrade-design.md`

---

## File Structure

| 文件 | 类型 | 责任 |
|---|---|---|
| `pyproject.toml` | 修改 | 依赖下限 `>=0.1.73` → `>=0.1.76` |
| `uv.lock` | 修改 | `uv sync` 产物，0.1.74 → 0.1.76 |
| `server/agent_runtime/service.py` | 修改 | `_build_status_event_payload` 注入 `api_error_status` 字段；`result` 消息错误路径新增结构化 `logger.warning` |
| `server/agent_runtime/session_manager.py` | 修改 | `_can_use_tool` default-deny 分支读 `getattr(_context, "decision_reason", None)` 拼到 hint |
| `tests/agent_runtime/test_sdk_0_1_76_features.py` | 创建 | 4 个单测覆盖新字段消费 |

---

## Task 1: 升级 SDK 依赖到 0.1.76

**Files:**
- Modify: `pyproject.toml`（搜索字符串 `claude-agent-sdk>=0.1.73`）
- Modify: `uv.lock`（自动产物）

- [ ] **Step 1: 修改 pyproject.toml**

定位文件中 `"claude-agent-sdk>=0.1.73",` 一行，替换为 `"claude-agent-sdk>=0.1.76",`。

可用以下命令验证当前内容：

```bash
grep -n "claude-agent-sdk" pyproject.toml
```

预期：输出 `"claude-agent-sdk>=0.1.73",`（升级前）。

用 Edit 工具改成：

```
"claude-agent-sdk>=0.1.76",
```

- [ ] **Step 2: 重锁依赖**

```bash
uv sync
```

预期：`Resolved N packages` 输出，`claude-agent-sdk` 版本变为 `0.1.76`。

- [ ] **Step 3: 验证 lock 已升级**

```bash
grep -A1 'name = "claude-agent-sdk"' uv.lock | head -10
```

预期：能看到 `claude_agent_sdk-0.1.76` 字符串（sdist URL 或版本号）。

- [ ] **Step 4: 跑现有 agent_runtime 测试套件回归**

```bash
uv run python -m pytest tests/agent_runtime/ -v
```

预期：全部通过，无 ImportError，无 AttributeError。如出现失败，**停下来定位**，不要绕过——很可能是 0.1.74 atexit / `MemoryObjectReceiveStream` / `created_at` 修复导致的行为差异。

- [ ] **Step 5: 跑 agent_session_store 测试回归**

```bash
uv run python -m pytest tests/agent_session_store/ -v
```

预期：全部通过。该模块用 `fold_session_summary`、`project_key_for_directory`，需确认 0.1.76 这两个 API 未变。

- [ ] **Step 6: Commit**

```bash
git add pyproject.toml uv.lock
git commit -m "chore(deps): bump claude-agent-sdk >=0.1.73 → >=0.1.76"
```

---

## Task 2: 新增测试文件骨架 + api_error_status 透传单测（红）

**Files:**
- Create: `tests/agent_runtime/test_sdk_0_1_76_features.py`

- [ ] **Step 1: 创建测试文件**

用 Write 工具创建 `tests/agent_runtime/test_sdk_0_1_76_features.py`，写入以下内容：

```python
"""SDK 0.1.76 新字段消费的单元测试。

覆盖：
- ResultMessage.api_error_status 透传到 SSE 状态 payload
- ToolPermissionContext.decision_reason 拼到 _can_use_tool default-deny hint
"""

from __future__ import annotations

import logging
from types import SimpleNamespace
from typing import Any

import pytest

from server.agent_runtime.service import AssistantService


class TestApiErrorStatusInStatusPayload:
    """0.1.76 新字段 api_error_status 透传到 SSE payload。"""

    def test_api_error_status_present_when_set(self):
        result_message: dict[str, Any] = {
            "type": "result",
            "subtype": "error_during_execution",
            "is_error": True,
            "api_error_status": 429,
            "stop_reason": "api_error",
        }
        payload = AssistantService._build_status_event_payload(
            status="error",
            session_id="sess-1",
            result_message=result_message,
        )
        assert payload["api_error_status"] == 429

    def test_api_error_status_absent_when_none(self):
        result_message: dict[str, Any] = {
            "type": "result",
            "subtype": "success",
            "is_error": False,
            "api_error_status": None,
        }
        payload = AssistantService._build_status_event_payload(
            status="completed",
            session_id="sess-2",
            result_message=result_message,
        )
        assert "api_error_status" not in payload

    def test_api_error_status_absent_when_field_missing(self):
        # 老 SDK / 老消息没有 api_error_status 字段
        result_message: dict[str, Any] = {
            "type": "result",
            "subtype": "success",
            "is_error": False,
        }
        payload = AssistantService._build_status_event_payload(
            status="completed",
            session_id="sess-3",
            result_message=result_message,
        )
        assert "api_error_status" not in payload
```

- [ ] **Step 2: 跑测试，确认前两个 fail**

```bash
uv run python -m pytest tests/agent_runtime/test_sdk_0_1_76_features.py -v
```

预期：
- `test_api_error_status_present_when_set` **FAIL**：`KeyError: 'api_error_status'` 或 `assert ... in payload`
- `test_api_error_status_absent_when_none` **PASS**（payload 本来就不含该键）
- `test_api_error_status_absent_when_field_missing` **PASS**（同上）

记录第一个测试 FAIL 的具体错误信息，证明红→绿循环开始。

---

## Task 3: 实现 api_error_status 透传（绿）

**Files:**
- Modify: `server/agent_runtime/service.py`（`_build_status_event_payload`，约 line 725-748）

- [ ] **Step 1: 改造 `_build_status_event_payload`**

定位 `service.py` 中以下原始片段：

```python
        return {
            "status": status,
            "subtype": subtype,
            "stop_reason": stop_reason,
            "is_error": is_error,
            "session_id": session_id,
        }
```

替换为：

```python
        payload: dict[str, Any] = {
            "status": status,
            "subtype": subtype,
            "stop_reason": stop_reason,
            "is_error": is_error,
            "session_id": session_id,
        }
        api_error_status = message.get("api_error_status")  # SDK 0.1.76+
        if api_error_status is not None:
            payload["api_error_status"] = api_error_status
        return payload
```

注意：`message` 变量已在函数顶部定义为 `result_message if isinstance(result_message, dict) else {}`，可直接用。

- [ ] **Step 2: 跑测试，确认全部 pass**

```bash
uv run python -m pytest tests/agent_runtime/test_sdk_0_1_76_features.py::TestApiErrorStatusInStatusPayload -v
```

预期：3 个测试全部 PASS。

- [ ] **Step 3: Commit**

```bash
git add server/agent_runtime/service.py tests/agent_runtime/test_sdk_0_1_76_features.py
git commit -m "feat(sdk): 透传 ResultMessage.api_error_status 到 SSE 状态 payload"
```

---

## Task 4: 错误路径结构化日志单测（红）

**Files:**
- Modify: `tests/agent_runtime/test_sdk_0_1_76_features.py`（追加测试类）

- [ ] **Step 1: 追加日志测试类**

在测试文件末尾追加：

```python


class TestResultErrorLogging:
    """result 消息错误路径写结构化 logger.warning，含 api_error_status。"""

    @pytest.mark.asyncio
    async def test_logger_warning_emitted_on_error_with_api_status(
        self,
        caplog: pytest.LogCaptureFixture,
        tmp_path,
    ):
        from unittest.mock import MagicMock

        service = AssistantService(project_root=tmp_path)
        projector = MagicMock()
        projector.apply_message.return_value = {}  # 无 patch/delta/question

        result_message: dict[str, Any] = {
            "type": "result",
            "subtype": "error_during_execution",
            "is_error": True,
            "api_error_status": 429,
            "stop_reason": "api_error",
        }
        with caplog.at_level(logging.WARNING, logger="server.agent_runtime.service"):
            events, terminal = await service._dispatch_live_message(
                message=result_message,
                projector=projector,
                session_id="sess-err",
            )
        assert terminal is True
        warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
        assert any(
            getattr(r, "api_error_status", None) == 429 for r in warnings
        ), f"expected a warning with api_error_status=429, got {[r.__dict__ for r in warnings]}"

    @pytest.mark.asyncio
    async def test_no_warning_on_completed_result(
        self,
        caplog: pytest.LogCaptureFixture,
        tmp_path,
    ):
        from unittest.mock import MagicMock

        service = AssistantService(project_root=tmp_path)
        projector = MagicMock()
        projector.apply_message.return_value = {}

        result_message: dict[str, Any] = {
            "type": "result",
            "subtype": "success",
            "is_error": False,
        }
        with caplog.at_level(logging.WARNING, logger="server.agent_runtime.service"):
            await service._dispatch_live_message(
                message=result_message,
                projector=projector,
                session_id="sess-ok",
            )
        warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
        assert not warnings, f"unexpected warnings on completed result: {warnings}"
```

> ℹ️ 已核实：`_dispatch_live_message` 是 `service.py:477` 中 result 消息的 async 处理入口；`AssistantStreamProjector.apply_message`（`stream_projector.py:455`）是同步方法，可直接 MagicMock。`asyncio_mode = "auto"` 已在项目 pytest 配置生效，所以 `@pytest.mark.asyncio` 也可省略；保留写法确保意图明确。

- [ ] **Step 2: 跑测试，确认 1 个 fail**

```bash
uv run python -m pytest tests/agent_runtime/test_sdk_0_1_76_features.py::TestResultErrorLogging -v
```

预期：
- `test_logger_warning_emitted_on_error_with_api_status` **FAIL**：当前 result 路径没有 `logger.warning` 调用
- `test_no_warning_on_completed_result` **PASS**：当前没有日志，所以不会有 warning

---

## Task 5: 实现错误路径结构化日志（绿）

**Files:**
- Modify: `server/agent_runtime/service.py`（result 消息分支，参考 spec 3.2 段落）

- [ ] **Step 1: 定位 result 消息分支**

```bash
grep -n 'msg_type == "result"' server/agent_runtime/service.py
```

预期输出指向 `service.py:540` 附近。原始片段（约 line 540-551）：

```python
        if msg_type == "result":
            events.append(
                self._sse_event(
                    "status",
                    self._build_status_event_payload(
                        status=self._resolve_result_status(message),
                        session_id=session_id,
                        result_message=message,
                    ),
                )
            )
            return events, True
```

- [ ] **Step 2: 替换为含日志版本**

```python
        if msg_type == "result":
            status = self._resolve_result_status(message)
            if status == "error":
                logger.warning(
                    "assistant session result error",
                    extra={
                        "session_id": session_id,
                        "subtype": message.get("subtype"),
                        "is_error": message.get("is_error"),
                        "api_error_status": message.get("api_error_status"),  # SDK 0.1.76+
                        "stop_reason": message.get("stop_reason"),
                    },
                )
            events.append(
                self._sse_event(
                    "status",
                    self._build_status_event_payload(
                        status=status,
                        session_id=session_id,
                        result_message=message,
                    ),
                )
            )
            return events, True
```

注意：`logger` 已在文件顶部 `logger = logging.getLogger(__name__)` 定义，无需新 import。

- [ ] **Step 3: 跑测试，确认全部 pass**

```bash
uv run python -m pytest tests/agent_runtime/test_sdk_0_1_76_features.py -v
```

预期：5 个测试全部 PASS（前 3 个 + 新加 2 个）。

- [ ] **Step 4: Commit**

```bash
git add server/agent_runtime/service.py tests/agent_runtime/test_sdk_0_1_76_features.py
git commit -m "feat(sdk): result 错误路径写结构化日志，附 api_error_status"
```

---

## Task 6: decision_reason 拼到 default-deny 单测（红）

**Files:**
- Modify: `tests/agent_runtime/test_sdk_0_1_76_features.py`

- [ ] **Step 1: 追加 can_use_tool 测试类**

在测试文件末尾追加：

```python


class TestCanUseToolDecisionReason:
    """0.1.74 新字段 ToolPermissionContext.decision_reason 拼到 default-deny hint。"""

    @pytest.mark.asyncio
    async def test_default_deny_includes_decision_reason(self, tmp_path):
        from server.agent_runtime.session_manager import SessionManager
        from server.agent_runtime.session_store import SessionMetaStore

        data_dir = tmp_path / "data"
        data_dir.mkdir(parents=True, exist_ok=True)
        sm = SessionManager(
            project_root=tmp_path,
            data_dir=data_dir,
            meta_store=SessionMetaStore(),
        )
        callback = await sm._build_can_use_tool_callback(session_id="sess-x")
        ctx = SimpleNamespace(decision_reason="No matching allow rule")
        result = await callback("Bash", {"command": "rm -rf /"}, ctx)
        assert hasattr(result, "message"), f"expected PermissionResultDeny, got {type(result)}"
        assert "No matching allow rule" in result.message
        assert "上游决策原因" in result.message

    @pytest.mark.asyncio
    async def test_default_deny_without_decision_reason_does_not_raise(self, tmp_path):
        from server.agent_runtime.session_manager import SessionManager
        from server.agent_runtime.session_store import SessionMetaStore

        data_dir = tmp_path / "data"
        data_dir.mkdir(parents=True, exist_ok=True)
        sm = SessionManager(
            project_root=tmp_path,
            data_dir=data_dir,
            meta_store=SessionMetaStore(),
        )
        callback = await sm._build_can_use_tool_callback(session_id="sess-y")
        ctx = SimpleNamespace()  # no decision_reason
        result = await callback("Bash", {"command": "echo hi"}, ctx)
        assert hasattr(result, "message")
        assert "未授权的工具调用" in result.message
        assert "上游决策原因" not in result.message
```

> ℹ️ 已核实：`SessionManager.__init__(project_root, data_dir, meta_store)`（`session_manager.py:341`），`data_dir` 不会自动 mkdir（service 层负责），所以测试里要先 `mkdir(parents=True, exist_ok=True)`。`SessionMetaStore()` 默认构造无参可直接调。

- [ ] **Step 2: 跑测试，确认 1 个 fail**

```bash
uv run python -m pytest tests/agent_runtime/test_sdk_0_1_76_features.py::TestCanUseToolDecisionReason -v
```

预期：
- `test_default_deny_includes_decision_reason` **FAIL**：当前 hint 不含 "上游决策原因" 字符串
- `test_default_deny_without_decision_reason_does_not_raise` **PASS**：当前 hint 本来就不含这串

---

## Task 7: 实现 decision_reason 拼接（绿）

**Files:**
- Modify: `server/agent_runtime/session_manager.py`（`_can_use_tool` default-deny 分支，约 line 1772-1782）

- [ ] **Step 1: 定位 default-deny 分支**

```bash
grep -n "未授权的工具调用" server/agent_runtime/session_manager.py
```

预期输出指向 `session_manager.py:1774` 附近。

原始片段（约 line 1772-1782）：

```python
            if PermissionResultDeny is not None:
                hint = (
                    f"未授权的工具调用: {tool_name}"
                    f"({json.dumps(input_data, ensure_ascii=False)[:200]})\n"
                    "当前 Bash 白名单仅允许以下命令:\n"
                    "  - python .claude/skills/<skill>/scripts/<script>.py <args>（必须用相对路径）\n"
                    "  - ffmpeg / ffprobe\n"
                    "其他 Bash 命令均不可用。"
                    "请检查命令格式是否匹配白名单规则。"
                )
                return PermissionResultDeny(message=hint)
```

- [ ] **Step 2: 替换为含 decision_reason 拼接版本**

```python
            if PermissionResultDeny is not None:
                reason = getattr(_context, "decision_reason", None)  # SDK 0.1.74+
                reason_line = f"上游决策原因: {reason}\n" if reason else ""
                hint = (
                    f"未授权的工具调用: {tool_name}"
                    f"({json.dumps(input_data, ensure_ascii=False)[:200]})\n"
                    f"{reason_line}"
                    "当前 Bash 白名单仅允许以下命令:\n"
                    "  - python .claude/skills/<skill>/scripts/<script>.py <args>（必须用相对路径）\n"
                    "  - ffmpeg / ffprobe\n"
                    "其他 Bash 命令均不可用。"
                    "请检查命令格式是否匹配白名单规则。"
                )
                return PermissionResultDeny(message=hint)
```

- [ ] **Step 3: 跑测试，确认全部 pass**

```bash
uv run python -m pytest tests/agent_runtime/test_sdk_0_1_76_features.py -v
```

预期：7 个测试全部 PASS。

- [ ] **Step 4: Commit**

```bash
git add server/agent_runtime/session_manager.py tests/agent_runtime/test_sdk_0_1_76_features.py
git commit -m "feat(sdk): _can_use_tool default-deny 拼接 ToolPermissionContext.decision_reason"
```

---

## Task 8: 全套 lint + 覆盖率 + 端到端验收

**Files:**
- 无代码改动（除非 lint 自动修复）

- [ ] **Step 1: 跑 ruff check 修过的文件**

```bash
uv run ruff check server/agent_runtime/service.py server/agent_runtime/session_manager.py tests/agent_runtime/test_sdk_0_1_76_features.py
```

预期：`All checks passed!`。如有错误，按提示修复。

- [ ] **Step 2: 跑 ruff format**

```bash
uv run ruff format server/agent_runtime/service.py server/agent_runtime/session_manager.py tests/agent_runtime/test_sdk_0_1_76_features.py
```

预期：`X file(s) left unchanged` 或显示 reformatted。如 reformat，把改动也加进下一个 commit。

- [ ] **Step 3: 跑全套 agent_runtime 测试 + 覆盖率**

```bash
uv run python -m pytest tests/agent_runtime/ --cov=server/agent_runtime --cov-report=term -v
```

预期：
- 全部测试 PASS（包括新加的 7 个）
- agent_runtime 模块覆盖率 ≥ 升级前水平（不要求上升，只求不下降）

记录覆盖率数字到 commit message 或 PR 描述。

- [ ] **Step 4: 端到端手动验收（可选，但推荐）**

启动开发服务器：

```bash
uv run uvicorn server.app:app --reload --reload-dir server --reload-dir lib --port 1241
```

打开前端（如已 `pnpm dev`），跑一次完整 assistant 对话：
- 选一个项目，发一条消息
- 等回复完成（看到 result 事件）
- 在浏览器 devtools Network 面板找到 `/api/v1/assistant/sessions/{id}/stream` SSE，
  确认正常事件流没有破坏
- 如果想验 `api_error_status` 透传，可临时把 API key 设错触发 401/429，
  确认 SSE status 事件 payload 含 `api_error_status` 字段

如果端到端有异常，**不要 force ahead**——回到失败位置定位根因。

- [ ] **Step 5: 提交 lint/format 改动（如有）**

```bash
git status
# 如有 ruff format 改动：
git add -u
git commit -m "style: ruff format 0.1.76 升级相关文件"
```

如无改动，跳过此步。

- [ ] **Step 6: 检查 git log 概览**

```bash
git log --oneline main..HEAD
```

预期能看到（顺序无所谓）：
1. `chore(deps): bump claude-agent-sdk >=0.1.73 → >=0.1.76`
2. `feat(sdk): 透传 ResultMessage.api_error_status 到 SSE 状态 payload`
3. `feat(sdk): result 错误路径写结构化日志，附 api_error_status`
4. `feat(sdk): _can_use_tool default-deny 拼接 ToolPermissionContext.decision_reason`
5.（可选）`style: ruff format 0.1.76 升级相关文件`
6.（可选 docs commit 已在 brainstorming 阶段做完）

实施完成。

---

## 验收清单（与 spec 对齐）

- [ ] `pyproject.toml` 下限到 `>=0.1.76`
- [ ] `uv.lock` `claude-agent-sdk` 版本块为 0.1.76
- [ ] `_build_status_event_payload` 透传 `api_error_status`（None 时不写键）
- [ ] result 错误路径写结构化 `logger.warning`，含 `api_error_status` 字段
- [ ] `_can_use_tool` default-deny 拼 `decision_reason`（无字段时不抛异常）
- [ ] 新增 7 个单测全部通过
- [ ] `tests/agent_runtime/` 现有测试无回归
- [ ] `tests/agent_session_store/` 现有测试无回归
- [ ] ruff check / format 全绿
- [ ] agent_runtime 覆盖率不下降
- [ ] 端到端 SSE 流式回复跑通（手动）

不在范围：前端对 `api_error_status` 的展示、`include_hook_events` 集成等
（详见 spec "不在本范围内" 段）。
