|
| 1 | +# Tencent is pleased to support the open source community by making tRPC-Agent-Python available. |
| 2 | +# |
| 3 | +# Copyright (C) 2026 Tencent. All rights reserved. |
| 4 | +# |
| 5 | +# tRPC-Agent-Python is licensed under Apache-2.0. |
| 6 | + |
| 7 | +"""Sandboxed execution of the review scanners (issue #92, requirement 4 & slice 2). |
| 8 | +
|
| 9 | +Runs the skill's ``run_checks.py`` outside the review process, with a timeout and an output-size cap, |
| 10 | +and records a ``SandboxRunResult`` for every run — including timeouts and failures — so one bad run |
| 11 | +degrades a source without crashing the task. |
| 12 | +
|
| 13 | +Two runtimes: |
| 14 | +- ``run_local``: subprocess execution (the dev fallback the issue permits) — real process boundary, |
| 15 | + timeout and output cap; verified without Docker. |
| 16 | +- ``run_container``: production isolation via the framework's Container workspace runtime; requires |
| 17 | + Docker and the scanner image (see ``skills/code-review/Dockerfile``). |
| 18 | +""" |
| 19 | +from __future__ import annotations |
| 20 | + |
| 21 | +import json |
| 22 | +import subprocess |
| 23 | +import sys |
| 24 | +import tempfile |
| 25 | +import time |
| 26 | +from pathlib import Path |
| 27 | + |
| 28 | +from .types import Finding, SandboxRunResult |
| 29 | + |
| 30 | +_SKILL_SCRIPT = Path(__file__).resolve().parents[3] / "skills" / "code-review" / "scripts" / "run_checks.py" |
| 31 | +DEFAULT_TIMEOUT_SEC = 60.0 |
| 32 | +MAX_OUTPUT_BYTES = 1_048_576 # 1 MiB per stream |
| 33 | + |
| 34 | + |
| 35 | +def _truncate(text: str | bytes | None, cap: int) -> tuple[str, int]: |
| 36 | + """Return (possibly-truncated text, original byte length). The cap bounds what we persist.""" |
| 37 | + if text is None: |
| 38 | + return "", 0 |
| 39 | + raw = text.encode("utf-8", "replace") if isinstance(text, str) else text |
| 40 | + n = len(raw) |
| 41 | + if n <= cap: |
| 42 | + return raw.decode("utf-8", "ignore"), n |
| 43 | + return raw[:cap].decode("utf-8", "ignore") + "\n...[truncated]", n |
| 44 | + |
| 45 | + |
| 46 | +def parse_findings_json(payload: dict) -> list[Finding]: |
| 47 | + """Map the skill's findings.json (docs/OUTPUT_SCHEMA.md) into Finding objects.""" |
| 48 | + out: list[Finding] = [] |
| 49 | + for f in payload.get("findings", []): |
| 50 | + try: |
| 51 | + out.append( |
| 52 | + Finding(severity=f.get("severity", "low"), |
| 53 | + category=f.get("category", "unknown"), |
| 54 | + file=f.get("file", ""), |
| 55 | + line=f.get("line"), |
| 56 | + title=f.get("title", ""), |
| 57 | + evidence=f.get("evidence", ""), |
| 58 | + recommendation=f.get("recommendation", ""), |
| 59 | + confidence=float(f.get("confidence", 0.5)), |
| 60 | + source=f.get("source", "static"), |
| 61 | + rule_id=f.get("rule_id"))) |
| 62 | + except Exception: # noqa: BLE001 - a malformed row is skipped, not fatal |
| 63 | + continue |
| 64 | + return out |
| 65 | + |
| 66 | + |
| 67 | +def run_local( |
| 68 | + scan_dir: str, |
| 69 | + *, |
| 70 | + timeout: float = DEFAULT_TIMEOUT_SEC, |
| 71 | + max_bytes: int = MAX_OUTPUT_BYTES, |
| 72 | +) -> tuple[list[Finding], SandboxRunResult]: |
| 73 | + """Run the scanners in a subprocess against ``scan_dir``; never raises.""" |
| 74 | + out_file = Path(tempfile.mkdtemp(prefix="cr_out_")) / "findings.json" |
| 75 | + cmd = [sys.executable, str(_SKILL_SCRIPT), "--target", scan_dir, "--out", str(out_file)] |
| 76 | + |
| 77 | + started = time.monotonic() |
| 78 | + timed_out = False |
| 79 | + exit_code = 0 |
| 80 | + stdout: str | bytes | None = "" |
| 81 | + stderr: str | bytes | None = "" |
| 82 | + try: |
| 83 | + proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, check=False) |
| 84 | + exit_code, stdout, stderr = proc.returncode, proc.stdout, proc.stderr |
| 85 | + except subprocess.TimeoutExpired as exc: |
| 86 | + timed_out, exit_code = True, -1 |
| 87 | + stdout, stderr = exc.stdout, exc.stderr |
| 88 | + duration = time.monotonic() - started |
| 89 | + |
| 90 | + findings: list[Finding] = [] |
| 91 | + if not timed_out and out_file.exists(): |
| 92 | + try: |
| 93 | + findings = parse_findings_json(json.loads(out_file.read_text(encoding="utf-8"))) |
| 94 | + except Exception: # noqa: BLE001 - unreadable output degrades the source, not the task |
| 95 | + findings = [] |
| 96 | + |
| 97 | + _, out_bytes = _truncate(stdout, max_bytes) |
| 98 | + _, err_bytes = _truncate(stderr, max_bytes) |
| 99 | + result = SandboxRunResult(script="run_checks.py", |
| 100 | + exit_code=exit_code, |
| 101 | + duration_sec=round(duration, 3), |
| 102 | + timed_out=timed_out, |
| 103 | + stdout_bytes=out_bytes, |
| 104 | + stderr_bytes=err_bytes) |
| 105 | + return findings, result |
| 106 | + |
| 107 | + |
| 108 | +async def run_container( |
| 109 | + scan_dir: str, |
| 110 | + *, |
| 111 | + timeout: float = DEFAULT_TIMEOUT_SEC, |
| 112 | + max_bytes: int = MAX_OUTPUT_BYTES, |
| 113 | + memory_mb: int = 512, |
| 114 | + image: str = "cr-scanners:latest", |
| 115 | +) -> tuple[list[Finding], SandboxRunResult]: |
| 116 | + """Production isolation: run the scanners inside a container workspace. Requires Docker + image. |
| 117 | +
|
| 118 | + Stages the changed files into the workspace, runs ``run_checks.py`` with a timeout and resource |
| 119 | + limits, collects ``findings.json``, and truncates captured output to the byte cap. |
| 120 | + """ |
| 121 | + from trpc_agent_sdk.code_executors import (WorkspaceOutputSpec, WorkspacePutFileInfo, WorkspaceResourceLimits, |
| 122 | + WorkspaceRunProgramSpec, create_container_workspace_runtime) |
| 123 | + from trpc_agent_sdk.code_executors.container import ContainerConfig |
| 124 | + |
| 125 | + runtime = create_container_workspace_runtime(container_config=ContainerConfig(image=image)) |
| 126 | + manager, fs, runner = runtime.manager(None), runtime.fs(None), runtime.runner(None) |
| 127 | + exec_id = "cr-" + Path(scan_dir).name |
| 128 | + |
| 129 | + started = time.monotonic() |
| 130 | + ws = await manager.create_workspace(exec_id) |
| 131 | + try: |
| 132 | + files = [ |
| 133 | + WorkspacePutFileInfo(path=str(p.relative_to(scan_dir)), content=p.read_bytes()) |
| 134 | + for p in Path(scan_dir).rglob("*") if p.is_file() |
| 135 | + ] |
| 136 | + await fs.put_files(ws, files) |
| 137 | + run = await runner.run_program( |
| 138 | + ws, |
| 139 | + WorkspaceRunProgramSpec(cmd="python", |
| 140 | + args=["/opt/skill/run_checks.py", "--target", ".", "--out", "findings.json"], |
| 141 | + timeout=timeout, |
| 142 | + limits=WorkspaceResourceLimits(memory_mb=memory_mb))) |
| 143 | + collected = await fs.collect_outputs(ws, WorkspaceOutputSpec(globs=["findings.json"])) |
| 144 | + findings: list[Finding] = [] |
| 145 | + for cf in getattr(collected, "files", collected) or []: |
| 146 | + try: |
| 147 | + findings = parse_findings_json(json.loads(cf.content.decode("utf-8"))) |
| 148 | + except Exception: # noqa: BLE001 |
| 149 | + continue |
| 150 | + _, out_bytes = _truncate(run.stdout, max_bytes) |
| 151 | + _, err_bytes = _truncate(run.stderr, max_bytes) |
| 152 | + result = SandboxRunResult(script="run_checks.py", |
| 153 | + exit_code=run.exit_code, |
| 154 | + duration_sec=round(time.monotonic() - started, 3), |
| 155 | + timed_out=run.timed_out, |
| 156 | + stdout_bytes=out_bytes, |
| 157 | + stderr_bytes=err_bytes) |
| 158 | + return findings, result |
| 159 | + finally: |
| 160 | + await manager.cleanup(exec_id) |
0 commit comments