diff --git a/examples/skills_code_review_agent/.env.example b/examples/skills_code_review_agent/.env.example new file mode 100644 index 00000000..cf4b0d32 --- /dev/null +++ b/examples/skills_code_review_agent/.env.example @@ -0,0 +1,32 @@ +# Code Review Agent — LLM 配置样例（复制为 .env 后按需修改） +# +# 所有变量均为可选；缺省即降级为 FakeLlm（不调用任何模型）。 +# 支持任意 OpenAI 兼容端点：OpenAI / Azure OpenAI / 本地 vLLM / 内部 LLM 网关。 + +# 是否启用真实 LLM 二次研判（也可通过 CLI --enable-llm 覆盖） +# 取值：true | false +LLM_ENABLED=false + +# 供应商标识（仅用于记录/诊断，不影响端点选择） +LLM_PROVIDER=openai + +# 真实模型 API Key；留空 = 禁用真实模型 +LLM_API_KEY= + +# OpenAI 兼容的 Chat Completions 端点基址 +# OpenAI: https://api.openai.com/v1 +# Azure: https://.openai.azure.com/openai +# 本地 vLLM: http://localhost:8000/v1 +LLM_BASE_URL=https://api.openai.com/v1 + +# 模型名（对应端点上的 deployment / model id） +LLM_MODEL=gpt-4o-mini + +# 采样温度（越低越确定，适合判定任务） +LLM_TEMPERATURE=0.1 + +# 单次响应最大 token 数 +LLM_MAX_TOKENS=1024 + +# 请求超时（秒） +LLM_TIMEOUT=30 diff --git a/examples/skills_code_review_agent/README.md b/examples/skills_code_review_agent/README.md new file mode 100644 index 00000000..106190df --- /dev/null +++ b/examples/skills_code_review_agent/README.md @@ -0,0 +1,424 @@ +# Code Review Agent + +## 快速运行 + +```powershell +# 在仓库根目录执行 +$env:PYTHONPATH = "examples/skills_code_review_agent" +python examples/skills_code_review_agent/run_agent.py --fixture security --dry-run --output-dir examples/skills_code_review_agent/out --db-path examples/skills_code_review_agent/cr_agent.db + +# 使用 diff 文件 +python examples/skills_code_review_agent/run_agent.py --diff-file path/to/change.diff --mode dry-run --output-dir out --db-path cr_agent.db + +# 审查本地 git 工作区变更 +python examples/skills_code_review_agent/run_agent.py --repo-path path/to/repo --mode dry-run --output-dir out --db-path cr_agent.db +``` + +## 常用参数 + +| 参数 | 说明 | +| --- | --- | +| `--diff-file PATH` | 读取 unified diff / PR patch 文件。 | +| `--repo-path PATH` | 在指定 git 仓库中读取工作区变更。 | +| `--fixture NAME` | 使用内置 fixture，便于 dry-run 和测试。 | +| `--skill-dir PATH` | 指定 code-review Skill 目录，默认 `skills/code-review`。 | +| `--mode dry-run\|real` | `dry-run` 使用 fake runner；`real` 走沙箱 runtime。 | +| `--dry-run` | `--mode dry-run` 的快捷写法。 | +| `--output-dir DIR` | 输出 `review_report.json` 和 `review_report.md`。 | +| `--db-path PATH` | SQLite 数据库路径。 | +| `--require-sandbox` | real 模式下强制使用声明的沙箱，不可用则失败。 | +| `--telemetry` | 输出 telemetry span，便于审计链路。 | +| `--enable-llm` | 可选开启真实 LLM 二次研判；无 Key 时默认不需要。 | + +--- +# CR Agent 架构设计 + +> 自动代码评审 Agent 原型 — 基于 tRPC-Agent Skill 体系 +> 本文是**整体架构设计**,不含完整实现代码,但给出可落地的契约:目录结构、建表 DDL、SKILL.md frontmatter、规则清单、监控字段、去重算法与验收标准逐条对照。 + +--- + +## 1. 概述 + +目标不是"让 LLM 评论代码",而是把 **Skills、沙箱执行、数据库、Filter 治理、审查规则、结果结构化、监控审计、安全边界** 串成一个**可验证系统**。输入 git diff / PR patch / 本地变更目录,输出结构化 findings,并把审查任务、拦截记录、监控摘要、结果写入数据库,支持后续评测、监控、回放。 + +系统采用**六层流水线**架构,数据自上而下流动,其中 Filter 治理与沙箱执行构成安全治理层,监控审计贯穿全链路。 + +--- + +## 2. 目录结构 + +``` +examples/skills_code_review_agent/ +├── ARCHITECTURE.md # 本文档 +├── README.md # 使用说明 +├── agent.py # Agent 入口(CLI 编排) +├── skills/ +│ └── code-review/ +│ ├── SKILL.md # Skill 契约:入口/规则/沙箱策略 +│ ├── rules/ # 6 类规则文档 +│ │ ├── security.md +│ │ ├── async_errors.md +│ │ ├── resource_leak.md +│ │ ├── missing_tests.md +│ │ ├── sensitive_info.md +│ │ └── db_lifecycle.md +│ └── scripts/ # 沙箱内可执行脚本 +│ ├── parse_diff.py # unified diff 解析 → ChangeSet +│ ├── run_checks.py # 规则匹配 → 原始诊断 +│ ├── dedupe.py # 去重 + 置信度分流 +│ └── mask_secrets.py # 敏感信息脱敏 +├── db/ +│ ├── schema.sql # 建表 DDL +│ ├── storage.py # 存储接口(SQLite 默认,可切 SQL 后端) +│ └── init_db.py # 初始化/迁移脚本 +├── sandbox/ +│ ├── runtime.py # runtime 抽象:local/container/cube +│ └── policy.py # 安全边界:超时/输出/env/脱敏/失败 +├── filters/ +│ └── governance.py # Filter 治理:四类检查 + 决策 +├── tests/ +│ ├── fixtures/ # 8 条 diff 样本 +│ │ ├── 01_clean.diff +│ │ ├── 02_security.diff +│ │ ├── 03_async_leak.diff +│ │ ├── 04_db_lifecycle.diff +│ │ ├── 05_missing_tests.diff +│ │ ├── 06_duplicate.diff +│ │ ├── 07_sandbox_fail.diff +│ │ └── 08_sensitive_info.diff +│ └── test_cr_agent.py # 链路测试 +└── examples/ + └── review_report.json # 示例报告输出 +``` + +--- + +## 3. 分层架构 + +系统分为六层,数据流自上而下。数据处理层(蓝)负责把 diff 变成结构化结论,安全治理层(珊瑚)在风险点与执行边界上把关,底部监控审计带贯穿全链路。 + +| 层 | 名称 | 职责 | 关键产出 | +|----|------|------|----------| +| L1 | 输入解析 | 接收 `--diff-file` / `--repo-path` / fixture,解析 unified diff | `ChangeSet`(文件·hunk·行号) | +| L2 | Skill 加载 | `skill_load` 加载 SKILL.md + 6 类规则 + 脚本目录 | `RuleSet` + 脚本清单 | +| L3 | Filter 治理 | 脚本前置决策:高风险/禁止路径/非白名单网络/超预算 | `allow` / `deny` / `needs_human_review` | +| L4 | 沙箱执行 | Container/Cube·E2B(默认)或本地 fallback 跑检查 | 原始诊断(已脱敏) | +| L5 | 去重结构化 | 行级去重 + 置信度分流 + 9 字段组装 | `findings` / `warnings` / `needs_human_review` | +| L6 | 存储输出 | 写 SQLite,生成 report.json + report.md | 数据库记录 + 双格式报告 | + +**监控审计(贯穿)**:总耗时、沙箱耗时、工具调用次数、拦截次数、finding 数量、severity 分布、异常类型分布。 + +--- + +## 4. 评审数据流 + +一次完整 review 的主链(9 步),含两条异常分支: + +``` +CLI 输入 → 解析 diff → skill_load → [Filter 决策] → 沙箱执行 → 去重降噪 → 组装 findings → 落库 → 生成报告 + │ │ + deny/review ─┘ └ 沙箱失败 → 降级空诊断 + │ │ + └──────────→ 都汇入去重降噪层 ←──────┘ +``` + +- **Filter 分支**:`deny` / `needs_human_review` 时记录拦截原因,跳过沙箱,直接进入去重层(无诊断)。 +- **沙箱失败分支**:超时/崩溃时降级为空诊断,记录失败,继续评审——**异常不崩任务**。 +- **dry-run 模式**:L4 替换为本地 fake runner(直接跑规则脚本,不调真实模型 API),其余链路完全一致,无 API Key 也可验证解析→落库→报告全链路。 + +--- + +## 5. 数据库 Schema + +七张表围绕 `review_task` 聚合,`task_id` 为全局外键。SQLite 默认实现,`storage.py` 接口保留切换 SQL 后端的空间(只暴露 `ReviewStore` 抽象,底层连接可替换)。 + +### 5.1 表关系 + +| 主表 | 关系 | 子表 | 语义 | +|------|------|------|------| +| review_task | 1 — N | input_diff | 一个任务可含多个变更文件 | +| review_task | 1 — N | sandbox_run | 一个任务可跑多次沙箱 | +| review_task | 1 — N | finding | 一个任务产出多条 finding | +| review_task | 1 — N | filter_block | 一个任务可有多条拦截 | +| review_task | 1 — 1 | monitor_summary | 一个任务一条监控汇总 | +| review_task | 1 — 1 | review_report | 一个任务一份最终报告 | + +### 5.2 建表 DDL(`db/schema.sql`) + +```sql +CREATE TABLE IF NOT EXISTS review_task ( + id TEXT PRIMARY KEY, + created_at TEXT NOT NULL, + status TEXT NOT NULL, -- pending|running|done|failed + input_type TEXT NOT NULL, -- diff|repo|fixture + input_ref TEXT, + mode TEXT NOT NULL, -- dry-run|real + total_duration_ms INTEGER +); + +CREATE TABLE IF NOT EXISTS input_diff ( + id TEXT PRIMARY KEY, + task_id TEXT NOT NULL REFERENCES review_task(id), + file_path TEXT, + sha256 TEXT, + hunk_count INTEGER, + line_count INTEGER, + summary TEXT +); + +CREATE TABLE IF NOT EXISTS sandbox_run ( + id TEXT PRIMARY KEY, + task_id TEXT NOT NULL REFERENCES review_task(id), + runtime TEXT, -- local|container|cube + script TEXT, + status TEXT, -- ok|timeout|failed|truncated + duration_ms INTEGER, + exit_code INTEGER, + output_bytes INTEGER, + timed_out INTEGER, -- 0|1 + masked_count INTEGER +); + +CREATE TABLE IF NOT EXISTS finding ( + id TEXT PRIMARY KEY, + task_id TEXT NOT NULL REFERENCES review_task(id), + severity TEXT, -- critical|high|medium|low + category TEXT, + file TEXT, + line INTEGER, + title TEXT, + evidence TEXT, + recommendation TEXT, + confidence REAL, + source TEXT, -- rule|sandbox|llm + bucket TEXT -- findings|warnings|needs_human_review +); +CREATE INDEX IF NOT EXISTS idx_finding_dedup ON finding(task_id, file, line, category); +CREATE INDEX IF NOT EXISTS idx_finding_task ON finding(task_id); + +CREATE TABLE IF NOT EXISTS filter_block ( + id TEXT PRIMARY KEY, + task_id TEXT NOT NULL REFERENCES review_task(id), + reason TEXT, -- high-risk|forbidden-path|network|budget + target TEXT, + decision TEXT, -- deny|needs_human_review + detail TEXT +); +CREATE INDEX IF NOT EXISTS idx_filter_task ON filter_block(task_id); + +CREATE TABLE IF NOT EXISTS monitor_summary ( + id TEXT PRIMARY KEY, + task_id TEXT NOT NULL REFERENCES review_task(id), + total_duration_ms INTEGER, + sandbox_duration_ms INTEGER, + tool_calls INTEGER, + blocks INTEGER, + finding_count INTEGER, + sev_critical INTEGER, + sev_high INTEGER, + sev_medium INTEGER, + sev_low INTEGER, + exception_types TEXT -- JSON,如 {"timeout":2,"oom":1} +); + +CREATE TABLE IF NOT EXISTS review_report ( + id TEXT PRIMARY KEY, + task_id TEXT NOT NULL REFERENCES review_task(id), + report_json_path TEXT, + report_md_path TEXT, + summary TEXT, + created_at TEXT +); +CREATE INDEX IF NOT EXISTS idx_report_task ON review_report(task_id); +``` + +### 5.3 关键设计 + +- **复合索引 `idx_finding_dedup (task_id, file, line, category)`**——直接服务去重查询,O(1) 命中"同文件同行同类"。 +- **`finding.bucket` 三档**——从 schema 层强制低置信度不混入高置信结论。 +- **`sandbox_run.timed_out` / `output_bytes` / `masked_count`**——把安全边界执行证据持久化,可回放可审计。 +- **`monitor_summary` severity 平铺成列**——避免运行时反序列化 JSON,监控查询直接走索引。 +- **按 task_id 查询路径**:`SELECT * FROM review_task WHERE id=?` → 关联 `sandbox_run` / `finding` / `filter_block` / `monitor_summary` / `review_report`,一次 join 取回完整审查记录。 + +--- + +## 6. 安全治理:Filter + 沙箱 + +### 6.1 Filter 门禁(L3,沙箱前置) + +脚本清单在进入沙箱前必须经过四类检查,任一命中即按策略决策: + +| 检查类 | 命中条件 | 默认决策 | +|--------|----------|----------| +| 高风险脚本特征 | 脚本含 `rm -rf` / `sudo` / 网络下载执行 / `eval` 等危险模式 | `deny` | +| 禁止路径 | 访问 `/etc` / `~/.ssh` / 系统目录 / 工作区外路径 | `deny` | +| 非白名单网络 | 脚本发起非白名单域名/端口的网络请求 | `needs_human_review` | +| 超预算执行 | 预估耗时/内存超出预算阈值 | `needs_human_review` | + +- `allow` → 进入沙箱。 +- `deny` / `needs_human_review` → **不进沙箱**,拦截原因写入 `filter_block` 表 + 报告。 +- 拦截记录字段:`reason` / `target` / `decision` / `detail`,可按 `task_id` 查询回放。 + +### 6.2 沙箱 runtime(L4) + +| runtime | 用途 | 说明 | +|---------|------|------| +| Container | 生产默认 | Docker 隔离,资源限额 | +| Cube · E2B | 生产可选 | 远程沙箱,更强隔离 | +| 本地 fallback | 仅 dry-run / 开发 | **不能作为生产方案** | + +`runtime.py` 暴露统一 `SandboxRuntime.run(script, input, policy)` 接口,三种实现可切换。 + +### 6.3 沙箱安全边界(强制五项) + +`policy.py` 在每次沙箱执行时强制套用,不可绕过: + +| 约束 | 默认值 | 失败行为 | +|------|--------|----------| +| 超时控制 | 30s | 中断进程,`timed_out=1`,降级空诊断 | +| 输出大小限制 | 1MB | 截断,`status=truncated` | +| env 白名单 | `PATH`/`HOME`/`LANG` | 非白名单变量不透传 | +| 敏感信息脱敏 | 正则 + 熵值双检 | 输出落地前脱敏,`masked_count` 记录 | +| 失败记录 | — | 异常写 `sandbox_run`,任务不崩 | + +脱敏规则覆盖:明文 API Key、token、password、私钥、连接串。脱敏检出率目标 ≥ 95%,报告和数据库中不得出现明文。 + +--- + +## 7. CR Skill 设计 + +### 7.1 SKILL.md 契约 + +```yaml +--- +name: code-review +description: 自动代码评审 Skill,加载 6 类规则与脚本,在沙箱中执行检查并产出结构化 findings +version: 1.0.0 +entry: scripts/run_checks.py +rules: + - rules/security.md + - rules/async_errors.md + - rules/resource_leak.md + - rules/missing_tests.md + - rules/sensitive_info.md + - rules/db_lifecycle.md +sandbox: + default_runtime: container + fallback: local + timeout_s: 30 + max_output_bytes: 1048576 + env_whitelist: [PATH, HOME, LANG] +--- +``` + +`skill_load` 读取 frontmatter,产出 `RuleSet`(规则文档列表)+ 脚本清单,送入 Filter 门禁。 + +### 7.2 六类规则(覆盖要求的 4 类以上) + +| 规则文档 | 覆盖问题 | 检测方式 | 默认 severity | +|----------|----------|----------|---------------| +| `security.md` | SQL 注入、命令注入、硬编码密钥、不安全反序列化 | 静态模式 + 沙箱 semgrep | critical / high | +| `async_errors.md` | 未 await 的协程、未处理 rejection、async 资源泄漏 | AST 解析 | high / medium | +| `resource_leak.md` | 未关闭文件/连接、try 无 finally | AST + 控制流分析 | high / medium | +| `missing_tests.md` | 新增公开函数无对应测试 | diff 关联分析 | low | +| `sensitive_info.md` | 明文 API key / token / password | 正则 + 熵值 | critical | +| `db_lifecycle.md` | 连接未关闭、事务未提交/回滚、连接池泄漏 | AST | high / medium | + +### 7.3 脚本目录职责 + +| 脚本 | 输入 | 输出 | 运行位置 | +|------|------|------|----------| +| `parse_diff.py` | diff 文本 | `ChangeSet` | 本地(L1) | +| `run_checks.py` | `ChangeSet` + `RuleSet` | 原始诊断列表 | 沙箱(L4) | +| `dedupe.py` | 原始诊断列表 | 分流后 findings | 本地(L5) | +| `mask_secrets.py` | 任意文本 | 脱敏后文本 + count | 沙箱输出落地前 | + +--- + +## 8. 监控审计字段 + +每次 review 写入 `monitor_summary` 一条记录: + +| 字段 | 含义 | 来源层 | +|------|------|--------| +| `total_duration_ms` | 评审总耗时 | 全链路 | +| `sandbox_duration_ms` | 沙箱执行总耗时(所有 run 累加) | L4 | +| `tool_calls` | 工具/skill 调用次数 | L2 / L4 | +| `blocks` | Filter 拦截次数 | L3 | +| `finding_count` | finding 总数(三 bucket 合计) | L5 | +| `sev_critical` | critical 级 finding 数 | L5 | +| `sev_high` | high 级 finding 数 | L5 | +| `sev_medium` | medium 级 finding 数 | L5 | +| `sev_low` | low 级 finding 数 | L5 | +| `exception_types` | 异常类型分布(JSON) | 全链路 | + +--- + +## 9. 去重降噪策略 + +`dedupe.py` 处理流程: + +1. 收集所有原始诊断(来自规则匹配 + 沙箱结果)。 +2. 按 `(file, line, category)` 三元组分组。 +3. **组内取 `confidence` 最高的一条**,其余丢弃——消除"同文件同行同类重复报"。 +4. 按置信度分流 `bucket`: + - `confidence < 0.6` → `needs_human_review` + - `0.6 ≤ confidence < 0.8` → `warnings` + - `confidence ≥ 0.8` → `findings` +5. 同一行不同 `category` 的问题**保留**(不误合并不同类问题)。 +6. 低置信度问题不混入高置信 `findings`,保证误报率可控。 + +--- + +## 10. 方案设计说明 + +本方案把自动代码评审拆成六层流水线:输入解析、Skill 加载、Filter 治理、沙箱执行、去重结构化、存储输出,监控审计贯穿全链路。Skill 设计上,`code-review` Skill 以 `SKILL.md` frontmatter 声明入口、六类规则清单与沙箱策略,通过 `skill_load` 产出 `RuleSet` 与脚本清单,实现规则与执行解耦。沙箱默认 Container/Cube·E2B 隔离执行,本地仅作 dry-run 与开发 fallback,绝不作为生产方案;沙箱内强制套用超时(30s)、输出限制(1MB)、env 白名单、敏感信息脱敏、失败记录五项安全边界,异常降级为空诊断而不崩任务。Filter 治理在沙箱前置,对高风险脚本、禁止路径、非白名单网络、超预算执行做 `allow`/`deny`/`needs_human_review` 决策,后两者不进沙箱,拦截原因写入报告与数据库。数据库采用七张表围绕 `review_task` 聚合的最小 schema,`task_id` 为全局外键,复合索引 `(task_id,file,line,category)` 直接服务去重查询,支持按 task_id 一次 join 取回完整审查记录;`finding.bucket` 三档从 schema 层强制低置信度不混入高置信结论。去重按 `(file,line,category)` 取最高置信,置信度低于阈值进 warnings 或 needs_human_review,保证误报率 ≤ 15%。dry-run 模式用 fake runner 跑规则脚本,无 API Key 可验证解析→沙箱→落库全链路,耗时 ≤ 2 分钟。 + +--- + +## 11. 验收标准对照 + +| # | 验收标准 | 设计落点 | +|---|----------|----------| +| 1 | 8 条 diff 样本全部可运行并生成报告 | L1-L6 完整链路 + `tests/fixtures/` 8 个样本 | +| 2 | 高危检出率 ≥ 80%,误报率 ≤ 15% | 6 类规则覆盖 + 置信度分流(`confidence` 阈值) | +| 3 | DB 完整记录 task/run/finding/report,按 task_id 查询 | 七表 schema + `idx_*_task` 索引 + join 查询路径 | +| 4 | 沙箱超时+输出限制,失败不崩任务 | `policy.py` 五项约束 + 沙箱失败降级分支 | +| 5 | 脱敏检出率 ≥ 95%,无明文密钥 | `mask_secrets.py` 正则+熵值双检 + `masked_count` 审计 | +| 6 | dry-run ≤ 2 分钟 | fake runner 跳过模型 API,本地跑规则脚本 | +| 7 | 高风险先过 Filter,deny/review 不进沙箱 | `governance.py` 前置决策 + `filter_block` 落库 | +| 8 | 报告含 findings/统计/复核/拦截/监控/沙箱/修复 | `review_report` 八段式模板(见下) | + +### 11.1 报告八段式模板(`review_report.json` / `.md`) + +``` +1. findings 摘要 — 高置信 finding 列表(9 字段) +2. 严重级别统计 — critical/high/medium/low 计数 +3. 人工复核项 — needs_human_review 列表 +4. Filter 拦截摘要 — 拦截原因/目标/决策 +5. 监控指标 — 耗时/调用/拦截/finding 分布/异常分布 +6. 沙箱执行摘要 — runtime/状态/耗时/截断/脱敏数 +7. 可执行修复建议 — 每条 finding 附 recommendation +8. warnings — 低置信度问题(不混入 findings) +``` + +--- + +## 12. finding 字段契约 + +每条 finding 至少包含 9 字段: + +| 字段 | 类型 | 说明 | +|------|------|------| +| `severity` | string | critical / high / medium / low | +| `category` | string | security / async / resource / tests / sensitive / db | +| `file` | string | 文件路径 | +| `line` | int | 行号 | +| `title` | string | 问题标题 | +| `evidence` | text | 代码证据(已脱敏) | +| `recommendation` | text | 修复建议 | +| `confidence` | float | 0.0-1.0 | +| `source` | string | rule / sandbox / llm | + +--- diff --git a/examples/skills_code_review_agent/agent/__init__.py b/examples/skills_code_review_agent/agent/__init__.py new file mode 100644 index 00000000..0ea6341b --- /dev/null +++ b/examples/skills_code_review_agent/agent/__init__.py @@ -0,0 +1,27 @@ +"""Code Review Agent package. + +Thin re-export surface so callers can do:: + + import agent + await agent._async_main(ns) # or agent.main(...) + +without reaching into ``agent.agent`` directly. The heavy lifting lives in +``agent.agent`` (orchestration), with ``db/``, ``filters/``, ``llm/``, +``sandbox/`` and ``telemetry/`` as sub-packages. +""" + +from .agent import main +from .agent import _async_main +from .agent import skill_load +from .agent import load_diff +from .agent import persist_changeset +from .agent import FIXTURES + +__all__ = [ + "main", + "_async_main", + "skill_load", + "load_diff", + "persist_changeset", + "FIXTURES", +] diff --git a/examples/skills_code_review_agent/agent/agent.py b/examples/skills_code_review_agent/agent/agent.py new file mode 100644 index 00000000..26d0d5b2 --- /dev/null +++ b/examples/skills_code_review_agent/agent/agent.py @@ -0,0 +1,876 @@ +#!/usr/bin/env python3 +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Code Review Agent — CLI entry & full orchestration (Phase 5). + +Phase 5 scope: a single ``agent.py`` that orchestrates the whole pipeline — +parse a unified diff into a structured ``ChangeSet`` → load the +``code-review`` Skill (rules + scripts + sandbox config) → run the +``FilterGovernance`` checks (deny/review skip the sandbox) → execute the +checks (``FakeRunner`` in dry-run, an SDK-backed sandbox otherwise — +honoring the Skill's ``default_runtime``/``fallback`` contract, e.g. a +``container`` backend with transparent ``local`` fallback) → dedupe & +triage → persist to the Phase-0 +:class:`ReviewStore` → render the eight-section ``review_report.json`` / +``.md``. + +Inputs supported +---------------- +* ``--diff-file PATH`` parse a unified-diff file. +* ``--repo-path PATH`` run ``git diff HEAD`` inside the repo (staged + unstaged). +* ``--fixture NAME`` use a built-in sample diff (``clean`` / ``security``). + +The Skill is resolved relative to this file by default +(``skills/code-review``) and can be overridden with ``--skill-dir``. +""" + +from __future__ import annotations + +import argparse +import asyncio +import hashlib +import json +import os +import re +import subprocess +import sys +from pathlib import Path + +# --- import wiring --------------------------------------------------------- # +# This file lives at examples/skills_code_review_agent/agent/agent.py. +_HERE = Path(__file__).resolve().parent # .../agent +_EXAMPLE_ROOT = _HERE.parent # .../skills_code_review_agent +# Make the `agent` package importable when run as a standalone script +# (e.g. `python agent/agent.py`) — normally run_agent.py / tests already do this. +if str(_EXAMPLE_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLE_ROOT)) +from agent.db import SQLiteStore # noqa: E402 +from agent.db import ReviewStore # noqa: E402 + +# Make the skill scripts importable (parse_diff / run_checks / mask_secrets / dedupe). +_SKILL_DIR_DEFAULT = _EXAMPLE_ROOT / "skills" / "code-review" +_SCRIPTS_DIR = _SKILL_DIR_DEFAULT / "scripts" +if str(_SCRIPTS_DIR) not in sys.path: + sys.path.insert(0, str(_SCRIPTS_DIR)) +from parse_diff import ChangeSet # noqa: E402 +from parse_diff import ChangedFile # noqa: E402 +from parse_diff import parse_diff # noqa: E402 + + +# --------------------------------------------------------------------------- # +# Built-in fixture diffs (for demo / dry-run without a real repo) +# --------------------------------------------------------------------------- # +FIXTURES: dict[str, str] = { + "clean": """\ +diff --git a/clean.py b/clean.py +--- a/clean.py ++++ b/clean.py +@@ -1,3 +1,5 @@ + def add(a, b): +- return a + b ++ return a + b ++ ++ + def mul(a, b): +""", + "security": """\ +diff --git a/app/db.py b/app/db.py +--- a/app/db.py ++++ b/app/db.py +@@ -20,3 +20,7 @@ + def get_user(conn, uid): +- cur = conn.execute("SELECT * FROM users WHERE id=" + str(uid)) +- return cur.fetchone() ++ # parameterized query — safe ++ cur = conn.execute("SELECT * FROM users WHERE id=?", (uid,)) ++ return cur.fetchone() +diff --git a/app/secret.py b/app/secret.py +--- /dev/null ++++ b/app/secret.py +@@ -0,0 +1,2 @@ ++API_KEY = "sk-1234567890abcdef1234567890abcdef" ++TOKEN = "ghp_aBcDeFgHiJkLmNoPqRsTuVwXyZ1234567890" +""", + "llm_probe": """\ +diff --git a/svc/client.py b/svc/client.py +--- /dev/null ++++ b/svc/client.py +@@ -0,0 +1,7 @@ ++import requests ++ ++API_KEY = "sk-abcdef1234567890abcdef1234567890ab" ++ ++def call(endpoint, payload): ++ # TLS 校验关闭：有时内网有意为之，需人工确认 ++ return requests.post(endpoint, json=payload, verify=False) +""", +} + + +# --------------------------------------------------------------------------- # +# SKILL.md frontmatter parsing +# --------------------------------------------------------------------------- # +def _split_frontmatter(text: str) -> tuple[dict, str]: + """Return (frontmatter_dict, body) from a SKILL.md text. + + Frontmatter is the YAML between leading ``---`` fences. If no fences + are present, returns ``({}, text)``. + """ + if not text.startswith("---"): + return {}, text + parts = text.split("---", 2) + if len(parts) < 3: + return {}, text + fm_text = parts[1].strip("\n") + body = parts[2].lstrip("\n") + return _parse_yaml(fm_text), body + + +def _parse_yaml(text: str) -> dict: + """Minimal YAML parser for SKILL.md frontmatter. + + Prefers PyYAML when available; falls back to a small indentation-aware + parser that handles the subset SKILL.md uses: scalars, block & flow + lists, nested dicts, and ``>-`` / ``>`` / ``|`` block scalars. + """ + try: + import yaml # type: ignore + + data = yaml.safe_load(text) + return data if isinstance(data, dict) else {} + except ImportError: + return _parse_yaml_fallback(text) + + +def _parse_yaml_fallback(text: str) -> dict: + """Indentation-aware YAML subset parser (no external deps).""" + lines = text.splitlines() + root: dict = {} + # Stack of (indent, container) where container is dict or list. + stack: list[tuple[int, dict | list]] = [(0, root)] + i = 0 + while i < len(lines): + raw = lines[i] + stripped = raw.strip() + if not stripped or stripped.startswith("#"): + i += 1 + continue + indent = len(raw) - len(raw.lstrip(" ")) + # Pop stack to the current indent level. + while stack and stack[-1][0] > indent: + stack.pop() + if stack and stack[-1][0] < indent: + # The previous line opened a block; its container is already on + # top via the list/dict push below. + pass + if stripped.startswith("- "): + # list item + item_val = _coerce(stripped[2:].strip()) + container = stack[-1][1] if stack else root + if isinstance(container, list): + container.append(item_val) + i += 1 + continue + # key: value + if ":" in stripped: + key, _, val = stripped.partition(":") + key = key.strip() + val = val.strip() + container = stack[-1][1] if stack else root + if not isinstance(container, dict): + i += 1 + continue + if val == "": + # Block follows: peek next non-empty line indent. + if i + 1 < len(lines): + nxt = lines[i + 1] + nxt_indent = len(nxt) - len(nxt.lstrip(" ")) + if nxt.strip().startswith("- "): + new_list: list = [] + container[key] = new_list + stack.append((nxt_indent, new_list)) + elif nxt_indent > indent: + new_dict: dict = {} + container[key] = new_dict + stack.append((nxt_indent, new_dict)) + else: + container[key] = None + elif val in (">-", ">", "|"): + # Block scalar — collect deeper-indented lines. + collected: list[str] = [] + j = i + 1 + while j < len(lines): + nxt = lines[j] + if nxt.strip() == "": + collected.append("") + j += 1 + continue + nxt_indent = len(nxt) - len(nxt.lstrip(" ")) + if nxt_indent <= indent: + break + collected.append(nxt.strip()) + j += 1 + container[key] = " ".join(x for x in collected if x) + i = j + continue + else: + container[key] = _coerce(val) + i += 1 + return root + + +def _coerce(val: str): + """Coerce a YAML scalar string to int/float/bool/None/list/str.""" + if val.startswith("[") and val.endswith("]"): + inner = val[1:-1].strip() + if not inner: + return [] + return [_coerce(x.strip()) for x in inner.split(",")] + if val.startswith('"') and val.endswith('"'): + return val[1:-1] + if val.startswith("'") and val.endswith("'"): + return val[1:-1] + if val.lower() in ("true", "false"): + return val.lower() == "true" + if val.lower() in ("null", "~", ""): + return None + try: + return int(val) + except ValueError: + pass + try: + return float(val) + except ValueError: + pass + return val + + +# --------------------------------------------------------------------------- # +# skill_load +# --------------------------------------------------------------------------- # +def skill_load(skill_dir: str | Path) -> dict: + """Load a code-review Skill via the SDK ``FsSkillRepository``. + + The SDK repository parses the SKILL.md frontmatter (name/description) and + loads the skill body + resources (rules/*.md). Two pieces are NOT exposed + by the SDK ``Skill`` object and are filled here: + * ``sandbox_config`` — parsed from the SKILL.md frontmatter (sandbox: block) + * ``scripts`` — scanned from ``scripts/*.py`` on disk + + Returns:: + + { + "name": "code-review", + "skill_dir": "/abs/path/to/skills/code-review", + "rules": ["rules/security.md", ...], + "scripts": ["scripts/parse_diff.py", ...], + "sandbox_config": {default_runtime, fallback, timeout_s, ...}, + } + """ + # Lazy import: callers that don't invoke skill_load (e.g. P2 rule-engine + # tests importing agent for FIXTURES) aren't forced to load the SDK. + from trpc_agent_sdk.skills import FsSkillRepository + + skill_dir = Path(skill_dir) + skill_md = skill_dir / "SKILL.md" + if not skill_md.exists(): + raise FileNotFoundError(f"SKILL.md not found: {skill_md}") + + # SDK: discover + load the skill (frontmatter name/description + body + resources). + repo = FsSkillRepository(str(skill_dir.parent)) + frontmatter, _ = _split_frontmatter(skill_md.read_text(encoding="utf-8")) + name = frontmatter.get("name", skill_dir.name) + skill = repo.get(name) + base = Path(skill.base_dir) + + # rules: prefer SDK resources (rules/*.md with content), fall back to disk scan. + def _res_path(r): + return getattr(r, "path", None) or (r.get("path") if isinstance(r, dict) else None) + + rules = sorted({ + p for r in skill.resources + if (p := _res_path(r)) and p.startswith("rules/") + }) or _rel_list(base / "rules", "*.md", base) + + # scripts: scan disk (SDK resources don't include .py scripts). + scripts = _rel_list(base / "scripts", "*.py", base) + + return { + "name": name, + "skill_dir": str(base), + "rules": rules, + "scripts": scripts, + "sandbox_config": frontmatter.get("sandbox", {}), + } + + +def _rel_list(directory: Path, pattern: str, base: Path) -> list[str]: + """Sorted list of paths under ``directory`` matching ``pattern``, relative to ``base``.""" + if not directory.exists(): + return [] + out = [] + for p in sorted(directory.glob(pattern)): + rel = p.relative_to(base).as_posix() + out.append(rel) + return out + + +# --------------------------------------------------------------------------- # +# Input collection +# --------------------------------------------------------------------------- # +def load_diff( + diff_file: str | None = None, + repo_path: str | None = None, + fixture: str | None = None, +) -> tuple[str, str, str]: + """Collect a diff text from one of three sources. + + Returns ``(diff_text, input_type, input_ref)`` where ``input_type`` is + ``diff`` / ``repo`` / ``fixture`` and ``input_ref`` is a human-readable + reference (file path, repo path, or fixture name). + """ + if diff_file: + path = Path(diff_file) + if not path.exists(): + raise FileNotFoundError(f"diff file not found: {path}") + return (path.read_text(encoding="utf-8"), "diff", str(path)) + if repo_path: + return (_git_diff_head(repo_path), "repo", str(repo_path)) + if fixture: + if fixture not in FIXTURES: + raise KeyError( + f"unknown fixture '{fixture}', available: {list(FIXTURES)}" + ) + return (FIXTURES[fixture], "fixture", fixture) + raise ValueError("no input source provided (use --diff-file/--repo-path/--fixture)") + + +def _git_diff_head(repo_path: str) -> str: + """Run ``git diff HEAD`` in ``repo_path`` (staged + unstaged changes).""" + try: + proc = subprocess.run( + ["git", "diff", "HEAD"], + cwd=repo_path, + capture_output=True, + text=True, + check=False, + ) + except FileNotFoundError as exc: + raise RuntimeError("git executable not found on PATH") from exc + if proc.returncode != 0 and proc.stderr: + # Non-fatal: empty diff on a fresh repo is fine; surface real errors. + if "not a git repository" in proc.stderr.lower(): + raise RuntimeError(f"not a git repository: {repo_path}") + return proc.stdout + + +# --------------------------------------------------------------------------- # +# Persistence helpers +# --------------------------------------------------------------------------- # +def _file_sha256(file: ChangedFile, path: str) -> str: + """Stable SHA-256 of a changed file's diff content for dedup/identity.""" + parts = [path] + for hunk in file.hunks: # type: ignore[attr-defined] + for ln in hunk.lines: # type: ignore[attr-defined] + parts.append(f"{ln.type}|{ln.content}") # type: ignore[attr-defined] + digest = hashlib.sha256("\n".join(parts).encode("utf-8")).hexdigest() + return digest + + +async def persist_changeset( + store: ReviewStore, + task_id: str, + cs: ChangeSet, +) -> int: + """Write each ``ChangedFile`` as an ``input_diff`` row. Returns row count.""" + count = 0 + for f in cs.files: + summary = f"{f.hunk_count} hunks, {f.added_lines}+, {f.deleted_lines}-" + await store.add_input_diff( + task_id=task_id, + file_path=f.path, + sha256=_file_sha256(f, f.path), + hunk_count=f.hunk_count, + line_count=f.line_count, + summary=summary, + ) + count += 1 + return count + + +# --------------------------------------------------------------------------- # +# CLI +# --------------------------------------------------------------------------- # +def _build_arg_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + description="Code Review Agent — parse diff, load skill, persist inputs.", + ) + src = p.add_mutually_exclusive_group(required=True) + src.add_argument("--diff-file", help="path to a unified diff file") + src.add_argument("--repo-path", help="repo root to run `git diff HEAD` in") + src.add_argument( + "--fixture", + choices=sorted(FIXTURES.keys()), + help="use a built-in sample diff", + ) + p.add_argument( + "--skill-dir", + default=str(_SKILL_DIR_DEFAULT), + help="code-review skill directory (default: skills/code-review)", + ) + p.add_argument( + "--db-path", + default="cr_agent.db", + help="SQLite database path (default: cr_agent.db)", + ) + p.add_argument( + "--mode", + choices=["dry-run", "real"], + default="dry-run", + help="review mode recorded on the task (default: dry-run)", + ) + p.add_argument( + "--print-changeset", + action="store_true", + help="also print the parsed ChangeSet as JSON", + ) + p.add_argument( + "--telemetry", + action="store_true", + help="enable OTel tracing (spans printed to stdout via ConsoleSpanExporter)", + ) + p.add_argument( + "--output-dir", + default=".", + help="directory for review_report.json + review_report.md (default: .)", + ) + p.add_argument( + "--dry-run", + action="store_true", + help="shortcut for --mode dry-run (no model API, full pipeline)", + ) + p.add_argument( + "--enable-llm", + action="store_true", + help="enable real-LLM second-opinion triage (also via LLM_ENABLED in .env)", + ) + p.add_argument( + "--require-sandbox", + action="store_true", + help="production: demand the Skill's default_runtime (e.g. container) " + "and refuse to fall back to local — enforce an isolated sandbox", + ) + p.add_argument( + "--llm-env", + default=None, + help="path to an explicit .env file for LLM config (default: project-root/.env)", + ) + return p + + +class FakeRunner: + """dry-run sandbox substitute: imports run_checks directly, no model API. + + Used when ``mode == "dry-run"`` (or as a degradation fallback when the real + sandbox fails). Produces the same ``list[RawFinding]`` as the sandbox path. + """ + + async def run(self, changeset, ruleset): + from run_checks import run_checks as _run + + return _run(changeset, ruleset) + + +def _run_checks_rel(skill: dict) -> str | None: + """Locate the run_checks checker among the skill's declared scripts. + + Returns the skill-relative path (e.g. ``scripts/run_checks.py``) or + ``None`` when the skill ships no checker. The Filter ``decisions`` dict is + keyed by these relative paths (see step 2 below), so we match on the + ``run_checks.py`` tail to find the decision that gates execution. + """ + for s in skill.get("scripts", []): + if s.endswith("run_checks.py"): + return s + return None + + +def _script_rel(skill: dict, name_tail: str) -> str | None: + """Return the skill-relative path of a declared script ending in ``name_tail``. + + Used to look up a script's Filter decision by its declared relative path + (e.g. ``scripts/parse_diff.py``), independent of where it sits on disk. + """ + for s in skill.get("scripts", []): + if s.endswith(name_tail): + return s + return None + + +def _gate(decisions: dict, rel_path: str | None): + """Return the FilterDecision if it is ``deny``/``needs_human_review``, else None. + + A non-allow verdict on a Skill script means that phase MUST NOT execute the + script. Callers should stop the phase early (emit an empty/blocked result) + so the denied script never enters the execution chain — this is the unified + gate that protects *every* declared script (parse_diff / run_checks / + dedupe / mask_secrets), not just run_checks.py. + """ + if not rel_path: + return None + d = decisions.get(rel_path) + if d is not None and getattr(d, "verdict", "allow") != "allow": + return d + return None + + +def build_report( + task_id: str, + dedupe_result, + sandbox_runs: list[dict], + filter_blocks: list[dict], + monitor: dict, + output_dir: str, +) -> tuple[str, str]: + """Render the eight-section ``review_report.json`` + ``review_report.md``. + + Returns ``(json_path, md_path)``. Both files share the same data source. + """ + out = Path(output_dir) + out.mkdir(parents=True, exist_ok=True) + report = { + "task_id": task_id, + "1_findings": [f.__dict__ for f in dedupe_result.findings], + "2_severity_stats": dedupe_result.severity_counts(), + "3_needs_human_review": [f.__dict__ for f in dedupe_result.needs_human_review], + "4_filter_blocks": filter_blocks, + "5_monitor": monitor, + "6_sandbox_runs": sandbox_runs, + "7_recommendations": [ + {"file": f.file, "line": f.line, "recommendation": f.recommendation} + for f in dedupe_result.findings + ], + "8_warnings": [f.__dict__ for f in dedupe_result.warnings], + } + json_path = out / "review_report.json" + json_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") + md_path = out / "review_report.md" + md_path.write_text(_render_report_md(report), encoding="utf-8") + return str(json_path), str(md_path) + + +def _render_report_md(report: dict) -> str: + """Eight-section Markdown — human-readable, same data as the JSON.""" + L: list[str] = [f"# Code Review Report — {report['task_id']}", ""] + L.append(f"## 1. Findings 摘要（{len(report['1_findings'])} 条）") + for f in report["1_findings"]: + L.append(f"- [{f['severity']}] {f['file']}:{f['line']} {f['title']} (conf={f['confidence']})") + L += ["", "## 2. 严重级别统计"] + ss = report["2_severity_stats"] + for sev in ("critical", "high", "medium", "low"): + L.append(f"- {sev}: {ss.get(sev, 0)}") + L.append(f"\n## 3. 人工复核项（{len(report['3_needs_human_review'])} 条）") + for f in report["3_needs_human_review"]: + L.append(f"- {f['file']}:{f['line']} {f['title']} (conf={f['confidence']})") + L.append(f"\n## 4. Filter 拦截摘要（{len(report['4_filter_blocks'])} 条）") + for b in report["4_filter_blocks"]: + L.append(f"- [{b['verdict']}] {b['reason']}: {b['target']} — {b['detail']}") + m = report["5_monitor"] + L += ["", "## 5. 监控指标"] + for k in ("total_duration_ms", "sandbox_duration_ms", "tool_calls", "blocks", + "finding_count", "exception_types"): + L.append(f"- {k}: {m.get(k)}") + L.append(f"\n## 6. 沙箱执行摘要（{len(report['6_sandbox_runs'])} 次）") + for s in report["6_sandbox_runs"]: + L.append(f"- {s.get('runtime','?')}: status={s.get('status','?')} " + f"dur={s.get('duration_ms',0)}ms timed_out={s.get('timed_out',0)} " + f"masked={s.get('masked_count',0)}") + L += ["", "## 7. 可执行修复建议"] + for r in report["7_recommendations"]: + L.append(f"- {r['file']}:{r['line']}: {r['recommendation']}") + L.append(f"\n## 8. Warnings（{len(report['8_warnings'])} 条，低置信度，不混入 findings）") + for f in report["8_warnings"]: + L.append(f"- [{f['severity']}] {f['file']}:{f['line']} {f['title']} (conf={f['confidence']})") + return "\n".join(L) + "\n" + + +async def _async_main(args) -> int: + """Full pipeline: parse → skill → filter → sandbox → dedupe → persist → report. + + Returns the process exit code: ``0`` on success, ``1`` when the pipeline + fails (so CI / acceptance harnesses can detect failure instead of a silent + success). Any failure is logged with ``[agent] ERROR`` and the partial + task is marked ``failed`` when possible. + """ + from agent.telemetry import TelemetryRecorder, init_telemetry, trace_stage + init_telemetry(enabled=args.telemetry) + recorder = TelemetryRecorder() + task_id: str | None = None + exit_code = 0 + if getattr(args, "dry_run", False): + args.mode = "dry-run" # --dry-run shortcut → --mode dry-run + + # 1. Load the skill (rules + scripts + sandbox config) via SDK FsSkillRepository. + async with trace_stage("l2_skill_load", recorder): + skill = skill_load(args.skill_dir) + print(f"[agent] skill : {skill['name']} ({len(skill['rules'])} rules, {len(skill['scripts'])} scripts)") + + # 2. Compute Filter decisions for EVERY declared Skill script (L3) BEFORE + # any script executes. The gate is authoritative: a non-allow verdict on + # ANY skill script must stop its phase *before* that script runs. This + # now covers parse_diff (L1), run_checks (L4), dedupe (L5) and the + # mask_secrets helper that dedupe / LLM triage invoke internally. + # (Previously only run_checks.py was gated, so parse_diff / dedupe / + # mask_secrets could still execute after a deny / needs_human_review.) + from agent.filters import FilterGovernance + + gov = FilterGovernance() + decisions: dict[str, object] = {} + filter_blocks_meta: list[dict] = [] + for script in skill["scripts"]: + sp = Path(skill["skill_dir"]) / script + content = sp.read_text(encoding="utf-8") if sp.exists() else "" + decision = gov.decide(str(sp), content, {}) + decisions[script] = decision + if decision.verdict != "allow": + filter_blocks_meta.append( + {"verdict": decision.verdict, "reason": decision.reason, + "target": script, "detail": decision.detail}) + + # 3. Collect + parse the diff (L1) — gated on parse_diff's verdict. + async with trace_stage("l1_parse", recorder): + diff_text, input_type, input_ref = load_diff( + diff_file=args.diff_file, repo_path=args.repo_path, fixture=args.fixture) + parse_rel = _script_rel(skill, "parse_diff.py") + parse_block = _gate(decisions, parse_rel) + if parse_block is not None: + # parse_diff is denied/reviewed → we cannot build a ChangeSet, so + # the entire review chain stops here. The filter_block above is the + # signal; nothing downstream (sandbox/dedupe) may run. + print(f"[agent] filter BLOCKED parse stage: {parse_block.verdict} ({parse_rel})") + cs = None + else: + cs = parse_diff(diff_text) + print(f"[agent] input : {input_type} = {input_ref}" + + (f" | {cs.file_count} file(s)" if cs is not None else " | (parse blocked)")) + if args.print_changeset and cs is not None: + print(cs.to_json()) + + store = None + sandbox_runs_meta: list[dict] = [] + raw_findings = [] + try: + # 4. create_task + persist input_diff + flush recorded filter blocks. + store = SQLiteStore(args.db_path) + async with trace_stage("l6_persist", recorder): + task_id = await store.create_task(input_type, input_ref, args.mode) + await store.update_task_status(task_id, "running") + if cs is not None: + await persist_changeset(store, task_id, cs) + for b in filter_blocks_meta: + await store.add_filter_block( + task_id, b["reason"], b["target"], b["verdict"], b["detail"]) + print(f"[agent] filter : {len(filter_blocks_meta)} block(s)") + + # 5–6.5: only when we actually have a ChangeSet (parse_diff was allowed). + # If parse_diff was filtered, the whole sandbox/dedupe/triage chain is + # skipped and an empty review is emitted — no Skill script executes. + if cs is None: + from cr_models import DedupeResult + raw_findings = [] + dedupe_result = DedupeResult() + else: + from agent.sandbox import ( + SandboxPolicy, + build_runtime_with_fallback, + select_runtime, + ) + policy = SandboxPolicy.from_config(skill["sandbox_config"]) + + # 5. Run checks (L4 sandbox). + # SECURITY GATE — the Filter decision on the checker is authoritative: + # if it is ``deny``/``needs_human_review``, the script MUST NOT be + # executed at all. Not in the sandbox, and NOT via the in-process + # FakeRunner fallback either — running a denied script "locally" + # outside the sandbox would silently defeat the gate. We emit no raw + # findings, keep the recorded filter_block as the signal, and mark + # the (non-) execution as ``blocked`` in the report/DB. + rc_rel = _run_checks_rel(skill) + rc_path = str(Path(skill["skill_dir"]) / rc_rel) if rc_rel else "" + rc_decision = decisions.get(rc_rel) if rc_rel else None + async with trace_stage("l4_sandbox", recorder): + try: + if rc_decision is not None and rc_decision.verdict != "allow": + # Filter blocked the checker → skip execution entirely. + # CRITICAL: we do NOT import run_checks here at all, so + # neither its module top-level nor load_rules() ever run + # for a denied/needs_human_review script. + raw_findings = [] + verdict = rc_decision.verdict # "deny" | "needs_human_review" + sandbox_runs_meta.append( + {"runtime": "blocked", "status": verdict, "duration_ms": 0, + "exit_code": 0, "output_bytes": 0, "timed_out": 0, "masked_count": 0}) + # P1-1: the DB must fully record every sandbox-execution + # decision, including a blocked one — otherwise the chain + # is not testable. + await store.add_sandbox_run( + task_id, "blocked", rc_rel or "run_checks.py", verdict, + 0, 0, 0, 0, 0) + else: + # Allowed → only NOW import the checker module (its + # top-level + load_rules run exclusively on the allow + # path; a denied checker never reaches this branch). + from run_checks import load_rules + rules_by_cat = load_rules(skill["skill_dir"], skill["rules"]) + if args.mode == "dry-run": + raw_findings = await FakeRunner().run(cs, {"_rules": rules_by_cat}) + sandbox_runs_meta.append( + {"runtime": "fake", "status": "ok", "duration_ms": 0, + "exit_code": 0, "output_bytes": 0, "timed_out": 0, "masked_count": 0}) + # P1-1: the DB must fully record every sandbox execution, + # including dry-run — otherwise the chain is not testable. + await store.add_sandbox_run( + task_id, "fake", rc_rel or "run_checks.py", "ok", 0, 0, 0, 0, 0) + else: + # Real mode: honor the skill's declared sandbox contract. + # G1 fix — previously hardcoded to LocalRuntime, ignoring + # `default_runtime`/`fallback` from sandbox_config. Now we + # try default_runtime (container) and transparently fall + # back to `fallback` (local) when it can't be provisioned. + sb_cfg = skill["sandbox_config"] + default_kind = sb_cfg.get("default_runtime", "local") + fallback_kind = sb_cfg.get("fallback", "local") + if getattr(args, "require_sandbox", False): + # Production: the Skill's declared default_runtime is + # the isolated sandbox contract (e.g. container). Do + # NOT silently fall back to local — if it can't be + # provisioned, fail loudly so the violation is visible. + runtime = select_runtime(default_kind, policy) + runtime.ensure_available() + actual_kind = default_kind + else: + runtime, actual_kind = build_runtime_with_fallback( + default_kind, fallback_kind, policy) + rr = await runtime.run( + rc_path, {"changeset": cs.to_dict(), "rules": rules_by_cat}, policy) + await store.add_sandbox_run( + task_id, actual_kind, rc_rel or "run_checks.py", rr.status, rr.duration_ms, + rr.exit_code, rr.output_bytes, int(rr.timed_out), rr.masked_count) + sandbox_runs_meta.append( + {"runtime": actual_kind, "status": rr.status, "duration_ms": rr.duration_ms, + "exit_code": rr.exit_code, "output_bytes": rr.output_bytes, + "timed_out": int(rr.timed_out), "masked_count": rr.masked_count}) + if rr.status == "ok" and rr.stdout: + from run_checks import RawFinding + raw_findings = [RawFinding(**d) for d in json.loads(rr.stdout)] + else: + raw_findings = await FakeRunner().run(cs, {"_rules": rules_by_cat}) + except Exception as exc: + # Sandbox failure degrades to FakeRunner — pipeline never + # crashes. (Only reachable when the checker was *allowed*; a + # denied checker never reaches this point, so it stays + # unexecuted.) + recorder.record_exception(type(exc).__name__) + raw_findings = await FakeRunner().run(cs, {"_rules": rules_by_cat}) + sandbox_runs_meta.append( + {"runtime": "degraded", "status": "failed", "duration_ms": 0, + "exit_code": -1, "output_bytes": 0, "timed_out": 0, "masked_count": 0}) + await store.add_sandbox_run( + task_id, "degraded", rc_rel or "run_checks.py", "failed", 0, -1, 0, 0, 0) + print(f"[agent] sandbox : {len(raw_findings)} raw findings") + + # 6. Dedupe (L5) — gated on dedupe AND mask_secrets verdicts. + # Running dedupe executes mask_secrets on every finding's evidence; + # if either script is deny/needs_human_review we must NOT run it, so + # we emit an empty DedupeResult instead (no findings leaked through). + # Compute the gate FIRST, then import the script module only on allow + # — otherwise importing ``dedupe`` (a filtered script) would execute + # its top-level even when the gate is denying it. + dedupe_block = _gate(decisions, _script_rel(skill, "dedupe.py")) \ + or _gate(decisions, _script_rel(skill, "mask_secrets.py")) + if dedupe_block is not None: + from cr_models import DedupeResult + print(f"[agent] filter BLOCKED dedupe stage: {dedupe_block.verdict} " + f"({dedupe_block.target})") + dedupe_result = DedupeResult() + else: + from cr_models import DedupeResult + from dedupe import dedupe + dedupe_result = dedupe(raw_findings) + print(f"[agent] dedupe : {len(dedupe_result.findings)} findings | " + f"{len(dedupe_result.warnings)} warnings | " + f"{len(dedupe_result.needs_human_review)} review") + + # 6.5 LLM second-opinion triage (optional, Phase 7). + # Gated on mask_secrets: the diff is masked before leaving the + # process, so if mask_secrets is deny/needs_human_review we MUST NOT + # send the unmasked diff to the model — skip triage entirely. + async with trace_stage("l5b_llm_triage", recorder): + from agent.llm import LlmTriage, get_llm_client + llm_client = get_llm_client( + enable=getattr(args, "enable_llm", False), + env_path=getattr(args, "llm_env", None)) + ms_block = _gate(decisions, _script_rel(skill, "mask_secrets.py")) + if llm_client.is_enabled and ms_block is None: + dedupe_result = await LlmTriage(llm_client).run( + dedupe_result, diff_text) + print(f"[agent] llm triage: {llm_client.kind} enabled | " + f"{len(dedupe_result.needs_human_review)} still need review") + elif ms_block is not None: + print(f"[agent] llm triage: SKIPPED — mask_secrets blocked " + f"({ms_block.verdict})") + else: + print("[agent] llm triage: disabled (LLM_ENABLED=false or no API key)") + + # 7. Persist findings + monitor + report (L6). + async with trace_stage("l6_persist2", recorder): + for finding, bucket in dedupe_result.all_with_bucket(): + await store.add_finding( + task_id, finding.severity, finding.category, finding.file, + finding.line, finding.title, finding.evidence, + finding.recommendation, finding.confidence, finding.source, bucket) + monitor = recorder.to_monitor_summary( + finding_count=dedupe_result.total, + sev_counts=dedupe_result.severity_counts(), + blocks=len(filter_blocks_meta)) + await store.set_monitor_summary(task_id, monitor) + json_path, md_path = build_report( + task_id, dedupe_result, sandbox_runs_meta, filter_blocks_meta, + monitor, args.output_dir) + summary = (f"{len(dedupe_result.findings)} findings, " + f"{len(dedupe_result.warnings)} warnings, " + f"{len(dedupe_result.needs_human_review)} review") + await store.set_report(task_id, json_path, md_path, summary) + await store.update_task_status(task_id, "done", total_duration_ms=monitor["total_duration_ms"]) + + print(f"[agent] task : {task_id} (done)") + print(f"[agent] report : {json_path}") + print(f"[agent] telemetry : total={monitor['total_duration_ms']}ms exceptions={monitor['exception_types']}") + except Exception as exc: + # Top-level fail-safe: record exception, mark task failed, still emit a report. + # P2-1: signal failure to the caller via a non-zero exit code. + exit_code = 1 + recorder.record_exception(type(exc).__name__) + if task_id: + try: + await store.update_task_status(task_id, "failed") + await store.set_monitor_summary(task_id, recorder.to_monitor_summary()) + except Exception: + pass + print(f"[agent] ERROR : {type(exc).__name__}: {exc}") + finally: + if store is not None: + await store.close() + return exit_code + + +def main(argv: list[str] | None = None) -> int: + args = _build_arg_parser().parse_args(argv) + return asyncio.run(_async_main(args)) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/examples/skills_code_review_agent/agent/db/__init__.py b/examples/skills_code_review_agent/agent/db/__init__.py new file mode 100644 index 00000000..b2bf1624 --- /dev/null +++ b/examples/skills_code_review_agent/agent/db/__init__.py @@ -0,0 +1,17 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Code Review Agent — persistence foundation (Phase 0). + +Exposes the :class:`ReviewStore` protocol and the default +:class:`SQLiteStore` implementation. Downstream phases (P1–P6) interact +with persisted review data exclusively through this package so that the +storage backend can be swapped without touching the orchestration layer. +""" + +from .storage import ReviewStore +from .storage import SQLiteStore + +__all__ = ["ReviewStore", "SQLiteStore"] diff --git a/examples/skills_code_review_agent/agent/db/init_db.py b/examples/skills_code_review_agent/agent/db/init_db.py new file mode 100644 index 00000000..61ed1e28 --- /dev/null +++ b/examples/skills_code_review_agent/agent/db/init_db.py @@ -0,0 +1,108 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Initialize the Code Review Agent database (SDK-backed, async). + +Table creation is driven by the ORM metadata in :mod:`db.models` via the +SDK's :class:`SqlStorage` (``metadata.create_all``). Idempotent — safe to +run repeatedly. SQLite ``PRAGMA foreign_keys=ON`` is applied automatically +by the SDK storage layer. + +Usage +----- + python agent/db/init_db.py # default ./cr_agent.db + python agent/db/init_db.py --db-path /tmp/cr.db + python -m examples.skills_code_review_agent.agent.db.init_db --db-path cr.db +""" + +from __future__ import annotations + +import argparse +import asyncio +import sys +from pathlib import Path + +# Allow running as a script (no package context). +_HERE = Path(__file__).resolve().parent # .../agent/db +if str(_HERE.parent.parent) not in sys.path: + sys.path.insert(0, str(_HERE.parent.parent)) # project root + +from agent.db import SQLiteStore # noqa: E402 +from agent.db.models import CRBase # noqa: E402 + +_EXPECTED_TABLES = ( + "review_task", + "input_diff", + "sandbox_run", + "finding", + "filter_block", + "monitor_summary", + "review_report", +) + + +async def init_db(db_path: str | Path = "cr_agent.db", *, verbose: bool = True) -> None: + """Create/upgrade the CR Agent database at ``db_path``. + + Safe to call repeatedly — ``metadata.create_all`` only adds missing + tables, and the SDK applies forward-only column migrations. + """ + store = SQLiteStore(db_path) + try: + # create_sql_engine() runs metadata.create_all + sqlite pragmas. + await store._ensure_engine() + + if verbose: + await _report(store, db_path) + finally: + await store.close() + + +async def _report(store: "SQLiteStore", db_path: str | Path) -> None: + """Print a concise summary of tables now present.""" + from sqlalchemy import inspect as sa_inspect + + await store._ensure_engine() + engine = store.storage._db_engine # underlying async engine + # Run sync inspect via the async engine. + def _sync_inspect(conn): + return sa_inspect(conn).get_table_names() + + async with engine.connect() as conn: + tables = await conn.run_sync(_sync_inspect) + table_set = {t for t in tables if not t.startswith("sqlite_")} + missing = [t for t in _EXPECTED_TABLES if t not in table_set] + + print(f"[init_db] database : {db_path}") + print(f"[init_db] tables : {len(table_set)} ({', '.join(sorted(table_set))})") + if missing: + print(f"[init_db] WARNING missing tables: {missing}", file=sys.stderr) + else: + print(f"[init_db] all {len(_EXPECTED_TABLES)} expected tables present") + print("[init_db] foreign_keys = ON (set by SDK SqlStorage)") + + +def _build_arg_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Initialize the Code Review Agent SQLite database (SDK-backed).", + ) + parser.add_argument( + "--db-path", default="cr_agent.db", + help="Path to the SQLite database file (default: cr_agent.db).", + ) + parser.add_argument( + "--quiet", action="store_true", help="Suppress the summary report.", + ) + return parser + + +def main(argv: list[str] | None = None) -> int: + args = _build_arg_parser().parse_args(argv) + asyncio.run(init_db(args.db_path, verbose=not args.quiet)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/examples/skills_code_review_agent/agent/db/models.py b/examples/skills_code_review_agent/agent/db/models.py new file mode 100644 index 00000000..c79e4278 --- /dev/null +++ b/examples/skills_code_review_agent/agent/db/models.py @@ -0,0 +1,170 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""SQLAlchemy ORM models for the Code Review Agent (Phase 0, SDK-backed). + +The seven business tables are declared as a custom ``CRBase(DeclarativeBase)`` +metadata and persisted through the SDK's generic :class:`SqlStorage` layer +(``SqlStorage(metadata=CRBase.metadata)``). This replaces the original +hand-rolled ``sqlite3`` implementation while keeping the same schema and +the same :class:`ReviewStore` protocol surface. + +Table layout mirrors ``db/schema.sql`` (ARCHITECTURE.md §5.2): seven tables +aggregated around ``review_task`` with ``task_id`` as the global FK. The +composite index ``idx_finding_dedup (task_id, file, line, category)`` is +declared on :class:`Finding` to serve the Phase-4 dedupe lookup. +""" + +from __future__ import annotations + +from sqlalchemy import ForeignKey +from sqlalchemy import Float +from sqlalchemy import Index +from sqlalchemy import Integer +from sqlalchemy import String +from sqlalchemy import Text +from sqlalchemy.orm import DeclarativeBase +from sqlalchemy.orm import Mapped +from sqlalchemy.orm import mapped_column + + +class CRBase(DeclarativeBase): + """Declarative base for all CR Agent business tables. + + Passed to ``SqlStorage(metadata=CRBase.metadata)`` so the SDK storage + layer creates/manages exactly these tables (independent of the SDK's own + session/memory tables). + """ + + +class ReviewTask(CRBase): + """L6 master table — one row per review run.""" + + __tablename__ = "review_task" + + id: Mapped[str] = mapped_column(String, primary_key=True) + created_at: Mapped[str] = mapped_column(String, nullable=False) + status: Mapped[str] = mapped_column(String, nullable=False) # pending|running|done|failed + input_type: Mapped[str] = mapped_column(String, nullable=False) # diff|repo|fixture + input_ref: Mapped[str | None] = mapped_column(String, nullable=True) + mode: Mapped[str] = mapped_column(String, nullable=False) # dry-run|real + total_duration_ms: Mapped[int | None] = mapped_column(Integer, nullable=True) + + +class InputDiff(CRBase): + """L1 parsed diff per changed file (summary only, never the full blob).""" + + __tablename__ = "input_diff" + __table_args__ = (Index("idx_input_diff_task", "task_id"),) + + id: Mapped[str] = mapped_column(String, primary_key=True) + task_id: Mapped[str] = mapped_column(String, ForeignKey("review_task.id"), nullable=False) + file_path: Mapped[str | None] = mapped_column(String, nullable=True) + sha256: Mapped[str | None] = mapped_column(String, nullable=True) + hunk_count: Mapped[int | None] = mapped_column(Integer, nullable=True) + line_count: Mapped[int | None] = mapped_column(Integer, nullable=True) + summary: Mapped[str | None] = mapped_column(Text, nullable=True) + + +class SandboxRun(CRBase): + """L4 sandbox execution evidence (replayable / auditable).""" + + __tablename__ = "sandbox_run" + __table_args__ = (Index("idx_sandbox_run_task", "task_id"),) + + id: Mapped[str] = mapped_column(String, primary_key=True) + task_id: Mapped[str] = mapped_column(String, ForeignKey("review_task.id"), nullable=False) + runtime: Mapped[str | None] = mapped_column(String, nullable=True) # local|container|cube + script: Mapped[str | None] = mapped_column(String, nullable=True) + status: Mapped[str | None] = mapped_column(String, nullable=True) # ok|timeout|failed|truncated + duration_ms: Mapped[int | None] = mapped_column(Integer, nullable=True) + exit_code: Mapped[int | None] = mapped_column(Integer, nullable=True) + output_bytes: Mapped[int | None] = mapped_column(Integer, nullable=True) + timed_out: Mapped[int | None] = mapped_column(Integer, nullable=True) # 0|1 (no native bool) + masked_count: Mapped[int | None] = mapped_column(Integer, nullable=True) + + +class Finding(CRBase): + """L5 structured findings (bucket enforces confidence tiering).""" + + __tablename__ = "finding" + # Composite index directly serves L5 dedupe: O(1) "same file/line/category". + __table_args__ = ( + Index("idx_finding_dedup", "task_id", "file", "line", "category"), + Index("idx_finding_task", "task_id"), + ) + + id: Mapped[str] = mapped_column(String, primary_key=True) + task_id: Mapped[str] = mapped_column(String, ForeignKey("review_task.id"), nullable=False) + severity: Mapped[str | None] = mapped_column(String, nullable=True) # critical|high|medium|low + category: Mapped[str | None] = mapped_column(String, nullable=True) + file: Mapped[str | None] = mapped_column(String, nullable=True) + line: Mapped[int | None] = mapped_column(Integer, nullable=True) + title: Mapped[str | None] = mapped_column(String, nullable=True) + evidence: Mapped[str | None] = mapped_column(Text, nullable=True) + recommendation: Mapped[str | None] = mapped_column(Text, nullable=True) + confidence: Mapped[float | None] = mapped_column(Float, nullable=True) # 0.0-1.0 + source: Mapped[str | None] = mapped_column(String, nullable=True) # rule|sandbox|llm + bucket: Mapped[str | None] = mapped_column(String, nullable=True) # findings|warnings|needs_human_review + + +class FilterBlock(CRBase): + """L3 filter governance blocks (deny / needs_human_review skip sandbox).""" + + __tablename__ = "filter_block" + __table_args__ = (Index("idx_filter_block_task", "task_id"),) + + id: Mapped[str] = mapped_column(String, primary_key=True) + task_id: Mapped[str] = mapped_column(String, ForeignKey("review_task.id"), nullable=False) + reason: Mapped[str | None] = mapped_column(String, nullable=True) # high-risk|forbidden-path|network|budget + target: Mapped[str | None] = mapped_column(String, nullable=True) + decision: Mapped[str | None] = mapped_column(String, nullable=True) # deny|needs_human_review + detail: Mapped[str | None] = mapped_column(Text, nullable=True) + + +class MonitorSummary(CRBase): + """Per-task telemetry rollup (severity flattened to columns for indexed reads).""" + + __tablename__ = "monitor_summary" + __table_args__ = (Index("idx_monitor_summary_task", "task_id"),) + + id: Mapped[str] = mapped_column(String, primary_key=True) + task_id: Mapped[str] = mapped_column(String, ForeignKey("review_task.id"), nullable=False) + total_duration_ms: Mapped[int | None] = mapped_column(Integer, nullable=True) + sandbox_duration_ms: Mapped[int | None] = mapped_column(Integer, nullable=True) + tool_calls: Mapped[int | None] = mapped_column(Integer, nullable=True) + blocks: Mapped[int | None] = mapped_column(Integer, nullable=True) + finding_count: Mapped[int | None] = mapped_column(Integer, nullable=True) + sev_critical: Mapped[int | None] = mapped_column(Integer, nullable=True) + sev_high: Mapped[int | None] = mapped_column(Integer, nullable=True) + sev_medium: Mapped[int | None] = mapped_column(Integer, nullable=True) + sev_low: Mapped[int | None] = mapped_column(Integer, nullable=True) + exception_types: Mapped[str | None] = mapped_column(Text, nullable=True) # JSON string + + +class ReviewReport(CRBase): + """Final report artefact paths + human-readable summary.""" + + __tablename__ = "review_report" + __table_args__ = (Index("idx_review_report_task", "task_id"),) + + id: Mapped[str] = mapped_column(String, primary_key=True) + task_id: Mapped[str] = mapped_column(String, ForeignKey("review_task.id"), nullable=False) + report_json_path: Mapped[str | None] = mapped_column(String, nullable=True) + report_md_path: Mapped[str | None] = mapped_column(String, nullable=True) + summary: Mapped[str | None] = mapped_column(Text, nullable=True) + created_at: Mapped[str | None] = mapped_column(String, nullable=True) + + +__all__ = [ + "CRBase", + "ReviewTask", + "InputDiff", + "SandboxRun", + "Finding", + "FilterBlock", + "MonitorSummary", + "ReviewReport", +] diff --git a/examples/skills_code_review_agent/agent/db/schema.sql b/examples/skills_code_review_agent/agent/db/schema.sql new file mode 100644 index 00000000..0ded3a33 --- /dev/null +++ b/examples/skills_code_review_agent/agent/db/schema.sql @@ -0,0 +1,100 @@ +-- Code Review Agent — schema (Phase 0) +-- Seven tables aggregated around `review_task`; `task_id` is the global FK. +-- All statements use `IF NOT EXISTS` so the script is idempotent. +-- See ../docs/skills_code_review_agent/ARCHITECTURE.md §5.2 for the contract. + +-- L6 master table: one row per review run. +CREATE TABLE IF NOT EXISTS review_task ( + id TEXT PRIMARY KEY, + created_at TEXT NOT NULL, + status TEXT NOT NULL, -- pending|running|done|failed + input_type TEXT NOT NULL, -- diff|repo|fixture + input_ref TEXT, + mode TEXT NOT NULL, -- dry-run|real + total_duration_ms INTEGER +); + +-- L1 parsed diff per changed file (summary only, never the full diff blob). +CREATE TABLE IF NOT EXISTS input_diff ( + id TEXT PRIMARY KEY, + task_id TEXT NOT NULL REFERENCES review_task(id), + file_path TEXT, + sha256 TEXT, + hunk_count INTEGER, + line_count INTEGER, + summary TEXT +); +CREATE INDEX IF NOT EXISTS idx_input_diff_task ON input_diff(task_id); + +-- L4 sandbox execution evidence (replayable / auditable). +CREATE TABLE IF NOT EXISTS sandbox_run ( + id TEXT PRIMARY KEY, + task_id TEXT NOT NULL REFERENCES review_task(id), + runtime TEXT, -- local|container|cube + script TEXT, + status TEXT, -- ok|timeout|failed|truncated + duration_ms INTEGER, + exit_code INTEGER, + output_bytes INTEGER, + timed_out INTEGER, -- 0|1 (SQLite has no native bool) + masked_count INTEGER +); +CREATE INDEX IF NOT EXISTS idx_sandbox_run_task ON sandbox_run(task_id); + +-- L5 structured findings (bucket enforces confidence tiering at schema level). +CREATE TABLE IF NOT EXISTS finding ( + id TEXT PRIMARY KEY, + task_id TEXT NOT NULL REFERENCES review_task(id), + severity TEXT, -- critical|high|medium|low + category TEXT, -- security|async|resource|tests|sensitive|db + file TEXT, + line INTEGER, + title TEXT, + evidence TEXT, + recommendation TEXT, + confidence REAL, -- 0.0-1.0 + source TEXT, -- rule|sandbox|llm + bucket TEXT -- findings|warnings|needs_human_review +); +-- Composite index directly serves L5 dedupe lookup: O(1) "same file/line/category". +CREATE INDEX IF NOT EXISTS idx_finding_dedup ON finding(task_id, file, line, category); +CREATE INDEX IF NOT EXISTS idx_finding_task ON finding(task_id); + +-- L3 filter governance blocks (deny / needs_human_review do not enter sandbox). +CREATE TABLE IF NOT EXISTS filter_block ( + id TEXT PRIMARY KEY, + task_id TEXT NOT NULL REFERENCES review_task(id), + reason TEXT, -- high-risk|forbidden-path|network|budget + target TEXT, + decision TEXT, -- deny|needs_human_review + detail TEXT +); +CREATE INDEX IF NOT EXISTS idx_filter_block_task ON filter_block(task_id); + +-- Per-task telemetry rollup (severity flattened to columns for indexed reads). +CREATE TABLE IF NOT EXISTS monitor_summary ( + id TEXT PRIMARY KEY, + task_id TEXT NOT NULL REFERENCES review_task(id), + total_duration_ms INTEGER, + sandbox_duration_ms INTEGER, + tool_calls INTEGER, + blocks INTEGER, + finding_count INTEGER, + sev_critical INTEGER, + sev_high INTEGER, + sev_medium INTEGER, + sev_low INTEGER, + exception_types TEXT -- JSON, e.g. {"timeout":2,"oom":1} +); +CREATE INDEX IF NOT EXISTS idx_monitor_summary_task ON monitor_summary(task_id); + +-- Final report artefact paths + human-readable summary. +CREATE TABLE IF NOT EXISTS review_report ( + id TEXT PRIMARY KEY, + task_id TEXT NOT NULL REFERENCES review_task(id), + report_json_path TEXT, + report_md_path TEXT, + summary TEXT, + created_at TEXT +); +CREATE INDEX IF NOT EXISTS idx_review_report_task ON review_report(task_id); diff --git a/examples/skills_code_review_agent/agent/db/storage.py b/examples/skills_code_review_agent/agent/db/storage.py new file mode 100644 index 00000000..45a5f879 --- /dev/null +++ b/examples/skills_code_review_agent/agent/db/storage.py @@ -0,0 +1,419 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Storage abstraction for the Code Review Agent (Phase 0, SDK-backed). + +Defines the :class:`ReviewStore` protocol — the single persistence surface +every downstream phase (P1–P6) talks to — plus the default +:class:`SQLiteStore` implementation backed by the SDK's generic +:class:`trpc_agent_sdk.storage.SqlStorage` layer. + +Design notes +------------ +* The seven business tables are declared in :mod:`db.models` as a custom + ``CRBase(DeclarativeBase)`` metadata and handed to + ``SqlStorage(metadata=CRBase.metadata)``. The SDK storage layer manages + table creation (``metadata.create_all``) and SQLite pragmas + (``PRAGMA foreign_keys=ON``) automatically. +* All methods are ``async`` — the SDK storage is async-first. Downstream + orchestration (``agent.py``) runs under ``asyncio``. +* Swapping SQLite for Postgres only requires a different ``db_url`` (e.g. + ``postgresql+asyncpg://...``) — the protocol and ORM stay unchanged. +* Identifiers are ``uuid4().hex``; ``executemany``-style bulk inserts use + ``session.add_all``. +""" + +from __future__ import annotations + +import json +import uuid +from datetime import datetime +from datetime import timezone +from pathlib import Path +from typing import Any +from typing import Protocol +from typing import runtime_checkable + +from sqlalchemy import delete as sa_delete +from sqlalchemy import select +from sqlalchemy import text +from sqlalchemy import update as sa_update +from sqlalchemy.pool import StaticPool + +from trpc_agent_sdk.storage import SqlCondition +from trpc_agent_sdk.storage import SqlKey +from trpc_agent_sdk.storage import SqlStorage + +from .models import CRBase +from .models import FilterBlock +from .models import Finding +from .models import InputDiff +from .models import MonitorSummary +from .models import ReviewReport +from .models import ReviewTask +from .models import SandboxRun + +# monitor_summary columns that map 1:1 from the summary dict (exception_types +# is JSON-encoded separately). +_MONITOR_COLUMNS = ( + "total_duration_ms", + "sandbox_duration_ms", + "tool_calls", + "blocks", + "finding_count", + "sev_critical", + "sev_high", + "sev_medium", + "sev_low", +) + + +def _now_iso() -> str: + """ISO-8601 UTC timestamp — single source for `created_at`.""" + return datetime.now(timezone.utc).isoformat(timespec="seconds") + + +def _new_id() -> str: + """Hex UUID4 — the id format for every table row.""" + return uuid.uuid4().hex + + +def _to_dict(obj) -> dict: + """Convert an ORM instance to a plain dict (column name → value).""" + if obj is None: + return None + return {c.name: getattr(obj, c.name) for c in obj.__table__.columns} + + +def _build_db_url(db_path: str) -> tuple[str, dict]: + """Return (db_url, extra_kwargs) for SqlStorage. + + ``:memory:`` needs ``StaticPool`` so every connection shares one DB + (async sqlite otherwise gives each connection its own private memory DB). + """ + if db_path == ":memory:": + return ( + "sqlite+aiosqlite://", + {"poolclass": StaticPool, "connect_args": {"check_same_thread": False}}, + ) + return f"sqlite+aiosqlite:///{db_path}", {} + + +@runtime_checkable +class ReviewStore(Protocol): + """Async persistence surface for one code-review pipeline run. + + All seven tables are written/read through this protocol. ``get_task`` + returns the fully joined record so a single call reconstructs an + entire review (task + inputs + sandbox runs + findings + filter blocks + + telemetry + report). + """ + + # —— task lifecycle —— + async def create_task(self, input_type: str, input_ref: str, mode: str) -> str: ... + async def update_task_status( + self, task_id: str, status: str, total_duration_ms: int | None = None + ) -> None: ... + + # —— child-table writes (each returns the new row id) —— + async def add_input_diff( + self, task_id, file_path, sha256, hunk_count, line_count, summary + ) -> str: ... + async def add_sandbox_run( + self, task_id, runtime, script, status, duration_ms, exit_code, + output_bytes, timed_out, masked_count, + ) -> str: ... + async def add_finding( + self, task_id, severity, category, file, line, title, evidence, + recommendation, confidence, source, bucket, + ) -> str: ... + async def add_filter_block( + self, task_id, reason, target, decision, detail + ) -> str: ... + async def set_monitor_summary(self, task_id: str, summary: dict) -> None: ... + async def set_report( + self, task_id: str, json_path: str, md_path: str, summary: str + ) -> str: ... + + # —— query —— + async def get_task(self, task_id: str) -> dict: ... + + +class SQLiteStore: + """Async :class:`ReviewStore` over a single SQLite file via SDK SqlStorage. + + Pass ``db_path=":memory:"`` for an ephemeral in-memory DB (tests). + """ + + def __init__(self, db_path: str | Path = "cr_agent.db"): + self.db_path = str(db_path) + self._db_url, self._extra_kwargs = _build_db_url(self.db_path) + self._storage = SqlStorage( + is_async=True, + db_url=self._db_url, + metadata=CRBase.metadata, + **self._extra_kwargs, + ) + self._engine_ready = False + + # ------------------------------------------------------------------ # + # engine / lifecycle + # ------------------------------------------------------------------ # + async def _ensure_engine(self) -> None: + """Create the async engine + tables (idempotent). + + Also explicitly enables ``PRAGMA foreign_keys=ON``: the SDK's + connect-event hook doesn't always fire for ``:memory:`` + StaticPool, + so we set it on the pooled connection to be safe. + """ + if not self._engine_ready: + await self._storage.create_sql_engine() + engine = self._storage._db_engine + async with engine.begin() as conn: + await conn.execute(text("PRAGMA foreign_keys=ON")) + self._engine_ready = True + + @property + def storage(self) -> SqlStorage: + """Underlying SDK storage (for advanced callers).""" + return self._storage + + async def close(self) -> None: + await self._storage.close() + self._engine_ready = False + + async def __aenter__(self) -> "SQLiteStore": + await self._ensure_engine() + return self + + async def __aexit__(self, exc_type, exc, tb) -> None: + await self.close() + + # ------------------------------------------------------------------ # + # internal helpers + # ------------------------------------------------------------------ # + def _session(self): + """An async session context manager (``async with store._session() as db``).""" + return self._storage.create_db_session() + + @staticmethod + def _query_all(db, model, task_id: str, order_col=None): + """Select all rows of ``model`` for ``task_id``, ordered.""" + stmt = select(model).where(model.task_id == task_id) + if order_col is not None: + stmt = stmt.order_by(order_col) + return db.execute(stmt).scalars().all() + + # ------------------------------------------------------------------ # + # task lifecycle + # ------------------------------------------------------------------ # + async def create_task(self, input_type: str, input_ref: str, mode: str) -> str: + await self._ensure_engine() + task_id = _new_id() + task = ReviewTask( + id=task_id, + created_at=_now_iso(), + status="pending", + input_type=input_type, + input_ref=input_ref, + mode=mode, + ) + async with self._session() as db: + await self._storage.add(db, task) + await self._storage.commit(db) + return task_id + + async def update_task_status( + self, task_id: str, status: str, total_duration_ms: int | None = None + ) -> None: + await self._ensure_engine() + async with self._session() as db: + await db.execute( + sa_update(ReviewTask) + .where(ReviewTask.id == task_id) + .values(status=status, total_duration_ms=total_duration_ms) + ) + await self._storage.commit(db) + + # ------------------------------------------------------------------ # + # child-table writes + # ------------------------------------------------------------------ # + async def add_input_diff( + self, task_id, file_path, sha256, hunk_count, line_count, summary + ) -> str: + await self._ensure_engine() + diff_id = _new_id() + obj = InputDiff( + id=diff_id, task_id=task_id, file_path=file_path, sha256=sha256, + hunk_count=hunk_count, line_count=line_count, summary=summary, + ) + async with self._session() as db: + await self._storage.add(db, obj) + await self._storage.commit(db) + return diff_id + + async def add_sandbox_run( + self, task_id, runtime, script, status, duration_ms, exit_code, + output_bytes, timed_out, masked_count, + ) -> str: + await self._ensure_engine() + run_id = _new_id() + obj = SandboxRun( + id=run_id, task_id=task_id, runtime=runtime, script=script, + status=status, duration_ms=duration_ms, exit_code=exit_code, + output_bytes=output_bytes, timed_out=timed_out, masked_count=masked_count, + ) + async with self._session() as db: + await self._storage.add(db, obj) + await self._storage.commit(db) + return run_id + + async def add_finding( + self, task_id, severity, category, file, line, title, evidence, + recommendation, confidence, source, bucket, + ) -> str: + await self._ensure_engine() + finding_id = _new_id() + obj = Finding( + id=finding_id, task_id=task_id, severity=severity, category=category, + file=file, line=line, title=title, evidence=evidence, + recommendation=recommendation, confidence=confidence, + source=source, bucket=bucket, + ) + async with self._session() as db: + await self._storage.add(db, obj) + await self._storage.commit(db) + return finding_id + + async def add_findings(self, task_id: str, findings: list[dict]) -> list[str]: + """Bulk-insert findings via ``session.add_all`` — one round-trip. + + A review can emit hundreds of findings, so this is preferred over + looping :meth:`add_finding`. + """ + if not findings: + return [] + await self._ensure_engine() + ids = [_new_id() for _ in findings] + rows = [ + Finding( + id=fid, task_id=task_id, + severity=f["severity"], category=f["category"], file=f["file"], + line=f["line"], title=f["title"], evidence=f["evidence"], + recommendation=f["recommendation"], confidence=f["confidence"], + source=f["source"], bucket=f["bucket"], + ) + for fid, f in zip(ids, findings) + ] + async with self._session() as db: + db.add_all(rows) + await self._storage.commit(db) + return ids + + async def add_filter_block( + self, task_id, reason, target, decision, detail + ) -> str: + await self._ensure_engine() + block_id = _new_id() + obj = FilterBlock( + id=block_id, task_id=task_id, reason=reason, target=target, + decision=decision, detail=detail, + ) + async with self._session() as db: + await self._storage.add(db, obj) + await self._storage.commit(db) + return block_id + + async def set_monitor_summary(self, task_id: str, summary: dict) -> None: + """Upsert the 1:1 telemetry rollup row for a task.""" + await self._ensure_engine() + exc = summary.get("exception_types") + if isinstance(exc, dict): + exc = json.dumps(exc, ensure_ascii=False) + async with self._session() as db: + # 1:1 relationship — delete any existing row first. + await db.execute( + sa_delete(MonitorSummary).where(MonitorSummary.task_id == task_id) + ) + summary_id = _new_id() + values = {col: summary.get(col) for col in _MONITOR_COLUMNS} + obj = MonitorSummary(id=summary_id, task_id=task_id, exception_types=exc, **values) + await self._storage.add(db, obj) + await self._storage.commit(db) + + async def set_report( + self, task_id: str, json_path: str, md_path: str, summary: str + ) -> str: + """Upsert the 1:1 report row for a task, return its id.""" + await self._ensure_engine() + async with self._session() as db: + existing = await db.execute( + select(ReviewReport).where(ReviewReport.task_id == task_id) + ) + row = existing.scalars().one_or_none() + if row is not None: + row.report_json_path = json_path + row.report_md_path = md_path + row.summary = summary + row.created_at = _now_iso() + report_id = row.id + else: + report_id = _new_id() + obj = ReviewReport( + id=report_id, task_id=task_id, report_json_path=json_path, + report_md_path=md_path, summary=summary, created_at=_now_iso(), + ) + await self._storage.add(db, obj) + await self._storage.commit(db) + return report_id + + # ------------------------------------------------------------------ # + # query + # ------------------------------------------------------------------ # + async def get_task(self, task_id: str) -> dict: + """Reconstruct the full review record via per-table queries. + + One session, one query per child table — keeps the result a clean + nested dict with stable ordering. + """ + await self._ensure_engine() + async with self._session() as db: + task_row = await db.execute( + select(ReviewTask).where(ReviewTask.id == task_id) + ) + task = task_row.scalars().one_or_none() + if task is None: + raise KeyError(f"review_task not found: {task_id}") + + input_diffs = (await db.execute( + select(InputDiff).where(InputDiff.task_id == task_id).order_by(InputDiff.id) + )).scalars().all() + sandbox_runs = (await db.execute( + select(SandboxRun).where(SandboxRun.task_id == task_id).order_by(SandboxRun.id) + )).scalars().all() + findings = (await db.execute( + select(Finding).where(Finding.task_id == task_id).order_by(Finding.id) + )).scalars().all() + filter_blocks = (await db.execute( + select(FilterBlock).where(FilterBlock.task_id == task_id).order_by(FilterBlock.id) + )).scalars().all() + ms_row = await db.execute( + select(MonitorSummary).where(MonitorSummary.task_id == task_id) + ) + monitor_summary = ms_row.scalars().one_or_none() + rep_row = await db.execute( + select(ReviewReport).where(ReviewReport.task_id == task_id) + ) + report = rep_row.scalars().one_or_none() + + return { + "task": _to_dict(task), + "input_diffs": [_to_dict(x) for x in input_diffs], + "sandbox_runs": [_to_dict(x) for x in sandbox_runs], + "findings": [_to_dict(x) for x in findings], + "filter_blocks": [_to_dict(x) for x in filter_blocks], + "monitor_summary": _to_dict(monitor_summary), + "report": _to_dict(report), + } diff --git a/examples/skills_code_review_agent/agent/filters/__init__.py b/examples/skills_code_review_agent/agent/filters/__init__.py new file mode 100644 index 00000000..33520bd9 --- /dev/null +++ b/examples/skills_code_review_agent/agent/filters/__init__.py @@ -0,0 +1,11 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Filters package — governance gate (Phase 3).""" +from .governance import CrGovernanceFilter +from .governance import FilterDecision +from .governance import FilterGovernance + +__all__ = ["FilterDecision", "FilterGovernance", "CrGovernanceFilter"] diff --git a/examples/skills_code_review_agent/agent/filters/governance.py b/examples/skills_code_review_agent/agent/filters/governance.py new file mode 100644 index 00000000..12d9a6d2 --- /dev/null +++ b/examples/skills_code_review_agent/agent/filters/governance.py @@ -0,0 +1,199 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Filter governance — pre-sandbox gate (Phase 3, L3). + +Four check classes run BEFORE a script enters the sandbox. ``deny`` and +``needs_human_review`` never reach the sandbox — the orchestrator (P5) +records the block in ``filter_block`` and skips execution. + +This module provides two layers: +* :class:`FilterGovernance` — the spec contract: a synchronous ``decide()`` + that inspects script content + budget and returns a :class:`FilterDecision`. +* :class:`CrGovernanceFilter` — an SDK :class:`BaseFilter` adapter that wraps + ``FilterGovernance`` so the same checks run when the Skill is invoked via + the SDK tool pipeline (``register_tool_filter``). The four check *strategies* + are self-implemented (the SDK ships no built-in policy filters). +""" +from __future__ import annotations + +import re +from dataclasses import dataclass +from pathlib import Path + +from trpc_agent_sdk.filter import BaseFilter +from trpc_agent_sdk.filter import FilterResult +from trpc_agent_sdk.filter import register_tool_filter + + +@dataclass +class FilterDecision: + """Outcome of a governance check.""" + + verdict: str # allow|deny|needs_human_review + reason: str # high-risk|forbidden-path|network|budget|ok + target: str # the script path / target inspected + detail: str + + +# --- high-risk script patterns (deny) --- +# Kept precise to avoid false positives on docs/comments: e.g. `rm -rf` only +# fires when followed by a path sigil (/, ~), not bare prose. +_HIGH_RISK_PATTERNS = [ + (re.compile(r"rm\s+-rf?\s+[/~]"), "rm -rf targeting a root/home path"), + (re.compile(r"\bsudo\b\s+\w"), "sudo invocation"), + (re.compile(r"\bcurl\b\s+https?://|\bwget\b\s+https?://"), "remote download+exec pattern (curl/wget)"), + (re.compile(r"\beval\s*\("), "eval() arbitrary code execution"), + (re.compile(r"os\.system\s*\(\s*['\"].*['\"]\s*\+"), "os.system with string concatenation (injection)"), + (re.compile(r"subprocess\.(?:call|run|Popen)\s*\([^)]*shell\s*=\s*True[^)]*\+"), "shell=True with concatenation (injection)"), +] + +# --- forbidden paths (deny) --- +_FORBIDDEN_PATH_PATTERNS = [ + re.compile(r"['\"]/etc/"), + re.compile(r"['\"]~/.ssh"), + re.compile(r"['\"]/?root/\."), + re.compile(r"['\"]/?var/log"), + re.compile(r"['\"]C:\\\\?Windows"), + re.compile(r"['\"]/?proc/"), +] + +# --- network access (needs_human_review) --- +_NETWORK_PATTERNS = [ + re.compile(r"\bsocket\.connect\s*\("), + re.compile(r"\brequests\.(?:get|post|put|delete|patch|head)\s*\("), + re.compile(r"\burllib\.request\.urlopen\s*\("), + re.compile(r"\baiohttp\.ClientSession\s*\("), + re.compile(r"\bhttpx\.(?:get|post|Client)\s*\("), +] +# Domains that are always allowed (won't trigger network review). +_NETWORK_WHITELIST = re.compile(r"localhost|127\.0\.0\.1|0\.0\.0\.0|::1") + +# Budget thresholds (over → needs_human_review). +_BUDGET_DURATION_S = 60 +_BUDGET_MEMORY_MB = 1024 + + +class FilterGovernance: + """Synchronous four-class governance check. + + Call :meth:`decide` with the script path + content + budget estimate. + The orchestrator records ``deny``/``needs_human_review`` into + ``filter_block`` and skips the sandbox for those verdicts. + """ + + def decide( + self, + script_path: str, + script_content: str, + budget: dict | None = None, + ) -> FilterDecision: + target = script_path + budget = budget or {} + + # Strip comment lines (first non-space char is '#') to reduce false + # positives — a doc/comment mentioning `rm -rf` should NOT trigger a + # deny. Simple heuristic; P6 can refine with AST-aware comment skip. + code = "\n".join( + ln for ln in script_content.splitlines() + if not ln.lstrip().startswith("#") + ) + + # 1. high-risk script features → deny + for pat, desc in _HIGH_RISK_PATTERNS: + m = pat.search(code) + if m: + return FilterDecision( + verdict="deny", + reason="high-risk", + target=target, + detail=f"{desc}: matched /{pat.pattern}/ → {m.group(0)!r}", + ) + + # 2. forbidden paths → deny + for pat in _FORBIDDEN_PATH_PATTERNS: + m = pat.search(code) + if m: + return FilterDecision( + verdict="deny", + reason="forbidden-path", + target=target, + detail=f"access to forbidden path: {m.group(0)!r}", + ) + + # 3. non-whitelisted network → needs_human_review + for pat in _NETWORK_PATTERNS: + m = pat.search(code) + if m: + # allow if the match context mentions a whitelisted host + window = code[max(0, m.start() - 40): m.end() + 60] + if _NETWORK_WHITELIST.search(window): + continue + return FilterDecision( + verdict="needs_human_review", + reason="network", + target=target, + detail=f"non-whitelisted network access: {m.group(0)!r}", + ) + + # 4. over-budget → needs_human_review + est_dur = float(budget.get("estimated_duration_s", 0) or 0) + est_mem = float(budget.get("estimated_memory_mb", 0) or 0) + if est_dur > _BUDGET_DURATION_S or est_mem > _BUDGET_MEMORY_MB: + return FilterDecision( + verdict="needs_human_review", + reason="budget", + target=target, + detail=f"over budget: duration={est_dur}s (>{_BUDGET_DURATION_S}), memory={est_mem}MB (>{_BUDGET_MEMORY_MB})", + ) + + return FilterDecision( + verdict="allow", reason="ok", target=target, detail="passed all governance checks" + ) + + +# --------------------------------------------------------------------------- # +# SDK BaseFilter adapter — same checks run when the Skill is invoked via the +# SDK tool pipeline. Registered as "cr_governance" tool filter. +# --------------------------------------------------------------------------- # +@register_tool_filter("cr_governance") +class CrGovernanceFilter(BaseFilter): + """SDK tool filter wrapping :class:`FilterGovernance`. + + When a skill_run/skill_exec tool call carries a script to execute, this + filter runs the four-class governance check in ``_before``. A ``deny`` + verdict sets ``is_continue=False`` so the call never reaches the sandbox; + ``needs_human_review`` likewise blocks (the orchestrator records it). + The block reason is surfaced via ``rsp.error``. + """ + + def __init__(self, governance: "FilterGovernance | None" = None): + self._gov = governance or FilterGovernance() + + async def _before(self, ctx, req, rsp: FilterResult) -> None: + # Extract script info from the tool-call request. The exact req shape + # depends on the SDK tool pipeline; we try common accessors and fall + # back to allow when no script is identifiable (P5 orchestrator wires + # the real script_path into the request args). + script_path = None + script_content = None + for attr in ("script_path", "path", "file"): + script_path = getattr(req, attr, None) + if script_path: + break + if script_path is None and isinstance(req, dict): + script_path = req.get("script_path") or req.get("path") + if script_path is None: + return # nothing to inspect → allow + try: + script_content = Path(script_path).read_text(encoding="utf-8") + except Exception: + script_content = "" + decision = self._gov.decide(str(script_path), script_content, {}) + if decision.verdict in ("deny", "needs_human_review"): + rsp.is_continue = False + rsp.error = ValueError( + f"[cr_governance] {decision.verdict}: {decision.reason} — {decision.detail}" + ) diff --git a/examples/skills_code_review_agent/agent/llm/__init__.py b/examples/skills_code_review_agent/agent/llm/__init__.py new file mode 100644 index 00000000..cda2fdbf --- /dev/null +++ b/examples/skills_code_review_agent/agent/llm/__init__.py @@ -0,0 +1,22 @@ +"""LLM integration for the Code Review Agent (Phase 7, optional). + +Real model calls are OFF by default. Configuration lives in ``.env`` +(see ``.env.example``). When ``LLM_ENABLED`` is false or no API key is +present, :func:`get_llm_client` returns a :class:`FakeLlm` that is a safe +no-op — the deterministic parse -> sandbox -> persist chain (and +``dry-run``) keep working with **zero model dependency**, so tests without +a real API key still exercise parsing, the sandbox, and the storage layer. +""" +from .client import FakeLlm, LlmClient, RealLlm, get_llm_client +from .config import LlmConfig, load_llm_config +from .triage import LlmTriage + +__all__ = [ + "LlmConfig", + "load_llm_config", + "LlmClient", + "FakeLlm", + "RealLlm", + "get_llm_client", + "LlmTriage", +] diff --git a/examples/skills_code_review_agent/agent/llm/client.py b/examples/skills_code_review_agent/agent/llm/client.py new file mode 100644 index 00000000..303605e0 --- /dev/null +++ b/examples/skills_code_review_agent/agent/llm/client.py @@ -0,0 +1,179 @@ +"""LLM client + factory (Phase 7). + +The real client wraps the OpenAI SDK, which is OpenAI-compatible — it works +with OpenAI, Azure OpenAI, local vLLM, and internal gateways by simply +changing ``LLM_BASE_URL``/``LLM_API_KEY`` in ``.env``. The fake client is a +safe no-op used when the model is disabled or unconfigured. +""" +from __future__ import annotations + +import json +import logging +import re +from abc import ABC, abstractmethod +from typing import Optional + +from .config import LlmConfig, load_llm_config + +logger = logging.getLogger("cr_agent.llm") + + +class LlmClient(ABC): + """Async client that judges low-confidence candidate findings.""" + + is_enabled: bool = True + kind: str = "base" + + @abstractmethod + async def triage(self, findings, diff_text: str) -> list[dict]: + """Return verdicts: list of + ``{"index": int, "verdict": "real"|"false_positive", + "confidence": float, "explanation": str}``. + + An empty list means "no change" (degrade to no-op). + """ + ... + + +class FakeLlm(LlmClient): + """No-op client — used when the model is disabled / no API key.""" + + is_enabled = False + kind = "fake" + + async def triage(self, findings, diff_text: str) -> list[dict]: + return [] + + +_SYSTEM_PROMPT = ( + "You are a senior static-analysis code reviewer. You will receive a code " + "diff and a list of candidate issues found by automated rules. For each " + "candidate, decide whether it is a REAL issue or a FALSE POSITIVE. " + "Respond with STRICT JSON only: " + '{"verdicts":[{"index":int,"verdict":"real"|"false_positive",' + '"confidence":float 0-1,"explanation":str}]}. ' + "Only include entries you can judge with confidence; omit uncertain ones " + "(they stay unchanged). 'confidence' should reflect your own certainty " + "for real issues." +) + +_USER_TEMPLATE = ( + "## Code diff (secrets masked)\n" + "```\n{diff}\n```\n\n" + "## Candidate issues (index = position)\n" + "{items}\n\n" + "Return the JSON verdicts now." +) + + +class RealLlm(LlmClient): + """Real OpenAI-compatible client. Lazily imports ``openai``.""" + + kind = "real" + + def __init__(self, config: LlmConfig, client=None): + self.config = config + self._client = client + + def _make_client(self): + from openai import AsyncOpenAI + + kwargs = { + "api_key": self.config.api_key, + "timeout": self.config.timeout, + } + if self.config.base_url: + kwargs["base_url"] = self.config.base_url + return AsyncOpenAI(**kwargs) + + @property + def client(self): + if self._client is None: + self._client = self._make_client() + return self._client + + async def _chat(self, messages): + resp = await self.client.chat.completions.create( + model=self.config.model, + messages=messages, + temperature=self.config.temperature, + max_tokens=self.config.max_tokens, + ) + return resp + + async def triage(self, findings, diff_text: str) -> list[dict]: + if not findings: + return [] + payload = [ + { + "index": i, + "file": f.file, + "line": f.line, + "category": f.category, + "title": f.title, + "evidence": f.evidence, + } + for i, f in enumerate(findings) + ] + messages = [ + {"role": "system", "content": _SYSTEM_PROMPT}, + { + "role": "user", + "content": _USER_TEMPLATE.format( + diff=diff_text, + items=json.dumps(payload, ensure_ascii=False, indent=2), + ), + }, + ] + try: + resp = await self._chat(messages) + content = resp.choices[0].message.content or "" + return _parse_verdicts(content) + except Exception as exc: # network / parse / timeout -> degrade + logger.warning("llm triage failed, degrading to no-op: %s", exc) + return [] + + +def _parse_verdicts(content: str) -> list[dict]: + """Best-effort JSON parse of the model response (handles code fences).""" + try: + s = content.strip() + if s.startswith("```"): + s = re.sub(r"^```[a-zA-Z]*\n?", "", s) + s = s.rstrip("`").strip() + data = json.loads(s) + vs = data.get("verdicts") or [] + except Exception: + return [] + out: list[dict] = [] + for v in vs: + try: + idx = int(v["index"]) + verdict = str(v.get("verdict", "real")).lower() + conf = float(v.get("confidence", 0.5)) + conf = max(0.0, min(1.0, conf)) + out.append( + { + "index": idx, + "verdict": verdict, + "confidence": conf, + "explanation": str(v.get("explanation", "")), + } + ) + except (KeyError, ValueError, TypeError): + continue + return out + + +def get_llm_client(enable=None, env_path: Optional[str] = None) -> LlmClient: + """Return a :class:`RealLlm` when enabled + keyed, else :class:`FakeLlm`. + + ``enable`` (e.g. from ``--enable-llm``) overrides the ``.env`` flag when + not ``None``. Without an API key the real client is never instantiated, + guaranteeing the no-key test path stays model-free. + """ + cfg = load_llm_config(env_path) + on = cfg.enabled if enable is None else enable + if not on or not cfg.has_key: + return FakeLlm() + return RealLlm(cfg) diff --git a/examples/skills_code_review_agent/agent/llm/config.py b/examples/skills_code_review_agent/agent/llm/config.py new file mode 100644 index 00000000..1cdb7aea --- /dev/null +++ b/examples/skills_code_review_agent/agent/llm/config.py @@ -0,0 +1,62 @@ +"""LLM configuration loaded from ``.env`` (Phase 7). + +All configuration is sourced from environment variables (a ``.env`` file +is auto-loaded via ``python-dotenv`` from the project root). Missing or +empty values fall back to safe defaults so the agent stays runnable with +no model configured. +""" +from __future__ import annotations + +import os +from dataclasses import dataclass +from pathlib import Path + +from dotenv import load_dotenv + +# Default location: /.env (llm/ lives under agent/llm). +_PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent +_DEFAULT_ENV = _PROJECT_ROOT / ".env" + + +@dataclass +class LlmConfig: + """Resolved LLM settings.""" + + enabled: bool + provider: str + api_key: str + base_url: str + model: str + temperature: float + max_tokens: int + timeout: int + + @property + def has_key(self) -> bool: + return bool(self.api_key.strip()) + + +def _truthy(value: str | None) -> bool: + return str(value or "").strip().lower() in ("1", "true", "yes", "on") + + +def load_llm_config(env_path: str | None = None) -> LlmConfig: + """Read LLM_* settings from the environment (auto-loading ``.env``). + + ``env_path`` lets callers point at an explicit ``.env`` (e.g. via the + ``--llm-env`` CLI flag); otherwise the project-root ``.env`` is used. + """ + dotenv_path = _DEFAULT_ENV if env_path is None else os.path.expanduser(env_path) + # override=False so an already-exported real env var wins over the file. + load_dotenv(dotenv_path=dotenv_path, override=False) + + return LlmConfig( + enabled=_truthy(os.getenv("LLM_ENABLED")), + provider=(os.getenv("LLM_PROVIDER") or "openai").strip(), + api_key=(os.getenv("LLM_API_KEY") or "").strip(), + base_url=(os.getenv("LLM_BASE_URL") or "").strip(), + model=(os.getenv("LLM_MODEL") or "gpt-4o-mini").strip(), + temperature=float(os.getenv("LLM_TEMPERATURE") or "0.1"), + max_tokens=int(os.getenv("LLM_MAX_TOKENS") or "1024"), + timeout=int(os.getenv("LLM_TIMEOUT") or "30"), + ) diff --git a/examples/skills_code_review_agent/agent/llm/triage.py b/examples/skills_code_review_agent/agent/llm/triage.py new file mode 100644 index 00000000..558d24fe --- /dev/null +++ b/examples/skills_code_review_agent/agent/llm/triage.py @@ -0,0 +1,73 @@ +"""LLM second-opinion triage over low-confidence findings (Phase 7). + +Runs *after* dedupe and *before* persistence. The model only sees the +``needs_human_review`` bucket (low confidence). Verdicts are applied to: + +* **false_positive** -> dropped entirely +* **real** -> confidence updated, ``source`` tagged ``+llm``, explanation + appended to ``recommendation``, and re-bucketed by the (model-given) + confidence (>=0.8 -> findings, >=0.6 -> warnings, else stays in review). + +When the client is disabled or the call fails, the original +:class:`DedupeResult` is returned unchanged (no-op degradation). +""" +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from cr_models import DedupeResult + from .client import LlmClient + + +def _apply_verdicts(dr: "DedupeResult", verdicts: list[dict]) -> "DedupeResult": + by_idx = {v["index"]: v for v in verdicts} + kept: list = [] + promoted_findings: list = [] + promoted_warnings: list = [] + + for i, f in enumerate(dr.needs_human_review): + v = by_idx.get(i) + if v is None: + kept.append(f) + continue + if v.get("verdict") == "false_positive": + continue # drop + conf = v.get("confidence", f.confidence) + f.confidence = max(0.0, min(1.0, conf)) + if "llm" not in f.source: + f.source = (f.source + "+llm") if f.source else "llm" + expl = (v.get("explanation") or "").strip() + if expl: + f.recommendation = f.recommendation + f"\n[LLM] {expl}" + if f.confidence >= 0.8: + promoted_findings.append(f) + elif f.confidence >= 0.6: + promoted_warnings.append(f) + else: + kept.append(f) + + dr.findings = list(dr.findings) + promoted_findings + dr.warnings = list(dr.warnings) + promoted_warnings + dr.needs_human_review = kept + return dr + + +class LlmTriage: + """Drives the optional LLM second-opinion pass.""" + + def __init__(self, client: "LlmClient"): + self.client = client + + async def run(self, dedupe_result: "DedupeResult", diff_text: str) -> "DedupeResult": + nh = dedupe_result.needs_human_review + if not nh or not self.client.is_enabled: + return dedupe_result + # Mask secrets before sending the diff to the model. + from mask_secrets import mask_secrets + + masked, _ = mask_secrets(diff_text or "") + verdicts = await self.client.triage(nh, masked) + if not verdicts: + return dedupe_result + return _apply_verdicts(dedupe_result, verdicts) diff --git a/examples/skills_code_review_agent/agent/sandbox/__init__.py b/examples/skills_code_review_agent/agent/sandbox/__init__.py new file mode 100644 index 00000000..518b0a09 --- /dev/null +++ b/examples/skills_code_review_agent/agent/sandbox/__init__.py @@ -0,0 +1,27 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Sandbox package — isolated script execution (Phase 3).""" +from .policy import SandboxPolicy +from .runtime import ContainerRuntime +from .runtime import CubeRuntime +from .runtime import LocalRuntime +from .runtime import RunResult +from .runtime import RuntimeUnavailable +from .runtime import SandboxRuntime +from .runtime import build_runtime_with_fallback +from .runtime import select_runtime + +__all__ = [ + "SandboxPolicy", + "RunResult", + "SandboxRuntime", + "LocalRuntime", + "ContainerRuntime", + "CubeRuntime", + "RuntimeUnavailable", + "select_runtime", + "build_runtime_with_fallback", +] diff --git a/examples/skills_code_review_agent/agent/sandbox/policy.py b/examples/skills_code_review_agent/agent/sandbox/policy.py new file mode 100644 index 00000000..1bcdef06 --- /dev/null +++ b/examples/skills_code_review_agent/agent/sandbox/policy.py @@ -0,0 +1,53 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Sandbox security policy (Phase 3). + +The five mandatory safety boundaries applied on every sandbox execution: +timeout, output-size cap, env-variable whitelist, secret masking, and +fail-safe (no crash on error). Built from the Skill's ``sandbox_config``. +""" +from __future__ import annotations + +from dataclasses import dataclass +from dataclasses import field + + +@dataclass +class SandboxPolicy: + """Five mandatory sandbox safety boundaries. + + Defaults match the code-review Skill's ``sandbox:`` frontmatter + (ARCHITECTURE.md §6.3). A policy is built per-run from the loaded + Skill config so the boundaries are always enforced, never bypassed. + """ + + timeout_s: int = 30 + """Max execution time; over-run → ``timed_out=True``, degraded to empty.""" + + max_output_bytes: int = 1_048_576 + """1 MB output cap; excess is truncated, ``status=truncated``.""" + + env_whitelist: list[str] = field( + default_factory=lambda: ["PATH", "HOME", "LANG"] + ) + """Only these env vars are passed into the sandbox; others dropped.""" + + mask_secrets: bool = True + """Redact secrets in stdout/stderr before truncation & persistence.""" + + @classmethod + def from_config(cls, sandbox_config: dict | None) -> "SandboxPolicy": + """Build a policy from the Skill's ``sandbox_config`` dict. + + Unknown keys are ignored; missing keys fall back to the safe defaults. + """ + cfg = sandbox_config or {} + return cls( + timeout_s=int(cfg.get("timeout_s", 30)), + max_output_bytes=int(cfg.get("max_output_bytes", 1_048_576)), + env_whitelist=list(cfg.get("env_whitelist", ["PATH", "HOME", "LANG"])), + mask_secrets=True, + ) diff --git a/examples/skills_code_review_agent/agent/sandbox/runtime.py b/examples/skills_code_review_agent/agent/sandbox/runtime.py new file mode 100644 index 00000000..f7c096d5 --- /dev/null +++ b/examples/skills_code_review_agent/agent/sandbox/runtime.py @@ -0,0 +1,613 @@ +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. +# +# Copyright (C) 2026 Tencent. All rights reserved. +# +# tRPC-Agent-Python is licensed under Apache-2.0. +"""Sandbox runtime — isolated script execution (Phase 3, L4). + +Adapter over the SDK's execution backends: the code-review scripts +(``run_checks.py`` etc.) are staged into an isolated workspace and executed, +returning raw stdout/stderr/exit_code. We map that to the CR Agent's +:class:`RunResult` and apply the five safety boundaries from +:class:`SandboxPolicy` (timeout, output cap, env whitelist, secret masking, +fail-safe). + +Runtime selection (G1 fix) +-------------------------- +The skill's ``sandbox_config`` declares ``default_runtime`` / ``fallback`` +(e.g. ``default_runtime: container``, ``fallback: local``). The agent no +longer hardcodes ``LocalRuntime`` — it honors that contract via +:func:`build_runtime_with_fallback`, which probes ``default_runtime`` first +and transparently degrades to ``fallback`` when the requested backend cannot +be provisioned (e.g. the docker daemon is absent). The *actual* runtime used +is returned so the pipeline can record it truthfully in ``sandbox_run``. + +Backends +-------- +* ``LocalRuntime`` — SDK ``create_local_workspace_runtime`` (process isolation, + dev / fallback). No external dependency. +* ``ContainerRuntime`` — SDK ``ContainerClient`` (real docker isolation). + Raises :class:`RuntimeUnavailable` when docker is not reachable so the agent + can fall back to ``LocalRuntime``. +* ``CubeRuntime`` — SDK ``CubeSandboxClient`` (remote Cube/E2B sandbox). Raises + :class:`RuntimeUnavailable` when the cube extra / credentials are missing. + +Shared notes +------------ +* The whole script directory is staged (not just one file) so the script can + ``import`` sibling modules (``run_checks.py`` imports ``parse_diff``). +* Secrets are masked **before** truncation so masking never misses bytes. +* Any exception is caught → ``status="failed"`` with empty stdout; the + pipeline never crashes on a sandbox failure. +""" +from __future__ import annotations + +import asyncio +import base64 +import json +import logging +import os +import re +import time +import uuid +from dataclasses import dataclass +from pathlib import Path +from typing import Protocol +from typing import runtime_checkable + +# Wire the skill scripts dir onto sys.path so we can import mask_secrets +# (shared with the sensitive_info rule set) for output redaction. +_HERE = Path(__file__).resolve().parent # .../agent/sandbox +_SCRIPTS_DIR = _HERE.parent.parent / "skills" / "code-review" / "scripts" +if str(_SCRIPTS_DIR) not in __import__("sys").path: + __import__("sys").path.insert(0, str(_SCRIPTS_DIR)) +from mask_secrets import mask_secrets # noqa: E402 + +from .policy import SandboxPolicy # noqa: E402 + +logger = logging.getLogger(__name__) + + +class RuntimeUnavailable(RuntimeError): + """The requested sandbox runtime cannot be provisioned in this environment. + + Raised by a runtime's ``ensure_available()`` probe (e.g. docker daemon + absent, cube not configured). The agent catches it and falls back to the + configured ``fallback`` runtime. + """ + + +@dataclass +class RunResult: + """One sandbox execution outcome, persisted to ``sandbox_run``.""" + + status: str # ok|timeout|failed|truncated + stdout: str + stderr: str + duration_ms: int + exit_code: int + output_bytes: int + timed_out: bool + masked_count: int + + +@runtime_checkable +class SandboxRuntime(Protocol): + """Execute a script in an isolated workspace under a policy.""" + + async def run( + self, script_path: str, input: dict, policy: "SandboxPolicy" + ) -> RunResult: ... + + +# --------------------------------------------------------------------------- # +# Shared post-processing — mask + truncate + status mapping (all backends) +# --------------------------------------------------------------------------- # +def _finalize( + stdout: str, + stderr: str, + exit_code: int, + timed_out: bool, + duration_s: float, + policy: SandboxPolicy, +) -> RunResult: + """Apply the safety boundaries shared by every backend and build RunResult.""" + masked_count = 0 + if policy.mask_secrets: + stdout, n1 = mask_secrets(stdout) + stderr, n2 = mask_secrets(stderr) + masked_count = n1 + n2 + + # Truncate to max_output_bytes (applied to stdout only; stderr kept). + truncated = len(stdout.encode("utf-8")) > policy.max_output_bytes + if truncated: + stdout = stdout.encode("utf-8")[: policy.max_output_bytes].decode( + "utf-8", errors="ignore" + ) + + # Map to RunResult.status: timeout > truncated > failed > ok. + if timed_out: + status = "timeout" + elif truncated: + status = "truncated" + elif exit_code != 0: + status = "failed" + else: + status = "ok" + + return RunResult( + status=status, + stdout=stdout, + stderr=stderr, + duration_ms=int(round((duration_s or 0) * 1000)), + exit_code=exit_code, + output_bytes=len(stdout.encode("utf-8")), + timed_out=bool(timed_out), + masked_count=masked_count, + ) + + +# Non-secret, platform-essential environment variables that the OS / python +# interpreter needs merely to *start* a subprocess on the host. Passing these +# into the local sandbox does NOT weaken the whitelist security boundary — +# the boundary is about not leaking arbitrary host vars (secrets, config, +# credentials). These are generic, non-sensitive process-start requirements. +_SYSTEM_REQUIRED = ( + "PATH", "SYSTEMROOT", "SYSTEMDRIVE", "WINDIR", "TEMP", "TMP", + "COMSPEC", "USERPROFILE", "USERNAME", "USER", "HOME", "LANG", + "LC_ALL", "PATHEXT", "PROCESSOR_ARCHITECTURE", "NUMBER_OF_PROCESSORS", + "LOGNAME", "TERM", +) + + +def _restrict_env_to_whitelist(policy: "SandboxPolicy") -> dict: + """Temporarily shrink ``os.environ`` to the whitelist + platform-required vars. + + The SDK local runtime builds the child env from ``os.environ.copy()`` and + then merges ``spec.env`` — so the *only* way to drop non-whitelisted host + variables is to shrink ``os.environ`` itself for the duration of the + subprocess. We keep exactly: + + * variables named in ``policy.env_whitelist`` (the Skill's declared + contract), and + * a small set of non-secret platform-required vars (``_SYSTEM_REQUIRED``) + without which python cannot even start on the host OS. + + Everything else (including ``CR_P3_VISIBLE``, ``OPENAI_API_KEY``, …) is + dropped, so it can never leak into the sandbox. The full host env is always + restored by the caller's ``finally`` block. CR Agent runs sandboxes + serially (awaited), so this global mutation is safe. + """ + _saved = dict(os.environ) + allowed = { + k: v for k, v in os.environ.items() + if k in policy.env_whitelist or k in _SYSTEM_REQUIRED + } + os.environ.clear() + os.environ.update(allowed) + return _saved + + +def _restore_env(saved: dict) -> None: + """Restore ``os.environ`` from a previously saved copy, tolerating vars + that the OS cannot hold. + + On Windows, ``os.environ`` rejects any value longer than 32767 characters + (``SetEnvironmentVariableW`` limit) — yet a process can *inherit* such a + var from its parent. Some desktop-agent environments ship a multi-hundred- + KB context var (e.g. ``ACC_PRODUCT_CONFIG_V3``). We must restore the host + env after a sandbox run, but re-setting an oversized var raises + ``ValueError``. We set each var individually and silently skip the ones the + OS refuses, so the restore never crashes the pipeline (the un-settable var + was already dropped during restriction and the sandbox never needed it). + """ + os.environ.clear() + for k, v in saved.items(): + try: + os.environ[k] = v + except (ValueError, OSError): + continue + + +# --------------------------------------------------------------------------- # +# LocalRuntime — SDK create_local_workspace_runtime +# --------------------------------------------------------------------------- # +class LocalRuntime: + """Sandbox backed by the SDK's local workspace runtime (process isolation). + + Creates a fresh workspace per run, stages the script directory, runs + ``python