构建定时仓库审查器：自动化代码审查指南

本教程指导开发者构建一个无人值守的定时代码审查代理，利用Claude Agent SDK的会话恢复功能实现跨运行连续性。通过只读工具、权限限制和预算控制确保安全，并演示如何通过cron调度、解析结构化输出及处理失败。适合需要定期审查代码仓库的工程实践。

技术背景

代码审查是保障代码质量的关键环节，但人工审查耗时且难以频繁执行。仓库在两次审查之间可能引入新问题，而定时审查器能自动扫描并报告发现。本教程针对需要记住上次审查状态的重复性任务（如依赖审计、文档新鲜度检查），利用会话恢复避免重复全量审查，节省成本并提高效率。

核心原理

核心原理是利用Claude Agent SDK的会话恢复功能，使代理在每次运行时携带上次审查的上下文。首次运行进行全量基线审查，后续运行恢复上次会话，从已有发现开始，仅报告增量变化。通过结构化输出（JSON Schema）确保结果可程序化处理，并通过权限模式、工具限制和预算控制实现安全无人值守运行。

架构分析

系统由审查代理、调度器（cron）、服务目录和日志组成。审查代理配置为只读工具（Read、Glob、Grep），权限模式dontAsk，并设置max_turns和max_budget_usd限制。服务目录存放脚本、环境变量和会话状态文件，与仓库分离。调度器通过cron定时触发脚本，脚本读取状态文件决定冷启动或恢复会话，输出结构化结果和可grep的日志行。

完整实战教程

简介

将重复的审查工作转化为一个无人值守的任务，由你掌控。即使是一个经过良好审查的仓库，在两次审查之间也会发生变化，而这些变化可能带来之前审查未发现的问题。本教程将这项工作交给一个代理（agent）来完成。

在你选择的计划时间，代理会扫描你的仓库，并返回一个程序可读的结论和发现。运行之间的连续性来自于会话恢复（resumed sessions），这是本教程核心的Agent SDK功能。恢复会话是定时审查器的一种设计选择。恢复的会话会携带整个前一次审查的内容，无需构建单独的存储。每次运行开始时，发现和审查者已读取的内容都已加载，而不是从空开始。*何时恢复、何时重新开始*一节将每种选择映射到适合的任务。

仓库在首次运行时进行一次完整的基线审查。之后的每次运行都恢复上一次运行的会话，并从代理已报告的发现开始。每个周期你阅读一份简短的后续报告，每份后续报告通过回显先前审查的发现来证明连续性。本笔记本手动运行前两个周期，最后将同一个审查器置于计划任务中。

你将学到什么

通过本教程，你将能够：

何时使用本教程

对于需要记住上一次审查的重复性审查任务，例如依赖审计或文档新鲜度检查，请使用本教程。你自己运行调度器，Claude Code提供读取工具、权限规则和可恢复会话。

如果你不想自己运行调度器或基础设施，以下托管选项可能更适合：

如果你想自己运行调度器和审查环境，请使用本教程。

先决条件

在继续本指南之前，请确保您具备：

所需知识：

* Python 基础知识：熟悉 async/await、函数和基本数据结构

* 对代理模式的基本理解：我们建议阅读 构建高效代理

* 对 Agent SDK 的 query() 函数的基本熟悉：如果您不熟悉该 SDK，请参阅 单行研究代理

所需工具：

* Python 3.11 或更高版本

* claude-agent-sdk 0.2.140 或更高版本，该版本添加了类型化的 ResultError，本配方在其失败路径中会捕获此错误

* Anthropic API 密钥（在此获取）

请从其自己的目录运行此笔记本，以便它写入的文件位于其旁边。

成本说明： 完整运行此笔记本的实时运行通常花费约 10 美分的 API 使用费，配套脚本的运行也类似。每次运行都会在其摘要行中打印其成本。

设置

首先，安装所需的依赖项：

%%capture
%pip install -U "claude-agent-sdk>=0.2.140" python-dotenv

注意： 确保你的 .env 文件包含以下内容：

ANTHROPIC_API_KEY=your_key_here

加载你的环境变量：

from dotenv import load_dotenv

_ = load_dotenv()

创建一个用于审查的示例仓库

创建一个带有已知缺陷的仓库，以便你可以检查审查者报告的内容。下面的代码在 demo_repo/ 中植入一个小型服务，供本笔记本中的审查使用。每个文件的源代码以字符串常量的形式出现，代码将其写入磁盘。这些文件从未被导入，因此缺陷保持惰性，最后的“清理”部分会在你完成时移除 demo_repo/。该服务的两个应用程序文件各携带一个真实缺陷：

* app/config.py 通过复制整个环境来构建其配置，然后打印结果。这两行代码会记录进程持有的每个环境变量，包括机密。

* app/math_utils.py 通过除以 len(values) 来计算平均值。如果传入一个空列表，它会引发 ZeroDivisionError。

from pathlib import Path

REPO_DIR = Path("demo_repo").resolve()

README_MD = """# sample-repo

A small service kept here so the scheduled reviewer has something to read.
"""

CONFIG_PY = """import os


def load_config(path):
    config = {"source": path}
    config["environment"] = dict(os.environ)
    print("loaded config:", config)
    return config
"""

MATH_UTILS_PY = """def divide(numerator, denominator):
    return numerator / denominator


def average(values):
    return divide(sum(values), len(values))
"""

SAMPLE_REPO_FILES = {
    "README.md": README_MD,
    "app/config.py": CONFIG_PY,
    "app/math_utils.py": MATH_UTILS_PY,
}

for relative_path, body in SAMPLE_REPO_FILES.items():
    target = REPO_DIR / relative_path
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(body, encoding="utf-8")

seeded = sorted(str(path.relative_to(REPO_DIR)) for path in REPO_DIR.rglob("*") if path.is_file())
print("sample repo:", seeded)

定义审查返回内容

示例仓库就绪后，下一部分是审查的答案契约。定时审查的每个消费者都是程序，因此审查者以 JSON 对象作为回复，按字段读取。output_format 接受 JSON 模式，回复会以符合该模式的结构出现在 ResultMessage.structured_output 中。每次审查都包含一个审查 ID、一个来自封闭集合的结论，以及每个都带有自身 ID 的发现项。

