gitea_admin/pragent
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
f9eef969e6352ed60d68c3db0d278e4ca3833501
pragent/pilot/review/opencode.py
T

724 lines
28 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""pragent pilot — opencode review engine (the "brain" host).
When `PRAGENT_ENGINE=opencode` (the default), `ai_review.review_pr` delegates the
analysis to this module instead of making one direct model call. It:
1. fetches the target repo's archive at the PR head sha into a temp workdir
(so the reviewer has the real files, not just the diff text);
2. sanitizes that workdir — the checkout is PR-author-controlled, so every
file an agent runtime would auto-load as *instructions* (AGENTS.md at any
depth, CLAUDE.md, .cursorrules, a repo-supplied opencode.json…) is deleted
before opencode ever starts;
3. writes a `.pragent/brief.md` (title, description, diff, repo config, prior
reviews, sha, anchor hint) for the `pragent` agent to read, with the
author-controlled parts fenced in explicit untrusted-data markers;
4. drops pragent's `opencode.json` + `.opencode/` factory into the workdir;
5. runs `opencode run --pure --agent pragent --dir <workdir> --model <model>`
headlessly with an **allow-listed** environment (no bot token, no webhook
secret) and returns the agent's stdout (the summary + findings JSON).
Threat model: the agent's `bash` permission is `"*": "allow"` over hostile
files. So the containment is (a) no credentials in its environment, (b) no
author-controlled instruction files on disk, (c) untrusted-data framing in the
brief, (d) the Python shell — not the agent — does all Gitea I/O. See
"Threat model" in pilot/README-webhook.md.
The caller (`ai_review.review_pr`) parses that stdout into `(summary, findings)`,
validates the findings against diff anchors, and posts the review to Gitea — so
this module does NO Gitea I/O and NO parsing. It is pure review-engine glue.
Stdlib only. Fail-open: `run()` raises on failure; `review_pr` catches and posts
a short failure note.
Env:
PRAGENT_FACTORY_DIR repo root holding opencode.json + .opencode/ (default:
this file's parent's parent — the pragent repo root).
PRAGENT_OPENCODE_BIN path to the opencode CLI (default: shutil.which / the
known linuxbrew path).
PRAGENT_RTK_DIR dir holding the `rtk` binary, prepended to PATH for the
agent's bash tool (default: unset).
PRAGENT_WORK_ROOT parent for temp workdirs (default: /tmp/pragent-work).
PRAGENT_KEEP_WORK if set, leave the workdir on disk for debugging.
PRAGENT_REVIEW_TIMEOUT seconds to allow opencode to run (default: 480).
"""
import io
import json
import os
import re
import shutil
import subprocess
import tarfile
import tempfile
import time
import urllib.error
import urllib.request
from ai_review import _SEVERITY_EMOJI, is_test_path
from .opencode_workspace import (
BRIEF_PATH, changed_files, drop_factory, fetch_archive,
_extract_tar_strip_one, install_config,
sanitize_workdir, write_brief,
)
from .opencode_lens_config import (
ReviewerSpec, default_reviewers, parse_reviewers_config,
parse_triage_config, resolve_reviewers,
)
from .opencode_synthesis import (
_normalize_lens_finding, _synthesize_summary_fields, posthash,
synthesize,
)
from .opencode_lenses import (
filter_by_skip_if, intersect_with_triage, merge_usage, run_lenses,
)
from . import opencode_runtime as _runtime
from .budget import Budget, BudgetState
_filter_by_skip_if = filter_by_skip_if
_intersect_with_triage = intersect_with_triage
# Where the factory lives (opencode.json + .opencode/). Default: the pragent
# repo root (this file is at <root>/pilot/opencode_review.py).
_DEFAULT_FACTORY = os.path.dirname(
os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
)
RTK_DIR = os.environ.get("PRAGENT_RTK_DIR", "")
WORK_ROOT = os.environ.get("PRAGENT_WORK_ROOT", "/tmp/pragent-work")
TIMEOUT = int(os.environ.get("PRAGENT_REVIEW_TIMEOUT", "480"))
def _factory_dir() -> str:
return os.environ.get("PRAGENT_FACTORY_DIR", _DEFAULT_FACTORY)
def _opencode_bin() -> str:
b = os.environ.get("PRAGENT_OPENCODE_BIN")
if b and os.path.isfile(b):
return b
found = shutil.which("opencode")
if found:
return found
# last resort: the known linuxbrew path on the dev host.
return "/home/linuxbrew/.linuxbrew/bin/opencode"
# ---------------------------------------------------------------------------
# opencode invocation
# ---------------------------------------------------------------------------
def _new_usage() -> dict:
return {
"input": 0, "output": 0, "reasoning": 0,
"cache_read": 0, "cache_write": 0, "total": 0,
"cost": 0.0, "steps": 0, "tool_calls": 0, "iterations": [],
}
def parse_opencode_events(stdout: str) -> tuple[str, dict | None]:
"""Parse `opencode run --format json` NDJSON stdout into (text, usage).
- assistant text: concatenation of every `{"type":"text","part":{"text":…}}`
event, in order → the agent's full message (prose + the findings ```json
block). This is what `ai_review.parse_review_output` then extracts the
findings JSON from.
- usage: summed across every `{"type":"step_finish","part":{"tokens":…,
"cost":…}}` event (one per model turn). Returns a dict with input/output/
reasoning/cache_read/cache_write/total/cost/steps, or None if no
step_finish was seen (e.g. empty/failed run).
Tolerant: non-JSON lines, missing fields, or non-dict events are skipped
(warm-up / log noise / tool events we don't care about). Never raises.
"""
text_parts: list[str] = []
usage = _new_usage()
saw_step = False
for line in (stdout or "").splitlines():
line = line.strip()
if not line or not line.startswith("{"):
continue
try:
ev = json.loads(line)
except json.JSONDecodeError:
continue
if not isinstance(ev, dict):
continue
etype = ev.get("type")
part = ev.get("part") or {}
if etype in ("tool_use", "tool_result", "tool_call"):
usage["tool_calls"] += 1
if etype == "text" and isinstance(part, dict):
t = part.get("text")
if isinstance(t, str):
text_parts.append(t)
elif etype == "step_finish" and isinstance(part, dict):
tok = part.get("tokens") or {}
if isinstance(tok, dict):
saw_step = True
usage["steps"] += 1
usage["input"] += int(tok.get("input") or 0)
usage["output"] += int(tok.get("output") or 0)
usage["reasoning"] += int(tok.get("reasoning") or 0)
cache = tok.get("cache") or {}
if isinstance(cache, dict):
usage["cache_read"] += int(cache.get("read") or 0)
usage["cache_write"] += int(cache.get("write") or 0)
usage["total"] += int(tok.get("total") or 0)
cost = part.get("cost")
if isinstance(cost, (int, float)):
usage["cost"] += float(cost)
usage["iterations"].append({
"step": usage["steps"],
"input": int(tok.get("input") or 0),
"output": int(tok.get("output") or 0),
"reasoning": int(tok.get("reasoning") or 0),
"cache_read": int((cache or {}).get("read") or 0)
if isinstance(cache, dict) else 0,
"cache_write": int((cache or {}).get("write") or 0)
if isinstance(cache, dict) else 0,
"total": int(tok.get("total") or 0),
"cost": float(cost) if isinstance(cost, (int, float)) else 0.0,
})
return "".join(text_parts), (usage if saw_step else None)
_PROMPT = (
"Read .pragent/brief.md and review this pull request as pragent. "
"Load the review-methodology and findings-schema skills, inspect the "
"changed files and surrounding code in this repo, run any available "
"linters/typecheck on the changed files via bash, and delegate to the "
"security/tests/perf subagents only if the diff is large or "
"security-sensitive. End your message with a short prose summary followed "
"by the findings JSON code block per the findings-schema skill."
)
# ---------------------------------------------------------------------------
# Multi-lens orchestration (config-driven fan-out + synthesis)
# ---------------------------------------------------------------------------
#
# When `.pr-review.json:reviewers[]` is configured (or PRAGENT_REVIEWERS=1), the
# `run()` entry point forks N parallel opencode subprocesses — one per lens
# (security, docs, code-quality, tests, perf by default). Each runs in a
# shared workdir, reads the same brief, and emits its own findings JSON.
# `synthesize()` then merges + dedups by posthash (the same key the feedback
# loop uses, so FP-vote data lines up automatically). Absent/empty reviewers[]
# falls back to the legacy single-primary path (no behavior change).
#
# Env:
# PRAGENT_MAX_PARALLEL_LENSES per-review lens fan-out cap (default 4).
# The webhook's _review_slots still bounds
# total concurrent reviews; this bounds the
# subprocess fan-out inside one review.
# PRAGENT_LENS_TIMEOUT seconds per lens subprocess (default 540).
# PRAGENT_REVIEWERS set to "1" to force the fan-out path even
# when the repo's config is absent.
import concurrent.futures as _cf
import dataclasses as _dc
MAX_PARALLEL_LENSES = int(os.environ.get("PRAGENT_MAX_PARALLEL_LENSES", "4"))
LENS_TIMEOUT_S = int(os.environ.get("PRAGENT_LENS_TIMEOUT", "540"))
# Length caps per finding field. Cheap insurance against DoorDash's "noise on
# clean code" failure mode — one lens writing 200 words + another writing 10
# bullets = inconsistent review, regardless of synthesis.
FINDING_TITLE_MAX = 120
FINDING_BODY_MAX = 600
FINDING_SUGGESTION_MAX = 280
PER_FILE_CAP = 2
PER_PR_CAP = 7
# Tone-strip regex — drops the mushy AI-tone openers that turn a finding into
# a hedge. Applied to the title AND body before length capping. DoorDash's
# same problem (different lenses wrote different prose styles); deterministic
# regex is the cheapest fix.
_TONE_STRIP_RE = re.compile(
r"^(consider|it might be worth|perhaps|maybe|i think|i would suggest|"
r"you may want to|you could|it would be better to|it's worth|"
r"one option is|one approach is|note that|be aware that|"
r"as a general rule|as a best practice)\s*[:\-—,]?\s*",
re.I,
)
# Lens id rules. Lowercase kebab-case, ≤ 32 chars. Must match `[a-z0-9-]+`.
_LENS_ID_RE = re.compile(r"^[a-z0-9-]{1,32}$")
SEVERITY_ORDER = ("low", "medium", "high", "critical")
SEVERITY_RANK = {s: i for i, s in enumerate(SEVERITY_ORDER)}
# ---------------------------------------------------------------------------
# Per-lens subprocess + parallel fan-out
# ---------------------------------------------------------------------------
def _extract_json_object(text: str) -> dict | None:
"""Last balanced {...} JSON object in text, or None. Tolerant: scans for
a ```json fence first, then falls back to a balanced-brace scan of the
whole text. Reused by `_run_one_lens` to parse a lens's output."""
if not text:
return None
# 1. Try the last ```json ... ``` fence.
fences = list(re.finditer(r"```(?:json)?\s*\n", text))
for m in reversed(fences):
start = m.end()
# find the matching ```
end = text.find("```", start)
if end == -1:
continue
block = text[start:end].strip()
try:
obj = json.loads(block)
except json.JSONDecodeError:
# balanced-brace scan inside the block
for cand in _balanced_jsons(block):
try:
return json.loads(cand)
except json.JSONDecodeError:
continue
continue
if isinstance(obj, dict):
return obj
if isinstance(obj, list) and obj and isinstance(obj[0], dict):
return {"findings": obj}
# 2. Balanced scan over the whole text.
for cand in reversed(list(_balanced_jsons(text))):
try:
obj = json.loads(cand)
except json.JSONDecodeError:
continue
if isinstance(obj, dict):
return obj
if isinstance(obj, list) and obj and isinstance(obj[0], dict):
return {"findings": obj}
return None
def _balanced_jsons(text: str):
"""Yield each top-level balanced {...} substring (greedy on the inside)."""
depth = 0
start = None
for i, ch in enumerate(text):
if ch == "{":
if depth == 0:
start = i
depth += 1
elif ch == "}":
if depth > 0:
depth -= 1
if depth == 0 and start is not None:
yield text[start:i + 1]
start = None
def _run_process(
cmd, *, cwd, env, timeout, parse_events, budget=None, budget_state=None,
model="",
):
return _runtime._run_process(
cmd, cwd=cwd, env=env, timeout=timeout, parse_events=parse_events,
budget=budget, budget_state=budget_state, model=model,
runner=subprocess.run,
)
def triage(
workdir: str,
triage_cfg: dict,
reviewers: list[ReviewerSpec],
default_model: str,
factory_root: str,
budget: Budget | None = None,
budget_state: BudgetState | None = None,
) -> list[str] | None:
"""Run the triage agent. Returns the lens subset with surface.
Three outcomes, kept distinct on purpose:
* ``[lens, …]`` — run exactly these.
* ``[]`` — the agent deliberately returned an empty list: no lens
has surface on this diff, so the fan-out is skipped entirely. Only a
literally-empty ``lenses`` list produces this.
* ``None`` — fail open, run everything. Covers triage disabled, a
crash, unparseable output, a malformed `lenses` value, AND the case
where the agent named only ids that don't exist (a hallucinated roster
is not a verdict of "nothing to review").
`triage_cfg.enabled = False` → skip triage, return None.
"""
if not triage_cfg.get("enabled", True):
return None
bin_ = _opencode_bin()
home = _shared_home()
_warm_opencode(home, default_model)
env = _build_env(home)
lens_ids = [r.id for r in reviewers]
prompt = (
f"You are the triage agent. Read .pragent/brief.md. "
f"Available lens ids: {','.join(lens_ids)}. "
f"Return STRICT JSON on a single line: {{\"lenses\":[\"<id>\",...]}}. "
f"Include a lens only if the diff gives it real surface. "
f"Empty list = no lenses needed. No prose."
)
cmd = [
bin_, "run", "--pure", "--format", "json",
"--agent", "triage", "--dir", workdir, "--model", default_model,
prompt,
]
try:
proc = _run_process(
cmd, cwd=workdir, env=env, timeout=min(120, budget.max_duration_seconds)
if budget else 120, parse_events=parse_opencode_events,
budget=budget, budget_state=budget_state, model=default_model,
)
except (subprocess.TimeoutExpired, Exception) as e:
print(f"pragent: triage crashed: {e}; falling back to all lenses", flush=True)
return None
text, _ = parse_opencode_events(proc.stdout or "")
obj = _extract_json_object(text) if text.strip() else None
if obj is None:
print("pragent: triage no parseable output; falling back to all lenses", flush=True)
return None
lenses = obj.get("lenses")
if not isinstance(lenses, list):
return None
if not lenses:
# Deliberate "no lens needed" verdict — the one case that skips.
print("pragent: triage selected no lenses (no review surface)", flush=True)
return []
valid = [lid for lid in lenses if isinstance(lid, str) and lid in lens_ids]
if not valid:
# The agent named lenses, but none of them exist. That's a bad roster,
# not an empty one — fail open rather than silently skipping the review.
print(
f"pragent: triage named no known lenses ({lenses!r}); "
f"falling back to all lenses",
flush=True,
)
return None
cap = triage_cfg.get("max_lenses", 5)
selected = valid[:cap]
print(f"pragent: triage selected {selected}", flush=True)
return selected
# ---------------------------------------------------------------------------
# Multi-lens entry point
# ---------------------------------------------------------------------------
def run_lenses_review(
*,
api: str,
repo: str,
index: str,
sha: str,
token: str,
title: str,
body: str,
diff: str,
config: dict | None,
prior_reviews: list[str] | None,
model: str,
compression_note: str = "",
additional_context: str = "",
budget: Budget | None = None,
budget_state: BudgetState | None = None,
) -> tuple[str, dict | None]:
"""Fan-out + synthesize path. Returns (merged-text, merged-usage).
`text` is a synthesized prose summary + the merged findings JSON (the
downstream `ai_review.parse_review_output` expects the same shape it
always has: prose + a final ```json fence with the legacy schema).
"""
os.makedirs(WORK_ROOT, exist_ok=True)
budget = budget or Budget.from_config(config)
budget_state = budget_state or BudgetState(budget)
workdir = tempfile.mkdtemp(prefix=f"{repo.replace('/', '_')}-{sha[:8]}-", dir=WORK_ROOT)
keep = bool(os.environ.get("PRAGENT_KEEP_WORK"))
t0 = time.monotonic()
try:
fetch_archive(api, repo, sha, token, workdir)
sanitize_workdir(workdir)
write_brief(
workdir,
repo=repo, index=index, sha=sha, title=title, description=body,
diff=diff, config=config, prior_reviews=prior_reviews,
compression_note=compression_note,
additional_context=additional_context,
)
drop_factory(workdir)
reviewers = resolve_reviewers(config)
if not reviewers:
# Edge case: reviewers[] present but every entry had activation:off.
# Fall back to single-primary.
return _fallback_single_primary(
workdir=workdir, model=model, budget=budget,
budget_state=budget_state,
)
triage_cfg = parse_triage_config((config or {}).get("triage"))
changed_paths = changed_files(diff)
reviewers = _filter_by_skip_if(reviewers, changed_paths)[:budget.max_lenses]
selected = triage(
workdir, triage_cfg, reviewers, model, _factory_dir(),
budget=budget, budget_state=budget_state,
)
if selected is not None:
if not selected:
# Triage says nothing here has review surface. Skip the
# fan-out and post a clean empty review — running all N
# lenses anyway would burn N subprocesses to contradict it.
return _no_surface_response(repo, index, sha, len(reviewers))
reviewers = _intersect_with_triage(reviewers, selected)
if not reviewers:
# Every lens was filtered out (skip_if_all_changed_paths, or a
# triage subset naming lenses this repo doesn't enable). Same
# outcome as the triage skip: nothing to run, nothing to say.
return _no_surface_response(repo, index, sha, 0)
factory_root = _factory_dir()
results = run_lenses(
workdir, reviewers, model, factory_root, budget, budget_state,
)
# Merge findings + usage across lenses
findings_per_lens = {lid: r[0] for lid, r in results.items()}
merged = synthesize(findings_per_lens, reviewers)
merged_usage = merge_usage([r[1] for r in results.values()])
merged_usage.update({f"budget_{k}": v for k, v in budget_state.snapshot().items()})
# Build a synthetic text response that ai_review.parse_review_output
# can consume (prose summary + final ```json fence with legacy schema).
lens_names = ", ".join(sorted({f["_lens"] for f in merged})) or "—"
sev_counts = {"critical": 0, "high": 0, "medium": 0, "low": 0}
for f in merged:
sev_counts[f["severity"]] = sev_counts.get(f["severity"], 0) + 1
summary = (
f"Multi-lens review of {repo}#{index} "
f"(sha {sha[:8]}). Lenses: {lens_names}. "
f"Findings: critical={sev_counts['critical']} "
f"high={sev_counts['high']} medium={sev_counts['medium']} "
f"low={sev_counts['low']}."
)
# Strip internal _lens/_posthash/_ruleId/_multi_lens/_lens_model keys from
# the merged findings so the legacy parser doesn't see them. (They
# remain in the DB via feedback_harvest which re-derives posthash.)
clean_findings = [
{k: v for k, v in f.items() if not k.startswith("_")}
for f in merged
]
# Synthesize the review-level meta (walkthrough / risk_verdict /
# test_coverage) from the merged findings + diff. Real implementation
# arrives in Task 8; the stub keeps the synthesized JSON shape stable
# so ai_review.parse_review_output can extract the three new fields
# (it defaults them to [] / "" when missing — backward compatible).
walkthrough, risk_verdict, test_coverage = _synthesize_summary_fields(
merged, diff, changed_paths=changed_paths,
)
synthesized_payload = {
"summary": summary,
"summary_changes": [],
"risks": [],
"walkthrough": walkthrough,
"risk_verdict": risk_verdict,
"test_coverage": test_coverage,
"findings": clean_findings,
}
text = (
f"{summary}\n\n"
f"## Findings (multi-lens)\n\n"
f"```json\n{json.dumps(synthesized_payload, indent=2)}\n```\n"
)
if merged_usage is not None:
merged_usage["duration_s"] = round(time.monotonic() - t0, 1)
merged_usage["lenses"] = sorted(results.keys())
merged_usage["lens_steps"] = merged_usage.get("steps", 0)
return text, merged_usage
finally:
if not keep:
shutil.rmtree(workdir, ignore_errors=True)
def _no_surface_response(
repo: str, index: str, sha: str, n_lenses: int
) -> tuple[str, dict | None]:
"""A well-formed 'nothing to review' result for the no-lens paths.
Returns the same shape every other path returns — prose plus a final
```json fence with an empty `findings` array — so
`ai_review.parse_review_output` parses it normally. Returning bare `""`
here (the old behaviour) landed in ai_review's unparseable-output branch
and posted "AI review produced no parseable output", which reads as a
malfunction rather than a verdict.
"""
if n_lenses:
summary = (
f"Triage found no review surface in {repo}#{index} "
f"(sha {sha[:8]}): none of the {n_lenses} configured lens(es) "
f"apply to this diff. No findings."
)
else:
summary = (
f"No lens applies to {repo}#{index} (sha {sha[:8]}) after path "
f"filtering. No findings."
)
text = (
f"{summary}\n\n"
f"## Findings (multi-lens)\n\n"
f"```json\n{json.dumps({'summary': summary, 'findings': []}, indent=2)}\n```\n"
)
return text, None
def _fallback_single_primary(
workdir: str, model: str, budget: Budget | None = None,
budget_state: BudgetState | None = None,
) -> tuple[str, dict | None]:
"""Used when reviewers[] resolves to empty (all activation:off)."""
try:
text, usage = run_opencode(
workdir, model, budget=budget, budget_state=budget_state,
)
return text, usage
except Exception as e:
print(f"pragent: fallback single-primary failed: {e}", flush=True)
return "", None
def _shared_home() -> str:
return _runtime.shared_home(WORK_ROOT)
def _ensure_global_config(home: str) -> None:
_runtime.ensure_global_config(home, _factory_dir(), install_config)
def _build_env(home: str) -> dict:
return _runtime.build_env(home, RTK_DIR)
def _warm_opencode(home: str, model: str) -> None:
_runtime.warm_opencode(
home, model, opencode_bin=_opencode_bin(),
ensure_config=_ensure_global_config,
build_environment=_build_env,
runner=subprocess.run,
)
def run_opencode(
workdir: str, model: str, timeout: int | None = None,
budget: Budget | None = None, budget_state: BudgetState | None = None,
) -> tuple[str, dict | None]:
return _runtime.run_opencode(
workdir, model, opencode_bin=_opencode_bin(),
shared_home_fn=_shared_home, warm_fn=_warm_opencode,
build_environment=_build_env, parse_events=parse_opencode_events,
prompt=_PROMPT, timeout=timeout or TIMEOUT, budget=budget,
budget_state=budget_state, runner=subprocess.run,
)
# ---------------------------------------------------------------------------
# Orchestrator entry point
# ---------------------------------------------------------------------------
def run(
*,
api: str,
repo: str,
index: str,
sha: str,
token: str,
title: str,
body: str,
diff: str,
config: dict | None,
prior_reviews: list[str] | None,
model: str,
compression_note: str = "",
additional_context: str = "",
) -> tuple[str, dict | None]:
"""End-to-end: checkout archive → brief → drop factory → opencode → (text, usage).
Returns the reconstructed opencode assistant text (summary + findings JSON)
and a usage dict (token/cost totals + `duration_s`), or `(text, None)` when
no usage events were seen. Raises on any failure; the caller (`review_pr`)
fails open. The workdir is removed unless PRAGENT_KEEP_WORK is set.
`compression_note`: a small markdown block to append to the brief's PR
description (e.g. "diff compressed: 25k → 12k chars"). Empty string by
default. Appended AFTER the untrusted-data fence so the agent reads it as
guidance, not author input.
`additional_context`: pre-fetched markdown from
`additional_context_urls` / `PRAGENT_ADDITIONAL_CONTEXT_URL`. Rendered as
its own brief section. Empty string by default.
Routing:
* If `config:reviewers[]` is present OR `PRAGENT_REVIEWERS=1` env is set,
delegate to `run_lenses_review` (parallel fan-out + synth).
* Otherwise, the legacy single-primary path (calls `run_opencode`).
The no-config branch is the no-regression gate.
"""
use_fanout = bool((config or {}).get("reviewers")) or bool(
os.environ.get("PRAGENT_REVIEWERS")
)
if use_fanout:
return run_lenses_review(
api=api, repo=repo, index=index, sha=sha, token=token,
title=title, body=body, diff=diff, config=config,
prior_reviews=prior_reviews, model=model,
compression_note=compression_note,
additional_context=additional_context,
)
budget = Budget.from_config(config)
budget_state = BudgetState(budget)
os.makedirs(WORK_ROOT, exist_ok=True)
workdir = tempfile.mkdtemp(prefix=f"{repo.replace('/', '_')}-{sha[:8]}-", dir=WORK_ROOT)
keep = bool(os.environ.get("PRAGENT_KEEP_WORK"))
t0 = time.monotonic()
try:
fetch_archive(api, repo, sha, token, workdir)
removed = sanitize_workdir(workdir)
if removed:
print(
f"pragent: stripped {len(removed)} author-controlled instruction "
f"file(s) from {repo}#{index}: {', '.join(removed[:10])}",
flush=True,
)
write_brief(
workdir,
repo=repo, index=index, sha=sha, title=title, description=body,
diff=diff, config=config, prior_reviews=prior_reviews,
compression_note=compression_note,
additional_context=additional_context,
)
drop_factory(workdir)
text, usage = run_opencode(
workdir, model, budget=budget, budget_state=budget_state,
)
if not text.strip():
raise RuntimeError("opencode produced no output")
if usage is not None:
usage["duration_s"] = round(time.monotonic() - t0, 1)
return text, usage
finally:
if not keep:
shutil.rmtree(workdir, ignore_errors=True)
Reference in New Issue View Git Blame Copy Permalink