发现项 ID 驱动连续性证明。后续审查会引用前一次运行的 ID，表明审查者已将这些发现项向前推进。

from enum import StrEnum


class Verdict(StrEnum):
    """The two verdicts a scheduled review may report."""

    OK = "ok"
    CONCERNS = "concerns"


VERDICT_VALUES = [verdict.value for verdict in Verdict]

FINDING_SCHEMA = {
    "type": "object",
    "properties": {
        "id": {"type": "string"},
        "file": {"type": "string"},
        "summary": {"type": "string"},
    },
    "required": ["id", "file", "summary"],
}

FIRST_REVIEW_SCHEMA = {
    "type": "object",
    "properties": {
        "review_id": {"type": "string"},
        "verdict": {"type": "string", "enum": VERDICT_VALUES},
        "findings": {"type": "array", "items": FINDING_SCHEMA},
    },
    "required": ["review_id", "verdict", "findings"],
}

扩展后续审查的 schema

该配方运行两次审查，第二次需要一种方式指回第一次。previous_review_id 和 previous_finding_ids 指回第一次审查，而 resolved 列出不再适用的先前发现 ID。合并第一个 schema 将共享部分保留在一处，因此两个 schema 不会发生偏离。

previous_review_id 和 previous_finding_ids 承载连续性断言。当审查者将早期审查向前推进时，这些字段返回时持有第一次运行的审查 ID 和发现 ID。当审查者重新开始时，这些字段会说明这一点，你可以从该运行的日志行中得知。

CONTINUITY_PROPERTIES = {
    "previous_review_id": {"type": "string"},
    "previous_finding_ids": {"type": "array", "items": {"type": "string"}},
    "resolved": {"type": "array", "items": {"type": "string"}},
}

FOLLOW_UP_REVIEW_SCHEMA = FIRST_REVIEW_SCHEMA | {
    "properties": FIRST_REVIEW_SCHEMA["properties"] | CONTINUITY_PROPERTIES,
    "required": [*FIRST_REVIEW_SCHEMA["required"], *CONTINUITY_PROPERTIES.keys()],
}

print("follow-up required fields:", FOLLOW_UP_REVIEW_SCHEMA["required"])

配置有界、只读的审查者

定义好两个 schema 后，接下来是代理配置。这些选项适合无人值守运行的代理：

* tools：决定代理可用的内置工具。将列表限制为 Read、Glob 和 Grep，使审查者没有写入工具。

* allowed_tools：决定哪些工具调用无需权限提示即可运行。Read 授权限定在仓库路径内，在 dontAsk 模式下，超出该路径的读取会被作为工具错误拒绝。Glob 和 Grep 作为裸授权进入，批准整个工具，而 deny_reads_outside_repo 钩子则用于限制它们。运行摘要中的 StructuredOutput 条目来自 output_format 而非此列表，它仅携带回复。

* permission_mode="dontAsk"：从不提示，并拒绝任何未经预先批准的操作。auto 权限模式也无需常规提示即可运行，在运行时通过模型分类器批准或拒绝每个调用。此审查者使用 dontAsk 来保持其工具表面固定为允许规则。

* model：将审查固定到指定的模型。若不固定，SDK 将使用环境的默认模型，该模型可能随时间变化。固定模型可让计划任务的成本和审查行为保持在您设定的位置，直到您选择更改它们。

* strict_mcp_config=True：将会话限制为此配置声明的 MCP 服务器，即无服务器。若不设置，机器或账户上配置的 MCP 服务器将附加到会话，其工具定义会进入每个请求。

* max_turns 和 max_budget_usd：限制代理循环和花费。超过任一限制将以终端错误结果结束运行，即 error_max_turns 或 error_max_budget_usd，SDK 会将其作为 ResultError 抛出到您的代码中。一旦花费已超过预算，运行即停止，因此一个周期可能以超出上限结束。

* hooks：将 deny_reads_outside_repo 注册为 PreToolUse 回调。该回调会看到每个 Read、Grep 和 Glob 调用，并拒绝任何路径解析到仓库之外的调用，以及任何以 / 或 ~ 开头、包含 .. 段或包含 { 大括号的 Glob 模式。

* setting_sources=[]：使运行不读取机器的文件系统设置。这样，计划审查在您的笔记本电脑和运行 cron 任务的机器上行为一致。空列表还会使被审查仓库自身的 .claude/settings.json 和 CLAUDE.md 不进入会话。审查者的 cwd 是该仓库，而您不控制的仓库不应配置自己的审查者。

* output_format 和 resume：output_format 要求回复匹配您传入的 schema，resume 在运行之间传递连续性。冷启动时传入 None，在计划的后续运行中传入上一次运行的会话 ID。

这些选项实现的安全保障

上述选项集为无人值守运行代理实现了多项安全保障。只读工具授权、MCP 和设置隔离以及轮次和预算上限限制了运行可以执行的操作。计划的会话重置限制了会话上下文的持久时间。

读取被双重限定。allowed_tools 授权将 Read 限制在仓库路径内（在笔记本和脚本中均如此），并且读取仍遵循作业用户自身的文件访问权限。请使用访问权限仅限于审查所需内容的用户来执行部署步骤。

Claude Code 会尽力将 Read 规则应用于 Grep 和 Glob。deny_reads_outside_repo 使得读取限制不依赖于该覆盖范围也能成立。

配套脚本安装了相同的钩子，并且该钩子已经拒绝读取服务目录。脚本还添加了 disallowed_tools 规则作为独立的第二层，阻止 Read 工具访问该目录（环境变量和会话文件所在位置）。该规则仅在传入 service_dir 参数时设置，而笔记本中的调用未设置该参数。

同样的保护措施也用于防范提示注入，详见防范提示注入。当 tools 限制为 Read、Glob 和 Grep 时，会话没有 shell 和网络工具，审查者读取的内容除了回复本身无处可去。这就是为什么答案键在本笔记本中检查回复，以及为什么 review.log 与其审查的仓库受到同等对待的原因。

from typing import Any

from claude_agent_sdk import (
    AssistantMessage,
    ClaudeAgentOptions,
    HookContext,
    HookMatcher,
    ResultError,
    ResultMessage,
    TextBlock,
    ToolUseBlock,
    query,
)

READ_ONLY_TOOLS = ["Read", "Glob", "Grep"]
MODEL = "claude-sonnet-5"
FIRST_RUN_BUDGET_USD = 0.25
FOLLOW_UP_BUDGET_USD = 0.20
FIRST_RUN_MAX_TURNS = 20
FOLLOW_UP_MAX_TURNS = 14
TEXT_LOG_CHARS = 400


def clip(text: Any) -> str:
    """Flatten text onto one bounded log line, marking the cut when one happens.

    Findings arrive as free text and land in a log other tools grep. Dropping
    unprintable characters and collapsing whitespace keeps each field on one
    log line of its own.
    """
    printable = "".join(char if char.isprintable() else " " for char in str(text))
    flat = " ".join(printable.split())
    return flat if len(flat) <= TEXT_LOG_CHARS else flat[:TEXT_LOG_CHARS] + "..."


REVIEWER_SYSTEM_PROMPT = (
    "You are a repository review agent running on a schedule with nobody "
    "watching. Inspect the files with your read-only tools, keep the review "
    "short, and answer with the JSON object the caller's schema describes."
)


def deny_reads_outside_repo(repo: Path):
    """Deny any Read, Grep, or Glob call whose path resolves outside repo.

    The hook confines Grep and Glob deterministically instead of relying
    on Read-rule coverage of those tools. Candidate paths resolve
    symlinks, so a Read through an in-repository link is denied when it
    lands outside. Absolute Glob patterns, patterns carrying a `..` segment, and
    patterns carrying brace syntax the segment check can't reason about are
    denied rather than trusted to stay inside their search root.
    """

    def denial(reason: str) -> dict[str, Any]:
        return {
            "hookSpecificOutput": {
                "hookEventName": "PreToolUse",
                "permissionDecision": "deny",
                "permissionDecisionReason": reason,
            }
        }

    async def deny(
        input_data: dict[str, Any], tool_use_id: str | None, context: HookContext
    ) -> dict[str, Any]:
        tool_input = input_data.get("tool_input") or {}
        raw = tool_input.get("file_path") or tool_input.get("path")
        if raw:
            candidate = Path(raw).expanduser()
            if not candidate.is_absolute():
                candidate = repo / candidate
            candidate = candidate.resolve()
            if not candidate.is_relative_to(repo):
                return denial(f"path outside the repository under review: {candidate}")
        if input_data.get("tool_name") == "Glob":
            pattern = tool_input.get("pattern")
            if isinstance(pattern, str) and (
                pattern.startswith(("/", "~")) or "{" in pattern or ".." in pattern.split("/")
            ):
                return denial(f"glob pattern may leave the repository under review: {pattern}")
        return {}

    return deny


def review_options(
    repo: Path,
    schema: dict[str, Any],
    max_turns: int,
    resume_session_id: str | None = None,
    service_dir: Path | None = None,
) -> ClaudeAgentOptions:
    """Build the options for one scheduled review pass."""
    options = ClaudeAgentOptions(
        # The agent works inside the reviewed repository.
        cwd=str(repo),
        system_prompt=REVIEWER_SYSTEM_PROMPT,
        tools=READ_ONLY_TOOLS,
        # The //-anchored form is deliberate: with a single leading slash
        # the rule anchors at the settings source, not the filesystem
        # root, matches nothing, and every read is denied under dontAsk.
        # repo is absolute, so the rendered rule keeps the double slash.
        allowed_tools=[f"Read(/{repo}/**)", "Glob", "Grep"],
        permission_mode="dontAsk",
        model=MODEL,
        # Keep machine- and account-configured MCP servers out of the
        # session.
        strict_mcp_config=True,
        max_turns=max_turns,
        max_budget_usd=FOLLOW_UP_BUDGET_USD if resume_session_id else FIRST_RUN_BUDGET_USD,
        hooks={
            "PreToolUse": [
                HookMatcher(matcher="Read|Grep|Glob", hooks=[deny_reads_outside_repo(repo)])
            ]
        },
        # Same behavior on a laptop and on the box that runs the cron job.
        setting_sources=[],
        output_format={"type": "json_schema", "schema": schema},
        # None on a cold run, and the persisted session id on a follow-up.
        resume=resume_session_id,
    )
    if service_dir is not None:
        # A second, independent layer beside the hook, which already
        # denies these reads: block the Read tool on the service
        # directory, where the env and session files live.
        options.disallowed_tools = [f"Read(/{service_dir}/**)"]
    return options

使用回退读取回复

回复在到达时已经根据审查模式进行了验证。下面的两个读取器，verdict_of 和 finding_ids，在使用点对代码操作的两个字段 verdict 和 findings 重复该检查。每个读取器匹配其字段所需的形状，否则回退到固定的默认值。该单元格还定义了 RunOutcome（运行器填充的容器）和 string_items（连续性检查使用的列表读取器）。

缺失或无法识别的裁决读取为 concerns，即引起注意的状态。以任何其他形状到达的发现列表不会产生部分猜测，而是不产生任何 ID。代码最后对格式错误的输入执行两个默认值，因此下面的输出显示了回退。

from dataclasses import dataclass, field


@dataclass
class RunOutcome:
    """What one scheduled run reported back."""

    session_id: str | None = None
    subtype: str = "no_result"
    num_turns: int = 0
    total_cost_usd: float | None = None
    denials: int = 0
    payload: dict[str, Any] = field(default_factory=dict)
    tools_attempted: list[str] = field(default_factory=list)


def sequence_of(value: Any) -> list[Any]:
    """Return a list-shaped field's items, or nothing when it isn't one."""
    match value:
        case [*items]:
            return items
        case _:
            return []


def verdict_of(payload: dict[str, Any]) -> Verdict:
    """Read the verdict from the structured reply, falling back to concerns
    when the field is missing or unrecognized."""
    match payload:
        case {"verdict": str(value)} if value in VERDICT_VALUES:
            return Verdict(value)
        case _:
            return Verdict.CONCERNS


def finding_ids(payload: dict[str, Any]) -> list[str]:
    """Collect finding ids from the structured reply."""
    ids: list[str] = []
    for finding in sequence_of(payload.get("findings")):
        match finding:
            case {"id": str(finding_id)}:
                ids.append(finding_id)
    return ids


def string_items(value: Any) -> list[str]:
    """Collect the plain strings from a list-shaped field."""
    return [item for item in sequence_of(value) if isinstance(item, str)]


print("fallbacks:", verdict_of({}), finding_ids({"findings": "not-a-list"}))

构建审查运行器

模式、选项和读取器在运行器中汇聚。运行器流式传输消息，打印审查者的发言，列出审查者使用的工具，并保留最终 ResultMessage 的 session_id 和 structured_output。

每次运行结束时的报告会打印裁决，然后通过回复读取器逐行打印每条发现。逐块打印有上限，任何截断行上都有可见标记，该上限可防止一条长回复淹没计划日志。结构化输出运行通常产生很少的散文，运行可能会在工具调用前叙述一两行。在记录的运行中，两个周期仅流式传输工具调用。无论哪种方式，日志中的发现都来自回复的有效负载，而非流式文本。

这个相同的运行器作为 scheduled_review.py 与笔记本一同提供，这是你在将审查者部署到计划任务时使用的配套文件。

失败的运行会以非零状态退出，并且永远不会打印其完成行。两种失败类别都以这种方式结束：

* 超出限制：max_turns 或 max_budget_usd 以终端错误结果和非零 CLI 退出结束运行。SDK 将该结果作为携带结果子类型的 ResultError 异常暴露给你的代码。

* 无终端结果失败：在运行中途被终止的 CLI 进程或从未打开的连接会引发自己的异常类型，该类型会逃逸类型化捕获，因此不会打印失败行。

每次审查运行都会在其 run_review 调用周围捕获 ResultError，打印一行 REVIEW-CYCLE-INCOMPLETE，然后重新抛出。笔记本在失败时停止，而不是继续到成功行。在“证明失败路径”中诱导的运行会捕获 ResultError 而不重新抛出。触发失败是该单元格的重点。

async def run_review(
    *,
    repo: Path,
    label: str,
    prompt: str,
    schema: dict[str, Any],
    max_turns: int,
    resume_session_id: str | None = None,
    service_dir: Path | None = None,
) -> RunOutcome:
    """Run one review pass and return its structured result."""
    options = review_options(repo, schema, max_turns, resume_session_id, service_dir)
    outcome = RunOutcome()
    async for message in query(prompt=prompt, options=options):
        match message:
            case AssistantMessage(content=blocks):
                for block in blocks:
                    match block:
                        case TextBlock(text=text):
                            line = text.strip()
                            if line:
                                # Cap what a scheduled log absorbs per block.
                                print(f"[{label}] {clip(line)}")
                        case ToolUseBlock(name=tool_name):
                            outcome.tools_attempted.append(tool_name)
            case ResultMessage() as result:
                outcome.session_id = result.session_id
                outcome.subtype = result.subtype
                outcome.num_turns = result.num_turns
                outcome.total_cost_usd = result.total_cost_usd
                outcome.denials = len(result.permission_denials or [])
                if isinstance(result.structured_output, dict):
                    outcome.payload = result.structured_output
    return outcome


def report(label: str, outcome: RunOutcome) -> Verdict:
    """Print the scheduler-shaped lines for one run and return its verdict."""
    cost = f"{outcome.total_cost_usd:.4f}" if outcome.total_cost_usd is not None else "n/a"
    print(
        f"{label} session={outcome.session_id} subtype={outcome.subtype} "
        f"turns={outcome.num_turns} denials={outcome.denials} "
        f"cost_usd={cost} "
        f"tools_attempted={','.join(dict.fromkeys(outcome.tools_attempted)) or 'none'}"
    )
    verdict = verdict_of(outcome.payload)
    print(f"VERDICT: {verdict}")
    for finding in sequence_of(outcome.payload.get("findings")):
        match finding:
            case {"id": str(finding_id), "file": str(file_path), "summary": str(summary)}:
                print(f"  {clip(finding_id)} {clip(file_path)}: {clip(summary)}")
    return verdict

运行冷审查

两次实时运行从这里开始，作为计划中的前两个周期，间隔被压缩。第一次审查是冷运行。由于没有设置 resume，运行会读取整个仓库，并以首次审查的 schema 进行回答。运行以裁决、一组发现 ID 以及下一次运行恢复所需的 session_id 结束。

预期输出： 一个 RUN-1 摘要行，带有 subtype=success，然后是 VERDICT: concerns，其下包含覆盖两个植入缺陷的发现行。一个缺陷可能拆分为多个发现，但记录中的运行每个植入缺陷只打印一行。摘要行中的会话 ID 是运行 2 恢复时使用的。

摘要行中非零的 denials 计数表示读取限制在起作用。钩子会拒绝以根目录为锚点的 Read 尝试，例如 /app/config.py，而审查者会使用仓库内的路径重试。使用花括号展开的 Glob 模式，例如 **/*.{py,md}，也会计入同一计数。运行是否显示拒绝取决于审查者首先尝试的路径和模式。

拒绝会在轮次预算内消耗重试次数。如果你的仓库审查依赖花括号模式，请调整钩子以展开这些模式，并对每个展开结果运行相同的检查，而不是直接拒绝该模式。

import uuid

FIRST_REVIEW_ID = f"review-{uuid.uuid4().hex[:8]}"
FIRST_PROMPT = (
    f"Scheduled review {FIRST_REVIEW_ID} of the repository in the current "
    "directory. "
    "Read the source files and report correctness or security problems you can "
    "point at a specific file and function. Give each finding a summary of "
    "one or two sentences and an id of the form "
    'F1, F2, and so on. Use the verdict "concerns" when you report at least one '
    'finding and "ok" when the repository looks clean. Echo '
    f"{FIRST_REVIEW_ID} back as review_id."
)

try:
    first = await run_review(
        repo=REPO_DIR,
        label="run-1",
        prompt=FIRST_PROMPT,
        schema=FIRST_REVIEW_SCHEMA,
        max_turns=FIRST_RUN_MAX_TURNS,
    )
except ResultError as exc:
    cost = (exc.data or {}).get("total_cost_usd")
    cost_str = f"{cost:.4f}" if isinstance(cost, (int, float)) else "n/a"
    print(
        f"REVIEW-CYCLE-INCOMPLETE stage=run-1 subtype={exc.subtype} "
        f"reason={exc.terminal_reason} cost_usd={cost_str}"
    )
    raise

if first.subtype != "success" or first.session_id is None or not first.payload:
    print(f"REVIEW-CYCLE-INCOMPLETE stage=run-1 subtype={first.subtype}")
    raise RuntimeError("run 1 did not complete; there is no session to resume")
first_verdict = report("RUN-1", first)

在周期之间更改仓库

在部署中，周期之间落地的更改要么修复了审阅者报告的问题，要么引入了审阅者尚未见过的问题。下面的代码在后续运行之前各做了一处更改。

该代码修复了 app/math_utils.py 中的零值保护，这是应该返回为已解决的发现。它植入了一个新文件 app/retry.py，其中的重试循环没有限制，这是后续应该作为新问题捕获的问题。新问题位于运行 1 从未读取的文件中。捕获它证明了后续查看的是当前状态的仓库，而不是根据它已经知道的文件来回答。

MATH_UTILS_FIXED = """def divide(numerator, denominator):
    if denominator == 0:
        raise ValueError("denominator must be non-zero")
    return numerator / denominator


def average(values):
    if not values:
        return 0.0
    return divide(sum(values), len(values))
"""

RETRY_PY = """import time


def fetch_with_retry(fetch):
    while True:
        try:
            return fetch()
        except ConnectionError:
            time.sleep(1)
"""

(REPO_DIR / "app/math_utils.py").write_text(MATH_UTILS_FIXED, encoding="utf-8")
(REPO_DIR / "app/retry.py").write_text(RETRY_PY, encoding="utf-8")
changed = sorted(str(path.relative_to(REPO_DIR)) for path in REPO_DIR.rglob("*") if path.is_file())
print("repository now:", changed)

运行恢复的审查

下一次计划运行将第一次运行的 session_id 作为 resume 传递，并请求后续模式。提示的第一条指令是再次列出仓库的文件。如果没有这一步，恢复的代理可以根据第一次运行中读取的文件进行回答，而永远不会注意到 app/retry.py 是新的。下面的代码检查回复与第一次运行的一致性，并打印一行 RESUME-LINK，其中包含三个字段的连续性证明：

* same_session：两次运行共享一个会话，以表明恢复本身有效

* prior_review_id_echoed：后续回复提到了第一次审查的 ID

* recalled_findings：后续回复携带了多少第一次运行的发现 ID

当这三个条件都成立时，审查者根据已经完成的审查进行回答。当三个条件之一失败时，代码会在单独一行打印 RESUME-LINK-BROKEN 标记。回复还列出了 resolved，即不再适用的先前发现 ID。

预期输出：

* 一行 RESUME-LINK，其中 same_session=True，prior_review_id_echoed=True，以及完整的 recalled_findings 计数

* 一个 resolved 列表，包含零保护发现 ID

* 保留配置泄漏并添加新的重试错误的发现

* REVIEW-CYCLE-COMPLETE 行以结束循环

SECOND_REVIEW_ID = f"review-{uuid.uuid4().hex[:8]}"
FOLLOW_UP_PROMPT = (
    f"Scheduled review {SECOND_REVIEW_ID} of the same repository, one cycle "
    "later. List the repository's files again, then read the source files as "
    "they stand now. Answer from the most recent review you did in this "
    "conversation: put its review id in previous_review_id and the ids of the "
    "findings it reported in previous_finding_ids. Report the findings that "
    "still apply, keeping the ids you gave them earlier, list in resolved the "
    "ids of previous findings that no longer apply, and report anything newly "
    "wrong as a new finding with a summary of one or two sentences. Echo "
    f"{SECOND_REVIEW_ID} back as review_id."
)

try:
    second = await run_review(
        repo=REPO_DIR,
        label="run-2",
        prompt=FOLLOW_UP_PROMPT,
        schema=FOLLOW_UP_REVIEW_SCHEMA,
        max_turns=FOLLOW_UP_MAX_TURNS,
        resume_session_id=first.session_id,
    )
except ResultError as exc:
    cost = (exc.data or {}).get("total_cost_usd")
    cost_str = f"{cost:.4f}" if isinstance(cost, (int, float)) else "n/a"
    print(
        f"REVIEW-CYCLE-INCOMPLETE stage=run-2 subtype={exc.subtype} "
        f"reason={exc.terminal_reason} cost_usd={cost_str}"
    )
    raise

if second.subtype != "success" or second.session_id is None or not second.payload:
    print(f"REVIEW-CYCLE-INCOMPLETE stage=run-2 subtype={second.subtype}")
    raise RuntimeError("run 2 did not complete")
second_verdict = report("RUN-2", second)

prior_ids = finding_ids(first.payload)
recalled_ids = string_items(second.payload.get("previous_finding_ids"))
matched = sorted(set(recalled_ids) & set(prior_ids))
echoed = second.payload.get("previous_review_id") == FIRST_REVIEW_ID
print(
    f"RESUME-LINK run1_session={first.session_id} "
    f"run2_session={second.session_id} "
    f"same_session={second.session_id == first.session_id} "
    f"prior_review_id_echoed={echoed} "
    f"recalled_findings={len(matched)}/{len(set(prior_ids))}"
)
print(f"resolved={[clip(i) for i in string_items(second.payload.get('resolved'))]}")
if second.session_id != first.session_id or not echoed or (prior_ids and not matched):
    # Prints on a completed run by design, mirroring the script's marker.
    print(
        f"RESUME-LINK-BROKEN same_session={second.session_id == first.session_id} "
        f"recalled_findings={len(matched)}/{len(set(prior_ids))}"
    )
print(f"REVIEW-CYCLE-COMPLETE runs=2 verdicts={first_verdict},{second_verdict}")

对照植入的变更检查发现

仓库中的缺陷和周期间的变更是预先已知的，因此它们可以作为两次运行的答案键。

检查运行1的报告

运行1至少报告了两个植入的缺陷：

* app/config.py：加载器将整个环境复制到配置中并打印出来

* app/math_utils.py：divide没有零值保护，因此average([])会引发ZeroDivisionError

config["environment"] = dict(os.environ)是一行看似无害的代码。泄漏之所以存在，是因为下一行的print将其发布出去，而环境变量正是部署中保存机密的地方。报告它意味着审查者跨文件连接了这三个事实。零值保护测试仅检查审查者是否仔细阅读。

检查后续运行的增量

后续运行的报告将周期间的变更分为三个部分：

* 配置泄漏保留在findings中，使用其先前的ID

* 已修复的零值保护的ID移至resolved

* app/retry.py中的无界重试作为新发现出现

prior_review_id_echoed=True表示恢复的审查者命名了它在运行1中获得的ID，而recalled_findings统计了运行1中返回的发现ID数量。

预期运行间存在差异

模型输出在不同运行间会有所不同。审查者可能：

* 以不同的措辞描述发现

* 分配不同的ID

* 将一个植入的缺陷拆分为多个发现（零值保护可能作为一个发现出现，或作为单独的divide和average发现）

* 报告额外的低严重性观察

resolved中的ID遵循变更修复的内容，无论审查者如何拆分。无论措辞如何，正确的第一次运行会报告两个植入的缺陷，正确的后续运行会报告三部分增量，而RESUME-LINK字段携带连续性证明。

更改demo_repo/中的其他内容，并再次运行后续运行以观察增量的变化。笔记本将其基线固定到运行1，重复的后续运行会报告prior_review_id_echoed=False并打印RESUME-LINK-BROKEN标记，即使恢复工作正常。配套脚本在每次运行后保存新的基线，那里的重复后续运行会保持其链接完整。

证明失败路径

runner 部分描述了失败运行的样子。下面的单元格故意触发一次失败，对同一仓库在 max_turns=1 下运行新的审查，这个上限无法容纳一次完整的审查。运行以终端错误结果结束，SDK 将其作为 ResultError 抛出，except 块打印携带结果 subtype 和 terminal_reason 的失败行。

预期输出： 一行 REVIEW-CYCLE-INCOMPLETE，带有 stage=induced subtype=error_max_turns reason=max_turns 以及失败运行消耗的成本，并且没有完成行。

try:
    await run_review(
        repo=REPO_DIR,
        label="induced",
        prompt=FIRST_PROMPT,
        schema=FIRST_REVIEW_SCHEMA,
        max_turns=1,
    )
except ResultError as exc:
    cost = (exc.data or {}).get("total_cost_usd")
    cost_str = f"{cost:.4f}" if isinstance(cost, (int, float)) else "n/a"
    print(
        f"REVIEW-CYCLE-INCOMPLETE stage=induced subtype={exc.subtype} "
        f"reason={exc.terminal_reason} cost_usd={cost_str}"
    )
else:
    raise RuntimeError("the run completed under max_turns=1")

为审查者设定计划

上面的两个周期是手动运行的。在计划中，每个周期是 scheduled_review.py 的一次调用，它是本笔记本旁边的配套脚本。它从自己的服务目录运行，将要审查的仓库作为参数，并携带相同的模式和回复读取器。

脚本的标记行写的是 REVIEW-RUN，而笔记本的周期写的是 REVIEW-CYCLE，笔记本的标记永远不会出现在计划日志中。当服务目录中没有会话文件 .last_review_session 时，脚本会冷启动审查；当文件存在时，脚本会恢复。它的 read_state 函数承担了笔记本中变量所代表的持久化。

在安排计划之前，先通读一遍脚本。

在持久的主机上运行审查者。给任务分配一个用户，其文件访问权限仅限于审查所需，并以该用户身份执行以下每个步骤。恢复依赖于两个本地文件在周期之间存活：

* .last_review_session，脚本保存在服务目录中自己旁边的会话文件

* SDK 在 ~/.claude/projects/ 下写入的会话记录，或者当您设置 CLAUDE_CONFIG_DIR 变量时，在 CLAUDE_CONFIG_DIR 的 projects/ 目录下写入的会话记录

将服务目录放在您要审查的仓库之外。这样布局，干净的检出或分支切换不会删除会话文件，而且该文件永远不会出现在审查者读取的文件中。从本配方自己的目录进行试运行，python scheduled_review.py demo_repo，会在启动时被拒绝，并显示 reason=repository-inside-service-directory 的用法行。服务目录位于仓库内部也会以同样的方式被拒绝，显示 reason=service-directory-inside-repository 的用法行。

安排脚本

  1. 创建服务目录及其自己的环境。笔记本的 %pip 安装只影响笔记本自己的环境，而 cron 运行命令中指定的任何解释器。服务目录带有自己的虚拟环境，并安装了 SDK：
sudo mkdir -p /srv/reviewer && sudo chown $USER /srv/reviewer && cd /srv/reviewer
python3 -m venv .venv
.venv/bin/pip install "claude-agent-sdk>=0.2.140"
cp /path/to/scheduled_review.py .

使用 3.11 或更高版本的 python3，这是先决条件中指定的最低版本，因为较旧的解释器会在脚本的第一次导入时失败。

  1. 创建 cron 作业引用的环境文件，位于 /srv/reviewer/reviewer.env。调度程序必须提供环境，因为 cron 启动时几乎没有环境。使用 export 行，因为普通的 KEY=value 赋值设置的是 shell 变量，Python 进程永远不会继承。Cron 会将 PATH 削减到最低限度。将其设置为覆盖作业所需二进制文件所在的目录：
# /srv/reviewer/reviewer.env，由 cron 作业引用
export PATH=/usr/local/bin:$PATH
export ANTHROPIC_API_KEY=your_key_here

在放入密钥之前，先创建空文件并限制权限，使用 touch /srv/reviewer/reviewer.env && chmod 600 /srv/reviewer/reviewer.env，然后添加上面的行。默认情况下，脚本还拒绝自己对服务目录的 Read 访问，使密钥文件不会出现在任何发现中。同样的拒绝会使服务目录内任何仓库的审查失效。脚本会直接拒绝这种布局。将密钥保留在 crontab 行之外，因为任何能列出 crontab 的人都能读取它。

  1. 在服务目录下手动运行一次脚本，指向你想要审查的仓库：
cd /srv/reviewer && . ./reviewer.env && .venv/bin/python scheduled_review.py /path/to/your/repo

首次运行是冷启动，会打印 REVIEW-RUN-COMPLETE: cold，并创建 .last_review_session 文件和会话记录，后续每次定时运行都会基于该记录继续。如果运行以 error_max_budget_usd 退出并返回 1，请在调度前提高 FIRST_RUN_BUDGET_USD。冷启动无法在预算内完成就无法建立基线，之后每个定时周期都会重复同样的失败。

  1. 使用 crontab -e 添加一个 crontab 条目，用于每晚 02:00 进行审查。调度用户应与步骤 3 中运行的用户相同，因为会话记录存储在该用户的主目录下：
0 2 * * * cd /srv/reviewer && . ./reviewer.env && flock -n -E 99 .review.lock .venv/bin/python scheduled_review.py /path/to/your/repo >> /srv/reviewer/review.log 2>&1 || echo "REVIEW-RUN-EXIT: $?" >> /srv/reviewer/review.log

该条目的尾部将失败调用的退出状态转换为日志行。两个重定向都使用绝对路径指定日志文件，标记永远不会落在相对于 cron 启动目录的文件中。

flock -n 会在前一个调用仍持有锁时跳过本次调用，-E 99 为跳过提供独立的退出码。被跳过的运行会记录 REVIEW-RUN-EXIT: 99，绝不会伪装成失败。flock 来自 Linux 主机上的 util-linux。macOS 默认没有 flock，Homebrew 的 flock formula 提供了一个。

发现会引用仓库内容，因此请像对待被审查的仓库一样处理 review.log。像轮换其他服务日志一样轮换该日志。下一节的 REVIEW-RUN-EXIT 行展示了如何读取它。

读取输出行

你的调度器读取这些行：

| 行 | 触发时机 | 退出码 | 告警 |

|---|---|---|---|

| VERDICT: ok 或 VERDICT: concerns | 完成的审查报告结构化回复的结论。失败的运行从不打印 | 0 | 对 concerns 告警 |

| RESUME-LINK ... | 每次恢复的运行。same_session=True 且 recalled_findings 数量稳定表示连续性正常工作，current_findings 统计后续自身的发现 | 0 | 在自有告警中设置 recalled_findings 阈值。逐渐减少的数量不打印标记，是“按计划重置会话”一节描述的重置信号 |

| RESUME-LINK-BROKEN ... | 链接完全断开：会话丢失、回显丢失或召回归零。标记位于行首，便于 grep | 0 | 告警。运行已为下一周期保存新基线 |

| REVIEW-RUN-COMPLETE: cold 或 : resumed | 审查可验证地成功 | 0 | 无。其存在表示周期已闭合 |

| REVIEW-RUN-INCOMPLETE ... | 运行以终端错误结果结束，最常见的是超出边界。subtype 指明结束原因；边界和模式失败添加 reason。会话崩溃后，该行可能打印 cost_usd=0.0000，表示成本未知而非零 | 1 | 告警。边界失败保留会话文件，修复方式是配置更改 |

| REVIEW-RUN-INCOMPLETE ... subtype=error_max_structured_output_retries | 回复反复未通过模式验证，或模型回退撤回已完成的回复且无重试可替换。两者均非会话问题 | 1 | 告警。会话文件保留 |

| REVIEW-RUN-INCOMPLETE stage=usage ... | 调用本身错误，如缺少仓库参数 | 2 | 告警。修复 crontab 条目 |

| SESSION-FILE-CLEARED: next run reviews cold | 恢复运行失败，subtype=error_during_execution，最常见的是保存的会话无法恢复；瞬时连接失败也归于此，以一次冷重建换取自愈 | 1 | 除失败告警外无其他。下一周期冷重建基线 |

| SESSION-STATE-NOT-SAVED ... | 审查完成但状态无法写入，最常见的是磁盘已满或只读。标记上方的发现有效。后续的 SESSION-FILE-CLEARED 表示下一运行冷重建；SESSION-FILE-STALE 表示将恢复上一会话 | 1 | 告警。修复磁盘；审查本身无需重跑 |

| REVIEW-RUN-EXIT: <code> | crontab 条目的尾部记录非零退出，包括跳过的重叠运行留下的 99 | 不适用 | 对除 99 外的任何代码告警 |

| 无任何行 | 运行在终端结果前死亡，如进程被杀或连接从未打开 | 非零 | 对缺失完成行且无 99 跳过解释时告警 |

诱导的 max_turns=1 运行打印以 REVIEW-RUN-INCOMPLETE stage=cold subtype=error_max_turns reason=max_turns 开头的行，该前缀用于 grep，行尾的 cost_usd 补全信息。“验证失败路径”一节在笔记本中触发了相同的终端结果。

一个成功的周期会在日志中留下 REVIEW-RUN-COMPLETE，而失败的周期会留下 REVIEW-RUN-EXIT 标记。对任何非 99 的 REVIEW-RUN-EXIT 代码发出警报，或者对缺少完成行且没有解释跳过的 99 的周期发出警报。跳过是正常的重叠，重复跳过意味着运行超出了计划时间。将 grep 锚定在每个标记行的开头。查找文本被裁剪到单个缩进行上，因此标记只会出现在其自己的行首。stage=resumed 失败后跟 SESSION-FILE-CLEARED 行意味着保存的会话不再恢复，最常见的原因是 ~/.claude/projects/ 下的记录已丢失。

自定义审查器

* 提示词：将后续提示指向审查应跟踪的内容，例如新依赖项、破坏性 API 变更或 TODO 债务。

* 模式：添加你的警报所需的字段，例如每个发现的严重性。回复会根据你声明的任何内容进行验证。脚本自带模式和读取器的副本。自定义任一内容时，请同时更改两者。

* 边界：max_turns 和 max_budget_usd 位于脚本顶部。本笔记本单元格中的两个边界均针对三文件示例仓库设置，并高于记录运行自身的 cost_usd 数值。脚本的默认值为真实仓库留有余量。预算是失控时的后备机制，像轮次上限一样分配。基线审查在 FIRST_RUN_BUDGET_USD 下运行，每次后续审查在 FOLLOW_UP_BUDGET_USD 下运行。后续上限是计划任务倍增的那个。在默认设置下，夜间作业的最差月份约为 30 × $0.50 = $15，外加偶尔的冷重建，而记录运行每次仅花费几美分。观察摘要行中的 cost_usd 数值几个周期，以找到适合的上限。该数值是客户端估算值，而非计费数据。跟踪成本和使用情况 涵盖了差异。effort 选项 是另一个成本杠杆。文档建议对仅读取文件的代理使用 "low"，而此审查器正是如此。

* 模型：交换 MODEL 别名以在成本与能力之间权衡。固定设置可确保更改是有意的。

按计划重置会话

长时间运行的会话会按设计积累上下文。该上下文也会增加成本，并可能使审查器固守旧结论。按固定计划（每周或每当仓库结构发生变化时）删除脚本旁边的会话文件，下一次运行将冷启动并重建基线。

后续审查的 recalled_findings 计数是操作信号。当计数下降时，会话已退化，是时候重置了。

何时恢复会话，何时重新开始

计划中的审查任务要么恢复上一次运行的会话，要么每个周期重新开始，要么在重新开始时输入先前的结果。下表将每种选择映射到其适合的任务：

| 选择 | 适用场景 |

|---|---|

| 每次调用都使用新会话 | 每个工作单元独立评判。用于门控拉取请求的 CI 审查器需要这种模式：没有旧观点或携带的上下文影响当前判断。 |

| 使用新会话并输入先前的结果 | 任务只需要上次运行报告的内容。用于跟踪 TODO 债务的追踪器，报告上周哪些项目仍未完成，适合这种模式：回复中的发现 ID 携带了比较所需的一切，每个周期都重新评判代码，且没有保存的会话会在周期之间过期或丢失。 |

| 恢复会话 | 任务需要与上次运行看到的所有内容进行比较。用于夜间依赖审计，只标记自上次运行以来新增或更改的包，需要检查审查器读取的完整清单，而该清单的大部分从未进入回复。自行向前传递意味着序列化审查器观察到的所有内容，这正是恢复会话所节省的工作。 |

清理

笔记本向 demo_repo/ 写入了四个文件：两个周期的夹具以及周期之间的变更。下面的代码在检查目录包含夹具后将其删除。若要首先从答案键运行增量实验，请暂缓此清理，并在稍后需要这些文件时再次运行示例仓库代码。

如果你还运行了配套脚本，请同时从其服务目录中删除 .last_review_session 文件。运行还会在 ~/.claude/projects/ 下写入会话记录。删除那里匹配的项目文件夹以移除它们。

import shutil

config_path = REPO_DIR / "app" / "config.py"
if config_path.exists() and config_path.read_text(encoding="utf-8") == CONFIG_PY:
    shutil.rmtree(REPO_DIR, ignore_errors=True)
    print("removed", REPO_DIR.name)
else:
    print("left", REPO_DIR.name, "in place: it does not hold the seeded fixture")

你学到了什么

* 限制无人值守的代理：tools 和 allowed_tools 决定哪些工具存在以及哪些工具可以在无人提示的情况下运行，而 max_turns 和 max_budget_usd 则限制循环次数和花费，因此计划任务只能在您选择的范围内出错。

* 证明连续性：恢复先前会话的代理会记住它已经发现的内容。让每次运行都通过在其回复中回显先前运行的发现来证明这一点。这样，一旦发生中断，日志中就会立即显示出来。

* 让审查者按计划运行：无人值守运行的审查会向程序报告，因此其输出是退出代码和调度程序可以 grep 的固定行。对于此审查者，这意味着一个为仓库负责的裁决行，一个仅在成功运行后打印的完成行，以及一个标记每次失败的非零退出码。

了解更多

* 从代理获取结构化输出：本配方回复中使用的 output_format 模式和 structured_output 字段的完整文档

* 使用会话：除了本配方使用的恢复功能之外的会话生命周期，包括将会话分叉到两条路径

* 配置权限：本配方中 dontAsk 背后的权限模式模型，以及为交互式使用而构建的模式

* 安全性：Claude Code 的安全保障，包括处理不受信任内容的最佳实践

* 跟踪成本和使用情况：max_budget_usd 背后的成本表面以及摘要行中的成本数字

常见问题

常见错误包括：会话文件丢失导致恢复失败（打印SESSION-FILE-CLEARED）、预算超限（error_max_budget_usd）、工具权限拒绝（denials计数增加）、输出格式验证失败（error_max_structured_output_retries）。排查方法：检查日志中的REVIEW-RUN-INCOMPLETE行，确认subtype和reason；确保服务目录权限正确；调整预算和轮次上限；验证钩子函数正确拒绝外部路径。

企业应用扩展

生产环境优化：使用flock防止重叠运行，设置环境文件权限600，定期轮换日志，监控recalled_findings数量以决定重置会话。可自定义提示词、JSON Schema和模型以适配不同审查需求。考虑使用托管方案（如Claude Code routines或Managed Agents）以减少基础设施维护。

总结

本教程指导开发者构建一个无人值守的定时代码审查代理，利用Claude Agent SDK的会话恢复功能实现跨运行连续性。通过只读工具、权限限制和预算控制确保安全，并演示如何通过cron调度、解析结构化输出及处理失败。适合需要定期审查代码仓库的工程实践。

内容来源与验证

继续系统学习

站内还有更多完整实战教程

加入会员，解锁全部教程、完整代码与持续更新内容。

加入会员
查看全部AI 智能体教程