"""Host tools exposed to the agent through `omp_rpc.host_tool`. The agent uses these for any side effect that touches GitHub, the reproduction transcript store, or the orchestrator's bookkeeping. """ from __future__ import annotations import asyncio import json import logging import os import re import subprocess import time from collections.abc import Callable, Mapping from dataclasses import dataclass from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any, NoReturn from omp_rpc import HostTool, HostToolContext, RpcCommandError, host_tool from robomp import persona from robomp.config import Settings from robomp.db import Database, IssueState, issue_key from robomp.git_ops import GitCommandError, HeadDriftError from robomp.github_backend import GitHubBackend from robomp.github_client import GitHubError, IssueInfo, PullRequestFileInfo, RepoInfo from robomp.sandbox import ( GitTransport, Workspace, _prepare_slot_runtime_env, _safe_directory_env, _share_git_metadata_with_slots, _slot_permissions_active, _slot_subprocess_kwargs, rename_workspace_branch, validate_branch_slug, workspace_key, ) log = logging.getLogger(__name__) _PRE_PR_FIX_COMMAND = ("bun", "run", "fix") _PRE_PR_CHECK_COMMAND = ("bun", "check") _BUN_INSTALL_COMMAND = ("bun", "install", "--frozen-lockfile", "--ignore-scripts") _BUN_INSTALL_TIMEOUT_SECONDS = 300.0 _REPO_COMMAND_SCRUBBED_ENV_KEYS: tuple[str, ...] = ( "GITHUB_TOKEN", "GITHUB_WEBHOOK_SECRET", "ROBOMP_REPLAY_TOKEN", "ROBOMP_GH_PROXY_HMAC_KEY", ) _NEEDS_INFO_LABEL = "needs-info" _AGENT_HOME = Path("/srv/agent-home") _PRE_PR_FIX_TIMEOUT_SECONDS = 600.0 _PRE_PR_CHECK_TIMEOUT_SECONDS = 600.0 _PRE_PR_CHECK_MAX_OUTPUT = 12_000 @dataclass(slots=True) class AbortController: """Mutable handoff between the `abort_task` host tool and the worker. `signal()` is called from the host-tool thread to request an irrecoverable teardown of the omp subprocess. The worker pre-populates `stop` with a thread-safe terminator (the same one used for queue cancellation and the hard-timeout watchdog), and inspects `triggered` after `prompt_and_wait` unblocks to decide whether the resulting `RpcError` is an intentional abort (swallow, mark event `done`) vs an actual failure (propagate). """ triggered: bool = False reason: str = "" stop: Callable[[], None] | None = None def signal(self, reason: str) -> None: # Idempotent. Only the first call records its reason; later calls are # silent no-ops so a retry inside the tool can't overwrite the # original diagnosis with a generic follow-up message. if self.triggered: return self.triggered = True self.reason = reason if self.stop is not None: self.stop() @dataclass(slots=True, frozen=True) class ToolBindings: """Per-task closure that the host tools capture.""" db: Database github: GitHubBackend git_transport: GitTransport repo: RepoInfo issue: IssueInfo workspace: Workspace loop: asyncio.AbstractEventLoop author_name: str author_email: str settings: Settings | None = None # Number of the GitHub thread the inbound webhook arrived on. For an # issue comment this is the issue; for a PR conversation or review # comment it's the PR. `gh_post_comment` defaults its target here so # the agent's reply lands on the thread the human is actually reading. # `None` for tasks with no inbound thread (e.g. initial triage), in # which case the originating issue is used. inbound_thread_number: int | None = None # True iff the inbound thread is a pull request. Triage tools # (`classify_issue`, `set_issue_labels`) are not exposed on PR threads # — the originating issue has already been classified and the PR # itself does not carry triage labels. inbound_is_pr: bool = False # True only for incoming-PR review tasks. Review tools require it; mutating # branch/PR publication tools reject when it is set. review_mode: bool = False # Current task is driven by an allowlist/OWNER maintainer directive that # authorizes implementation. Gates first-PR creation on non-bug/doc issues. impl_authorized: bool = False slot_uid: int | None = None # Set by the worker before launching omp. Carries the abort-task signal # back out to the worker; `None` for unit tests that exercise tools # without a live RpcClient. abort: AbortController | None = None @property def issue_key(self) -> str: return issue_key(self.issue.repo, self.issue.number) @property def default_comment_number(self) -> int: return self.inbound_thread_number if self.inbound_thread_number is not None else self.issue.number def _run_coro(loop: asyncio.AbstractEventLoop, coro: Any) -> Any: """Block the agent thread until an async call completes on the worker loop.""" future = asyncio.run_coroutine_threadsafe(coro, loop) return future.result() def _issue_needs_info(bindings: ToolBindings) -> bool: row = bindings.db.get_issue(bindings.issue_key) return row is not None and row.state == "needs_info" def _optional_label_error(exc: Exception) -> str: return f"{type(exc).__name__}: {exc}" def _remove_needs_info_label(bindings: ToolBindings) -> bool: try: _run_coro( bindings.loop, bindings.github.remove_issue_label(bindings.repo.full_name, bindings.issue.number, _NEEDS_INFO_LABEL), ) except GitHubError as exc: if exc.status == 404: return True log.warning("needs-info label cleanup failed", extra={"issue": bindings.issue_key, "err": str(exc)}) return False except Exception as exc: # noqa: BLE001 - best-effort optional label cleanup log.warning( "needs-info label cleanup failed", extra={"issue": bindings.issue_key, "err": _optional_label_error(exc)}, ) return False return True def _advance_needs_info(bindings: ToolBindings, state: IssueState) -> bool: if not _issue_needs_info(bindings): return False label_cleared = _remove_needs_info_label(bindings) bindings.db.set_issue_state(bindings.issue_key, state) return label_cleared def _audit( bindings: ToolBindings, name: str, args: Mapping[str, Any], result: Any | None = None, error: str | None = None ) -> None: bindings.db.log_tool_call( issue_key=bindings.issue_key, tool=name, args=args, result=result if isinstance(result, Mapping) else ({"value": result} if result is not None else None), error=error, ) def _raise_command(message: str) -> NoReturn: raise RpcCommandError(message, error={"message": message}) def _git_identity_env(author_name: str, author_email: str) -> dict[str, str]: """Environment forcing agent git commits to use the configured bot identity.""" return { "GIT_AUTHOR_NAME": author_name, "GIT_AUTHOR_EMAIL": author_email, "GIT_COMMITTER_NAME": author_name, "GIT_COMMITTER_EMAIL": author_email, } def _repo_command_env(bindings: ToolBindings) -> dict[str, str]: """Environment for repo-owned commands (`bun`, formatter, local git). These commands execute code from the checked-out repository, so they must not inherit GitHub credentials from the orchestrator. They also need the exact same HOME/XDG/TMP/Bun cache paths as the agent process; otherwise host-side pre-publish gates validate a different machine than the agent saw. """ env = os.environ.copy() for key in _REPO_COMMAND_SCRUBBED_ENV_KEYS: env[key] = "" if _AGENT_HOME.is_dir(): env["HOME"] = str(_AGENT_HOME) env.update(_prepare_slot_runtime_env(bindings.workspace, bindings.slot_uid)) env.update(_safe_directory_env(bindings.workspace.repo_dir)) env.update(_git_identity_env(bindings.author_name, bindings.author_email)) env["GIT_TERMINAL_PROMPT"] = "0" return env def _run_repo_command( bindings: ToolBindings, cmd: list[str] | tuple[str, ...], *, timeout: float | None = None, extra_env: Mapping[str, str] | None = None, ) -> subprocess.CompletedProcess[str]: """Run a repo-local command with agent-equivalent permissions and env.""" env = _repo_command_env(bindings) if extra_env: env.update(extra_env) return subprocess.run( list(cmd), cwd=str(bindings.workspace.repo_dir), check=False, capture_output=True, text=True, timeout=timeout, env=env, **_slot_subprocess_kwargs(bindings.slot_uid), ) def _has_bun_script(repo_dir: Path, name: str) -> bool: """Return True iff `package.json` defines a `scripts.` entry. A malformed or unreadable `package.json` is treated as "present" so the repository-native error surfaces from `bun` instead of being silently swallowed here. """ package_json = repo_dir / "package.json" if not package_json.is_file(): return False try: package = json.loads(package_json.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return True if not isinstance(package, Mapping): return True scripts = package.get("scripts") return isinstance(scripts, Mapping) and isinstance(scripts.get(name), str) def _format_process_output(stdout: Any, stderr: Any) -> str: parts: list[str] = [] for stream in (stdout, stderr): if isinstance(stream, bytes): text = stream.decode(errors="replace") elif isinstance(stream, str): text = stream elif stream is None: continue else: text = str(stream) text = text.strip() if text: parts.append(text) output = "\n".join(parts) if not output: return "(no output)" if len(output) <= _PRE_PR_CHECK_MAX_OUTPUT: return output return ( f"... output truncated to last {_PRE_PR_CHECK_MAX_OUTPUT} characters ...\n{output[-_PRE_PR_CHECK_MAX_OUTPUT:]}" ) def ensure_workspace_dependencies(bindings: ToolBindings) -> None: """Bootstrap ``node_modules`` so the agent can resolve workspace packages. A per-issue worktree is a bare source checkout (``git worktree add`` off the shared clone pool): it has the repo's ``package.json``/``bun.lock`` but no ``node_modules``. With bun's ``hoisted`` linker the workspace links (``@oh-my-pi/pi-*``) only exist after an install, so without one any ``bun test``/``bun check`` the agent runs fails instantly with "Cannot find package" — the agent then reports it could not verify. We install before the agent starts, mirroring how the natives cache pre-populates ``.node`` artifacts. The links resolve into *this* worktree's ``packages/*`` (not the orchestrator's read-only ``/work/pi``), so tests exercise the PR's edited source. ``--frozen-lockfile`` keeps the lockfile pristine (no spurious diff for the agent to commit) and ``--ignore-scripts`` skips lifecycle scripts so an untrusted PR's ``postinstall``/``prepare`` cannot execute as the slot and the cached native build is not redone. Runs with the same scrubbed, slot-owned env as the other repo-owned bun commands (``bun run fix`` / ``bun check``). Skips non-bun repos. Otherwise runs unconditionally on every launch (including ``--continue`` resumes): a frozen install verifies an intact tree in ~20ms and re-links anything missing, so a previous install that timed out or crashed half-way self-heals instead of being skipped forever on a mere ``node_modules/`` directory existing. Best-effort: any failure (offline, or a PR that bumped deps so the frozen lockfile is stale) is logged and swallowed — the agent can still install itself or report the gap. """ repo_dir = bindings.workspace.repo_dir if not (repo_dir / "package.json").is_file() or not (repo_dir / "bun.lock").is_file(): return try: proc = _run_repo_command(bindings, _BUN_INSTALL_COMMAND, timeout=_BUN_INSTALL_TIMEOUT_SECONDS) except FileNotFoundError: log.warning("bun_install bootstrap skipped: bun not on PATH", extra={"issue": bindings.issue_key}) return except (OSError, subprocess.SubprocessError) as exc: log.warning("bun_install bootstrap failed", extra={"issue": bindings.issue_key, "err": str(exc)}) return if proc.returncode != 0: log.warning( "bun_install bootstrap nonzero exit", extra={ "issue": bindings.issue_key, "code": proc.returncode, "output": _format_process_output(proc.stdout, proc.stderr), }, ) return log.info("bun_install bootstrap ok", extra={"issue": bindings.issue_key}) def _run_pre_publish_bun_fix( bindings: ToolBindings, args: Mapping[str, Any], *, tool_name: str, stage: str, skip_checks: bool = False, ) -> None: """Run `bun run fix` then amend any working-tree diff into HEAD. Silently no-ops when the repository does not define a `scripts.fix` entry. Anything the formatter touches gets amended into the agent's HEAD commit so the downstream cleanliness gate sees a pristine worktree without littering PR history with standalone `style:` commits. Amending an already-pushed HEAD is safe: the push transport uses `--force-with-lease`, which exists precisely to recover from local history rewrites. When there is no commit that may safely absorb the diff — HEAD sits on `origin/` (pre-existing formatter drift) or is foreign-authored — the tool refuses with instructions instead of guessing. `tool_name` is the host tool calling this (audit attribution). `stage` is the human-readable verb used in error wording — "open PR" when called from `gh_open_pr`, "push" when called from `gh_push_branch`. `skip_checks` is the agent-supplied escape hatch: when True, the formatter is NOT invoked and any post-fix commit is skipped, so a broken-formatter situation on `main` (unrelated to the agent's diff) doesn't strand the push forever. The dirty-tree gate still runs — we never let uncommitted changes leak into a remote ref. """ if not _has_bun_script(bindings.workspace.repo_dir, "fix"): return # Dirty-tree gate BEFORE the formatter so any pre-existing uncommitted # edit isn't silently swept into the formatter amend by the # `git add -A` below. The agent owns the worktree end-to-end; any diff # not already in a commit is a workflow bug it must resolve before we # mutate the tree further. pre_status = _run_repo_command(bindings, ["git", "status", "--porcelain", "--untracked-files=normal"]) if pre_status.stdout.strip(): dirty = "\n ".join(pre_status.stdout.strip().splitlines()) msg = ( f"refusing to {stage}: dirty worktree before `bun run fix`.\n " f"{dirty}\n" "Commit (or `git stash`) every change before invoking the formatter — " "anything left uncommitted would be amended into your HEAD commit " "and silently land in the PR." ) _audit(bindings, tool_name, args, error=msg) _raise_command(msg) if skip_checks: _audit( bindings, tool_name, args, result={"skipped": "bun_run_fix", "reason": "skip_checks=true"}, ) return try: proc = _run_repo_command(bindings, _PRE_PR_FIX_COMMAND, timeout=_PRE_PR_FIX_TIMEOUT_SECONDS) except FileNotFoundError: msg = f"refusing to {stage}: `bun run fix` is required before {stage}, but `bun` is not on PATH." _audit(bindings, tool_name, args, error=msg) _raise_command(msg) except subprocess.TimeoutExpired as exc: output = _format_process_output(exc.stdout, exc.stderr) msg = ( f"refusing to {stage}: `bun run fix` timed out before {stage}.\n" f"{output}\n\n" f"Investigate the hang, rerun the formatter, commit any resulting changes, " f"and retry." ) _audit(bindings, tool_name, args, error=msg) _raise_command(msg) if proc.returncode != 0: output = _format_process_output(proc.stdout, proc.stderr) msg = ( f"refusing to {stage}: `bun run fix` failed before {stage} (exit {proc.returncode}).\n" f"{output}\n\n" f"Resolve the formatter failure, rerun `bun run fix` successfully, commit any " f"resulting changes, and retry." ) _audit(bindings, tool_name, args, error=msg) _raise_command(msg) status = _run_repo_command(bindings, ["git", "status", "--porcelain", "--untracked-files=normal"]) if not status.stdout.strip(): return # The formatter produced a diff. Fold it into the agent's HEAD commit — # but only when HEAD is a bot-authored commit not already on the base # branch. Amending a commit `origin/` contains would rewrite # shared history; amending a foreign-authored commit would bury our # diff in someone else's work. base = bindings.repo.default_branch ahead = _run_repo_command(bindings, ["git", "rev-list", "-n", "1", f"origin/{base}..HEAD"]) if ahead.returncode != 0 or not ahead.stdout.strip(): msg = ( f"refusing to {stage}: `bun run fix` changed files, but there is no commit of " f"yours to fold them into — the checkout matches `origin/{base}`, so the " f"formatter drift pre-exists on `{base}`. Inspect with `git status` / `git diff`; " "either commit the formatter output yourself or discard it " "(`git checkout -- . && git clean -fd`) and retry with `skip_checks=true`, " "documenting the bypass." ) _audit(bindings, tool_name, args, error=msg) _raise_command(msg) head_identity = _run_repo_command(bindings, ["git", "log", "-1", "--format=%an%x1f%ae", "HEAD"]) if head_identity.returncode != 0 or head_identity.stdout.strip("\n").split("\x1f") != [ bindings.author_name, bindings.author_email, ]: author = head_identity.stdout.strip("\n").replace("\x1f", " <") + ">" msg = ( f"refusing to {stage}: `bun run fix` changed files, but HEAD is authored by " f"{author} — refusing to fold the formatter diff into a foreign commit. " "Fix the identity first (`git commit --amend --reset-author --no-edit`) and retry." ) _audit(bindings, tool_name, args, error=msg) _raise_command(msg) add = _run_repo_command(bindings, ["git", "add", "-A"]) if add.returncode != 0: err = (add.stderr or add.stdout).strip() msg = f"refusing to {stage}: `git add -A` failed after `bun run fix`: {err}" _audit(bindings, tool_name, args, error=msg) _raise_command(msg) commit = _run_repo_command(bindings, ["git", "commit", "--amend", "--no-edit"]) if commit.returncode != 0: err = (commit.stderr or commit.stdout).strip() msg = f"refusing to {stage}: failed to amend `bun run fix` changes into HEAD: {err}" _audit(bindings, tool_name, args, error=msg) _raise_command(msg) def _run_pre_publish_bun_check( bindings: ToolBindings, args: Mapping[str, Any], *, tool_name: str, stage: str, skip_checks: bool = False, ) -> None: """Run `bun check` before publishing. When `skip_checks=True` the check is not invoked — used to escape pre-existing breakage on `main` that the agent's diff did not cause. """ if skip_checks: _audit( bindings, tool_name, args, result={"skipped": "bun_check", "reason": "skip_checks=true"}, ) return if not _has_bun_script(bindings.workspace.repo_dir, "check"): return try: proc = _run_repo_command(bindings, _PRE_PR_CHECK_COMMAND, timeout=_PRE_PR_CHECK_TIMEOUT_SECONDS) except FileNotFoundError: msg = f"refusing to {stage}: `bun check` is required before {stage}, but `bun` is not on PATH." _audit(bindings, tool_name, args, error=msg) _raise_command(msg) except subprocess.TimeoutExpired as exc: output = _format_process_output(exc.stdout, exc.stderr) msg = ( f"refusing to {stage}: `bun check` timed out before {stage}.\n" f"{output}\n\n" f"Fix the check hang/failure, rerun `bun check`, commit any resulting changes, " f"and retry." ) _audit(bindings, tool_name, args, error=msg) _raise_command(msg) if proc.returncode != 0: output = _format_process_output(proc.stdout, proc.stderr) msg = ( f"refusing to {stage}: `bun check` failed before {stage} (exit {proc.returncode}).\n" f"{output}\n\n" f"Fix the reported failures, rerun `bun check` successfully, commit any resulting changes, " f"and retry." ) _audit(bindings, tool_name, args, error=msg) _raise_command(msg) _AUTOCLOSE_INELIGIBLE_STATES: frozenset[str] = frozenset({"closed", "merged", "needs_info", "abandoned"}) def _should_schedule_autoclose(bindings: ToolBindings, target_number: int) -> float | None: """Return the configured close window (hours) when this comment should schedule the question auto-close job: feature enabled, same issue, classified as `question`, and the issue is not already in a terminal or waiting-for-reporter state. """ settings = bindings.settings if settings is None or not settings.question_autoclose_enabled: return None hours = float(settings.question_autoclose_hours) if hours <= 0: return None if target_number != bindings.issue.number: return None if bindings.inbound_is_pr: return None row = bindings.db.get_issue(bindings.issue_key) if row is None or row.classification != "question": return None if row.state in _AUTOCLOSE_INELIGIBLE_STATES: return None return hours def _schedule_autoclose(bindings: ToolBindings, *, comment_id: int, hours: float) -> str | None: """Insert (or refresh) a `pending_closures` row for the bot's answer. Failures are logged but never poisoned back to the agent — the human has already seen the comment and the orchestrator's bookkeeping shouldn't surface as a tool error. """ close_at_dt = datetime.now(UTC) + timedelta(hours=hours) close_at = close_at_dt.strftime("%Y-%m-%dT%H:%M:%S.%fZ") try: bindings.db.upsert_pending_closure( issue_key=bindings.issue_key, repo=bindings.issue.repo, number=bindings.issue.number, comment_id=comment_id, issue_author=bindings.issue.author, close_at=close_at, ) except Exception as exc: # pragma: no cover - defensive log.exception( "autoclose schedule failed", extra={"issue_key": bindings.issue_key, "comment_id": comment_id, "error": str(exc)}, ) return None return close_at # ---------- gh_post_comment ---------- def _build_post_comment(bindings: ToolBindings) -> HostTool[Any, Any]: def execute(args: dict[str, Any], _ctx: HostToolContext[Any]) -> str: body = args.get("body") if not isinstance(body, str) or not body.strip(): _raise_command("gh_post_comment requires a non-empty 'body'.") target_number = bindings.default_comment_number if isinstance(args.get("number"), int): target_number = int(args["number"]) # If this comment answers the originating question issue, append the # 👎-to-keep-open suffix so the auto-close scheduler has a reaction # surface to consult. schedule_close = _should_schedule_autoclose(bindings, target_number) body_to_post = body if schedule_close is not None: body_to_post = f"{body.rstrip()}\n\n{persona.question_autoclose_suffix(schedule_close)}" try: comment = _run_coro( bindings.loop, bindings.github.post_comment(bindings.repo.full_name, target_number, body_to_post), ) except GitHubError as exc: _audit(bindings, "gh_post_comment", args, error=str(exc)) _raise_command(f"GitHub rejected comment: {exc.status} {exc.message}") audit_result: dict[str, Any] = {"comment_id": comment.id} if schedule_close is not None: scheduled_at = _schedule_autoclose( bindings, comment_id=comment.id, hours=schedule_close, ) if scheduled_at is not None: audit_result["scheduled_close_at"] = scheduled_at _audit(bindings, "gh_post_comment", args, result=audit_result) return f"comment posted: id={comment.id}" return host_tool( name="gh_post_comment", description=persona.host_tool_description("gh_post_comment"), parameters={ "type": "object", "properties": { "body": { "type": "string", "description": persona.host_tool_parameter_description("gh_post_comment", "body"), }, "number": { "type": "integer", "description": persona.host_tool_parameter_description("gh_post_comment", "number"), }, }, "required": ["body"], "additionalProperties": False, }, execute=execute, ) def _repair_message_escapes(message: str) -> str | None: """Convert shell-literal ``\\n`` escapes in a commit message to newlines. Agents regularly run ``git commit -m 'subject\\n\\nbody'`` with single quotes, recording the two-character backslash-n sequence instead of a newline — the message then renders as one line of ``\\n``-littered text on GitHub. Escapes inside backtick code spans (`` `\\n` ``) are genuine content and are preserved. Returns the repaired message, or ``None`` when nothing needs repair. """ if "\\n" not in message: return None parts = message.split("`") changed = False for i in range(0, len(parts), 2): # even indexes sit outside code spans fixed = parts[i].replace("\\r\\n", "\n").replace("\\n", "\n") if fixed != parts[i]: parts[i] = fixed changed = True return "`".join(parts) if changed else None def _repair_commit_message_escapes(bindings: ToolBindings, args: Mapping[str, Any], *, tool_name: str) -> None: """Rewrite unpushed commits whose messages carry literal ``\\n`` escapes. Rebuilds ``origin/..HEAD`` with ``git commit-tree``, preserving every tree, parent topology, identity, and date — only messages change. Safe against already-pushed commits: the push transport uses ``--force-with-lease``. Once a broken message is detected the repair is mandatory — a git failure mid-rewrite refuses the push (the branch ref itself only ever moves via the compare-and-swap ``update-ref`` at the very end, so a refusal never leaves partial state). """ def fail(step: str, proc: subprocess.CompletedProcess[str]) -> NoReturn: err = (proc.stderr or proc.stdout).strip() or f"exit {proc.returncode}" msg = ( f"refusing to push: commit messages contain literal `\\n` escapes and the " f"automatic repair failed at `{step}`: {err}\n" "Reword the affected commits yourself (`git rebase -i origin/" + bindings.repo.default_branch + "`, real newlines via `git commit -F ` or multiple `-m` flags) and retry." ) _audit(bindings, tool_name, args, error=msg) _raise_command(msg) base = bindings.repo.default_branch rev_list = _run_repo_command(bindings, ["git", "rev-list", "--reverse", f"origin/{base}..HEAD"]) if rev_list.returncode != 0: return shas = rev_list.stdout.split() if not shas: return messages: dict[str, str] = {} repaired: list[str] = [] for sha in shas: show = _run_repo_command(bindings, ["git", "log", "-1", "--format=%B", sha]) if show.returncode != 0: if repaired: fail("git log", show) return message = show.stdout fixed = _repair_message_escapes(message) if fixed is not None: message = fixed repaired.append(sha) messages[sha] = message if not repaired: return needs_fix = set(repaired) rewritten: dict[str, str] = {} for sha in shas: meta = _run_repo_command( bindings, ["git", "log", "-1", "--format=%T%x1f%P%x1f%an%x1f%ae%x1f%aI%x1f%cn%x1f%ce%x1f%cI", sha], ) if meta.returncode != 0: fail("git log", meta) fields = meta.stdout.strip("\n").split("\x1f") if len(fields) != 8: fail("git log", meta) tree, parents_raw, a_name, a_email, a_date, c_name, c_email, c_date = fields parents_old = parents_raw.split() parents_new = [rewritten.get(p, p) for p in parents_old] if sha not in needs_fix and parents_new == parents_old: rewritten[sha] = sha continue cmd = ["git", "commit-tree", tree] for parent in parents_new: cmd += ["-p", parent] cmd += ["-m", messages[sha].rstrip("\n")] made = _run_repo_command( bindings, cmd, extra_env={ "GIT_AUTHOR_NAME": a_name, "GIT_AUTHOR_EMAIL": a_email, "GIT_AUTHOR_DATE": a_date, "GIT_COMMITTER_NAME": c_name, "GIT_COMMITTER_EMAIL": c_email, "GIT_COMMITTER_DATE": c_date, }, ) if made.returncode != 0 or not made.stdout.strip(): fail("git commit-tree", made) rewritten[sha] = made.stdout.strip() old_head, new_head = shas[-1], rewritten[shas[-1]] update = _run_repo_command( bindings, ["git", "update-ref", "-m", "robomp: repaired commit message escapes", "HEAD", new_head, old_head], ) if update.returncode != 0: fail("git update-ref", update) _audit(bindings, tool_name, args, result={"repaired_commit_messages": [sha[:12] for sha in repaired]}) log.info( "repaired commit message escapes", extra={"issue": bindings.issue_key, "commits": [sha[:12] for sha in repaired]}, ) def _guarded_push_branch(bindings: ToolBindings, args: Mapping[str, Any], tool_name: str, branch: str) -> str: if bindings.review_mode: msg = "refusing to push: PR review worktrees are read-only." _audit(bindings, tool_name, args, error=msg) _raise_command(msg) if branch != bindings.workspace.branch: _raise_command( f"refusing to push: branch={branch!r} does not match workspace branch {bindings.workspace.branch!r}." ) # Re-pin the configured identity right before push (cheap; idempotent). _run_repo_command(bindings, ["git", "config", "user.email", bindings.author_email]) _run_repo_command(bindings, ["git", "config", "user.name", bindings.author_name]) # Cosmetic repair BEFORE the head snapshot: commits whose messages carry # shell-literal `\n` escapes are rewritten in place (message-only). _repair_commit_message_escapes(bindings, args, tool_name=tool_name) repo_dir_path = bindings.workspace.repo_dir head_proc = _run_repo_command(bindings, ["git", "rev-parse", "HEAD"]) if head_proc.returncode != 0: err = (head_proc.stderr or head_proc.stdout).strip() or f"exit {head_proc.returncode}" _audit(bindings, tool_name, args, error=err) _raise_command(f"git rev-parse failed: {err}") head_sha = head_proc.stdout.strip() # Identity gate: every commit between the base branch and HEAD must # carry the configured author. Refuse to push otherwise so the agent # fixes it (`git commit --amend --reset-author --no-edit`). base = bindings.repo.default_branch identities = _run_repo_command( bindings, ["git", "log", "--format=%H%x09%ae%x09%an", f"origin/{base}..HEAD"], ) if identities.returncode != 0: err = (identities.stderr or identities.stdout).strip() msg = f"refusing to push: could not inspect commit authors for origin/{base}..HEAD: {err}" _audit(bindings, tool_name, args, error=msg) _raise_command(msg) offending: list[str] = [] for line in (identities.stdout or "").strip().splitlines(): parts = line.split("\t") if len(parts) < 3: continue sha, email, name = parts[0], parts[1], parts[2] if email != bindings.author_email or name != bindings.author_name: offending.append(f"{sha[:12]} {name} <{email}>") if offending: details = "\n ".join(offending) msg = ( "refusing to push: commit author identity mismatch. " f"Expected `{bindings.author_name} <{bindings.author_email}>`. " f"Offending commits:\n {details}\n" "Amend each commit with `git commit --amend --reset-author --no-edit` " "(or rebase with `git rebase -i origin/" + base + " --exec " "'git commit --amend --reset-author --no-edit'`) and try again." ) _audit(bindings, tool_name, args, error=msg) _raise_command(msg) # Working-tree cleanliness gate. Any uncommitted change (edits the agent # forgot to `git add && git commit`, files dropped by package managers, etc.) # would silently land in the PR review delta but not in the commit history. # Reject so the agent either commits or stashes them. status = _run_repo_command(bindings, ["git", "status", "--porcelain", "--untracked-files=normal"]) if status.stdout.strip(): dirty = "\n ".join(status.stdout.strip().splitlines()) msg = ( "refusing to push: working tree is dirty.\n " f"{dirty}\n" "Commit (or `git stash`) every change before pushing — anything in the " "worktree that isn't in a commit won't appear in the PR." ) _audit(bindings, tool_name, args, error=msg) _raise_command(msg) try: result = bindings.git_transport.push_branch( repo=bindings.repo.full_name, workspace_key=workspace_key(bindings.repo.full_name, bindings.issue.number), repo_dir=repo_dir_path, branch=branch, expected_head=head_sha, slot_uid=bindings.slot_uid, ) except HeadDriftError: msg = ( "refusing to push: HEAD changed between preflight and push " "(another commit landed; rerun the gate by re-issuing the push)." ) _audit(bindings, tool_name, args, error=msg) _raise_command(msg) except GitCommandError as exc: err = (exc.stderr or exc.stdout).strip() or f"exit {exc.returncode}" _audit(bindings, tool_name, args, error=err) _raise_command(f"git push failed: {err}") except GitHubError as exc: msg = f"gh-proxy rejected push: {exc.status} {exc.message}" _audit(bindings, tool_name, args, error=msg) _raise_command(msg) _share_git_metadata_with_slots(repo_dir_path, bindings.slot_uid) _audit(bindings, tool_name, args, result={"head": result.head, "branch": result.branch}) return result.head # ---------- gh_push_branch ---------- def _build_push_branch(bindings: ToolBindings) -> HostTool[Any, Any]: def execute(args: dict[str, Any], _ctx: HostToolContext[Any]) -> str: if bindings.review_mode: msg = "refusing to push: PR review worktrees are read-only." _audit(bindings, "gh_push_branch", args, error=msg) _raise_command(msg) _enforce_impl_authorization(bindings, "gh_push_branch", args, action="push branch") branch = str(args.get("branch") or bindings.workspace.branch) skip = bool(args.get("skip_checks", False)) # Same gate as gh_open_pr — formatter + check before bytes leave the # workstation, so CI doesn't blow up on a follow-up commit. The fix # pass auto-commits any formatter diff so the push includes it. # `skip_checks=true` bypasses the formatter/check (e.g. when `main` # itself is broken); dirty-tree gate still runs unconditionally. _run_pre_publish_bun_fix(bindings, args, tool_name="gh_push_branch", stage="push", skip_checks=skip) _run_pre_publish_bun_check(bindings, args, tool_name="gh_push_branch", stage="push", skip_checks=skip) head = _guarded_push_branch(bindings, args, "gh_push_branch", branch) suffix = " (pre-push checks skipped)" if skip else "" return f"pushed {branch} at {head[:12]} as {bindings.author_name} <{bindings.author_email}>{suffix}" return host_tool( name="gh_push_branch", description=persona.host_tool_description("gh_push_branch"), parameters={ "type": "object", "properties": { "branch": { "type": "string", "description": persona.host_tool_parameter_description("gh_push_branch", "branch"), }, "skip_checks": { "type": "boolean", "description": persona.host_tool_parameter_description("gh_push_branch", "skip_checks"), }, }, "additionalProperties": False, }, execute=execute, ) # ---------- gh_open_pr ---------- def _build_open_pr(bindings: ToolBindings) -> HostTool[Any, Any]: def execute(args: dict[str, Any], _ctx: HostToolContext[Any]) -> str: if bindings.review_mode: msg = "refusing to open PR: PR review tasks are read-only." _audit(bindings, "gh_open_pr", args, error=msg) _raise_command(msg) _enforce_impl_authorization(bindings, "gh_open_pr", args, action="open PR") title = args.get("title") body = args.get("body") if not isinstance(title, str) or not title.strip(): _raise_command("gh_open_pr requires a non-empty 'title'.") if not isinstance(body, str) or not body.strip(): _raise_command("gh_open_pr requires a non-empty 'body'.") for required in ("## Repro", "## Cause", "## Fix", "## Verification"): if required not in body: _raise_command( f"PR body missing required section header {required!r}. " "Follow the template in the system prompt verbatim." ) # Auto-close keyword. GitHub closes the linked issue on merge only when # one of `Fixes / Closes / Resolves #` is present in the PR body. n = bindings.issue.number accepted = [f"{kw} #{n}" for kw in ("Fixes", "Closes", "Resolves", "fixes", "closes", "resolves")] if not any(form in body for form in accepted): _raise_command( f"PR body must include `Fixes #{n}` (or `Closes #{n}` / `Resolves #{n}`) so " "GitHub auto-closes the issue when the PR merges. Put it at the end of the " "Verification section per the template." ) skip = bool(args.get("skip_checks", False)) _run_pre_publish_bun_fix(bindings, args, tool_name="gh_open_pr", stage="open PR", skip_checks=skip) _run_pre_publish_bun_check(bindings, args, tool_name="gh_open_pr", stage="open PR", skip_checks=skip) # Make sure the branch is pushed (idempotent) using the same preflight as gh_push_branch. _guarded_push_branch(bindings, args, "gh_open_pr", bindings.workspace.branch) base = args.get("base") or bindings.repo.default_branch was_needs_info = _issue_needs_info(bindings) try: pr = _run_coro( bindings.loop, bindings.github.open_pull_request( repo=bindings.repo.full_name, head=bindings.workspace.branch, base=str(base), title=title, body=body, draft=bool(args.get("draft", False)), ), ) except GitHubError as exc: _audit(bindings, "gh_open_pr", args, error=str(exc)) _raise_command(f"GitHub rejected PR: {exc.status} {exc.message}") bindings.db.set_issue_pr(bindings.issue_key, pr.number) bindings.db.set_issue_state(bindings.issue_key, "opened") needs_info_label_cleared = _remove_needs_info_label(bindings) if was_needs_info else False artifact = bindings.workspace.artifacts_dir / "pr.json" artifact.write_text( json.dumps( { "repo": pr.repo, "number": pr.number, "url": pr.html_url, "head": pr.head_ref, "base": pr.base_ref, }, indent=2, ), encoding="utf-8", ) result: dict[str, Any] = {"pr_number": pr.number, "url": pr.html_url} if needs_info_label_cleared: result["cleared_needs_info"] = True _audit(bindings, "gh_open_pr", args, result=result) return f"opened #{pr.number}: {pr.html_url}" return host_tool( name="gh_open_pr", description=persona.host_tool_description("gh_open_pr"), parameters={ "type": "object", "properties": { "title": {"type": "string"}, "body": { "type": "string", "description": persona.host_tool_parameter_description("gh_open_pr", "body"), }, "base": { "type": "string", "description": persona.host_tool_parameter_description("gh_open_pr", "base"), }, "draft": {"type": "boolean", "default": False}, "skip_checks": { "type": "boolean", "description": persona.host_tool_parameter_description("gh_open_pr", "skip_checks"), }, }, "required": ["title", "body"], "additionalProperties": False, }, execute=execute, ) # ---------- gh_request_review ---------- def _build_request_review(bindings: ToolBindings) -> HostTool[Any, Any]: def execute(args: dict[str, Any], _ctx: HostToolContext[Any]) -> str: reviewers = args.get("reviewers") or [] assignees = args.get("assignees") or [] if not isinstance(reviewers, list) or not isinstance(assignees, list): _raise_command("gh_request_review expects 'reviewers' and 'assignees' to be arrays of logins.") issue_row = bindings.db.get_issue(bindings.issue_key) pr_number = issue_row.pr_number if issue_row else None if pr_number is None: _raise_command("no PR recorded for this issue yet; call gh_open_pr first.") try: if reviewers: _run_coro( bindings.loop, bindings.github.request_reviewers( repo=bindings.repo.full_name, pr_number=pr_number, reviewers=[str(r) for r in reviewers], ), ) if assignees: _run_coro( bindings.loop, bindings.github.add_assignees( bindings.repo.full_name, pr_number, [str(a) for a in assignees], ), ) except GitHubError as exc: _audit(bindings, "gh_request_review", args, error=str(exc)) _raise_command(f"GitHub rejected review request: {exc.status} {exc.message}") _audit(bindings, "gh_request_review", args, result={"pr": pr_number}) return f"updated review/assignees on #{pr_number}" return host_tool( name="gh_request_review", description=persona.host_tool_description("gh_request_review"), parameters={ "type": "object", "properties": { "reviewers": {"type": "array", "items": {"type": "string"}}, "assignees": {"type": "array", "items": {"type": "string"}}, }, "additionalProperties": False, }, execute=execute, ) # ---------- repro_record ---------- def _build_repro_record(bindings: ToolBindings) -> HostTool[Any, Any]: def execute(args: dict[str, Any], _ctx: HostToolContext[Any]) -> str: title = args.get("title") command = args.get("command") output = args.get("output") exit_code = args.get("exit_code") if not isinstance(title, str) or not title.strip(): _raise_command("repro_record requires a non-empty 'title'.") if not isinstance(command, str) or not command.strip(): _raise_command("repro_record requires a non-empty 'command'.") if not isinstance(output, str): _raise_command("repro_record requires 'output' (may be empty string).") if not isinstance(exit_code, int): _raise_command("repro_record requires an integer 'exit_code'.") bindings.workspace.repro_dir.mkdir(parents=True, exist_ok=True) slug = "".join(c if c.isalnum() else "-" for c in title.lower()).strip("-")[:48] or "repro" ts = int(time.time()) target = bindings.workspace.repro_dir / f"{ts}-{slug}.md" target.write_text( f"# {title}\n\n" f"- exit_code: {exit_code}\n" f"- command:\n\n```\n{command}\n```\n\n" f"## Output\n\n```\n{output}\n```\n", encoding="utf-8", ) # Single-ownership invariant: workspace files belong to the active # slot. The orchestrator (root) wrote this file directly, so hand it # over before the audit row lands so the agent can edit/delete it. if _slot_permissions_active(bindings.slot_uid): assert bindings.slot_uid is not None os.chown(target, bindings.slot_uid, bindings.slot_uid) result: dict[str, Any] = {"path": str(target.relative_to(bindings.workspace.root))} if _advance_needs_info(bindings, "reproducing"): result["cleared_needs_info"] = True _audit(bindings, "repro_record", args, result=result) return "recorded" return host_tool( name="repro_record", description=persona.host_tool_description("repro_record"), parameters={ "type": "object", "properties": { "title": {"type": "string"}, "command": {"type": "string"}, "output": {"type": "string"}, "exit_code": {"type": "integer"}, "reproduced": { "type": "boolean", "description": persona.host_tool_parameter_description("repro_record", "reproduced"), }, }, "required": ["title", "command", "output", "exit_code"], "additionalProperties": False, }, execute=execute, ) # ---------- mark_unable_to_reproduce ---------- def _build_mark_unable(bindings: ToolBindings) -> HostTool[Any, Any]: def execute(args: dict[str, Any], _ctx: HostToolContext[Any]) -> str: diagnosis = args.get("diagnosis") needed = args.get("info_needed") if not isinstance(diagnosis, str) or not diagnosis.strip(): _raise_command("mark_unable_to_reproduce requires a 'diagnosis'.") if not isinstance(needed, str) or not needed.strip(): _raise_command("mark_unable_to_reproduce requires 'info_needed' explaining what to ask for.") body = persona.unable_to_reproduce_comment( diagnosis=diagnosis, info_needed=needed, ) try: comment = _run_coro( bindings.loop, bindings.github.post_comment(bindings.repo.full_name, bindings.issue.number, body), ) except GitHubError as exc: _audit(bindings, "mark_unable_to_reproduce", args, error=str(exc)) _raise_command(f"GitHub rejected comment: {exc.status} {exc.message}") result: dict[str, Any] = {"comment_id": comment.id, "state": "needs_info"} try: labels = _run_coro( bindings.loop, bindings.github.add_issue_labels(bindings.repo.full_name, bindings.issue.number, [_NEEDS_INFO_LABEL]), ) result["labels"] = list(labels) except GitHubError as exc: # Some repos have not created the optional status label yet. The # durable behavior is the non-terminal sqlite state plus the visible # info-request comment, so label setup must not block resumption. log.warning("needs-info label failed", extra={"issue": bindings.issue_key, "err": str(exc)}) result["label_error"] = f"{exc.status} {exc.message}" except Exception as exc: # noqa: BLE001 - best-effort optional label setup error = _optional_label_error(exc) log.warning("needs-info label failed", extra={"issue": bindings.issue_key, "err": error}) result["label_error"] = error bindings.db.set_issue_state(bindings.issue_key, "needs_info") _audit(bindings, "mark_unable_to_reproduce", args, result=result) return f"posted needs-info comment id={comment.id}" return host_tool( name="mark_unable_to_reproduce", description=persona.host_tool_description("mark_unable_to_reproduce"), parameters={ "type": "object", "properties": { "diagnosis": {"type": "string"}, "info_needed": {"type": "string"}, }, "required": ["diagnosis", "info_needed"], "additionalProperties": False, }, execute=execute, ) # ---------- abort_task ---------- def _build_abort_task(bindings: ToolBindings) -> HostTool[Any, Any]: def execute(args: dict[str, Any], _ctx: HostToolContext[Any]) -> str: reason = args.get("reason") if not isinstance(reason, str) or not reason.strip(): _raise_command("abort_task requires a non-empty 'reason' string.") reason = reason.strip() # Audit FIRST so the diagnosis is durable even if anything below # races against the imminent omp teardown. _audit(bindings, "abort_task", args, result={"reason": reason}) log.warning( "task_aborted", extra={"issue": bindings.issue_key, "reason": reason}, ) bindings.db.set_issue_state(bindings.issue_key, "abandoned") if bindings.abort is not None: bindings.abort.signal(reason) return "aborted" return host_tool( name="abort_task", description=persona.host_tool_description("abort_task"), parameters={ "type": "object", "properties": { "reason": {"type": "string"}, }, "required": ["reason"], "additionalProperties": False, }, execute=execute, ) # ---------- fetch_issue_thread ---------- def _build_fetch_thread(bindings: ToolBindings) -> HostTool[Any, Any]: def execute(args: dict[str, Any], _ctx: HostToolContext[Any]) -> str: try: issue = _run_coro( bindings.loop, bindings.github.get_issue(bindings.repo.full_name, bindings.issue.number), ) comments = _run_coro( bindings.loop, bindings.github.list_comments(bindings.repo.full_name, bindings.issue.number), ) except GitHubError as exc: _audit(bindings, "fetch_issue_thread", args, error=str(exc)) _raise_command(f"GitHub fetch failed: {exc.status} {exc.message}") lines = [ f"# {issue.repo}#{issue.number} ({issue.state})", f"title: {issue.title}", f"author: @{issue.author}", f"labels: {', '.join(issue.labels) if issue.labels else '(none)'}", "", "## Body", issue.body.strip() or "(empty)", "", f"## Comments ({len(comments)})", ] for c in comments: lines.extend(["", f"### @{c.author} at {c.created_at}", c.body.strip()]) rendered = "\n".join(lines) _audit(bindings, "fetch_issue_thread", args, result={"comments": len(comments)}) return rendered return host_tool( name="fetch_issue_thread", description=persona.host_tool_description("fetch_issue_thread"), parameters={ "type": "object", "properties": {}, "additionalProperties": False, }, execute=execute, ) _PRIMARY_TYPES = ("bug", "enhancement", "question", "proposal", "documentation", "invalid", "duplicate") _AUTO_PR_CLASSIFICATIONS = frozenset({"bug", "documentation"}) _PRIORITIES = ("prio:p0", "prio:p1", "prio:p2", "prio:p3") _FUNCTIONAL = ("agent", "tool", "tui", "cli", "prompting", "sdk", "auth", "setup", "ux", "providers") _PLATFORMS = ("platform:linux", "platform:macos", "platform:windows", "platform:wsl") _PR_RANKS = ("review:p0", "review:p1", "review:p2", "review:p3") _PR_TYPES = ("feat", "fix", "docs", "refactor", "perf", "test", "chore", "ci", "build") _CLOSING_ISSUE_RE = re.compile(r"\b(?:close[sd]?|fix(?:e[sd])?|resolve[sd]?)\s+#(\d+)", re.IGNORECASE) def _enforce_impl_authorization( bindings: ToolBindings, tool_name: str, args: Mapping[str, Any], *, action: str, ) -> None: """Refuse first publish on issue classes that require maintainer authorization.""" if bindings.impl_authorized: return if bindings.db.has_authorized_impl_event(bindings.issue_key): return row = bindings.db.get_issue(bindings.issue_key) if row is not None: if row.pr_number is not None: return classification = row.classification if classification in _AUTO_PR_CLASSIFICATIONS: return else: classification = None classification_phrase = f"classified `{classification}`" if classification else "not classified" msg = ( f"refusing to {action}: issue #{bindings.issue.number} is {classification_phrase}; " "a repo OWNER or allowlisted maintainer must @-mention you with an explicit go-ahead " "before any branch/PR. Post your analysis with `gh_post_comment` and stop." ) _audit(bindings, tool_name, args, error=msg) _raise_command(msg) def _require_review_mode(bindings: ToolBindings, name: str, args: Mapping[str, Any]) -> None: if bindings.review_mode: return msg = f"{name} is only available during incoming PR review tasks." _audit(bindings, name, args, error=msg) _raise_command(msg) def _format_pr_file(file: PullRequestFileInfo) -> str: return f"- `{file.path}` ({file.status}, +{file.additions}/-{file.deletions})" def _build_fetch_pr(bindings: ToolBindings) -> HostTool[Any, Any]: def execute(args: dict[str, Any], _ctx: HostToolContext[Any]) -> str: _require_review_mode(bindings, "fetch_pr", args) pr_number = bindings.default_comment_number try: pr = _run_coro(bindings.loop, bindings.github.get_pull_request(bindings.repo.full_name, pr_number)) files = _run_coro(bindings.loop, bindings.github.list_pr_files(bindings.repo.full_name, pr_number)) except GitHubError as exc: _audit(bindings, "fetch_pr", args, error=str(exc)) _raise_command(f"GitHub fetch failed: {exc.status} {exc.message}") linked = tuple(sorted({int(match.group(1)) for match in _CLOSING_ISSUE_RE.finditer(pr.body)})) lines = [ f"# {pr.repo}#{pr.number} ({pr.state})", f"title: {pr.title or '(untitled)'}", f"author: @{pr.author}", f"head: {pr.head_repo or pr.repo}:{pr.head_ref}", f"base: {pr.base_ref}", f"url: {pr.html_url}", "", "## Body", pr.body.strip() or "(empty)", "", "## Linked issues", ", ".join(f"#{n}" for n in linked) if linked else "(none found in PR body)", "", f"## Changed files ({len(files)})", ] lines.extend(_format_pr_file(file) for file in files) rendered = "\n".join(lines) _audit(bindings, "fetch_pr", args, result={"files": len(files), "linked_issues": list(linked)}) return rendered return host_tool( name="fetch_pr", description=persona.host_tool_description("fetch_pr"), parameters={"type": "object", "properties": {}, "additionalProperties": False}, execute=execute, ) def _build_classify_pr(bindings: ToolBindings) -> HostTool[Any, Any]: def execute(args: dict[str, Any], _ctx: HostToolContext[Any]) -> str: _require_review_mode(bindings, "classify_pr", args) rank = args.get("rank") if rank not in _PR_RANKS: msg = f"classify_pr 'rank' must be one of {_PR_RANKS}; got {rank!r}." _audit(bindings, "classify_pr", args, error=msg) _raise_command(msg) pr_type = args.get("type") if pr_type not in _PR_TYPES: msg = f"classify_pr 'type' must be one of {_PR_TYPES}; got {pr_type!r}." _audit(bindings, "classify_pr", args, error=msg) _raise_command(msg) rationale = args.get("rationale") if not isinstance(rationale, str) or not rationale.strip(): msg = "classify_pr requires a one-sentence 'rationale'." _audit(bindings, "classify_pr", args, error=msg) _raise_command(msg) labels: list[str] = ["triaged", str(rank), str(pr_type)] for area in args.get("area") or (): if isinstance(area, str) and area in _FUNCTIONAL: labels.append(area) provider = args.get("provider") if isinstance(provider, str) and provider.strip() and provider.startswith("provider:"): labels.append("providers") labels.append(provider) try: applied = _run_coro( bindings.loop, bindings.github.add_issue_labels(bindings.repo.full_name, bindings.default_comment_number, labels), ) except GitHubError as exc: _audit(bindings, "classify_pr", args, error=str(exc)) _raise_command(f"GitHub rejected labels: {exc.status} {exc.message}") bindings.db.set_issue_classification(bindings.issue_key, str(rank)) _audit( bindings, "classify_pr", args, result={"rank": rank, "type": pr_type, "labels": list(applied), "rationale": rationale}, ) return f"classified PR as {rank}; labels applied: {', '.join(applied)}." return host_tool( name="classify_pr", description=persona.host_tool_description("classify_pr"), parameters={ "type": "object", "properties": { "rank": { "type": "string", "enum": list(_PR_RANKS), "description": persona.host_tool_parameter_description("classify_pr", "rank"), }, "type": { "type": "string", "enum": list(_PR_TYPES), "description": persona.host_tool_parameter_description("classify_pr", "type"), }, "area": { "type": "array", "items": {"type": "string", "enum": list(_FUNCTIONAL)}, "description": persona.host_tool_parameter_description("classify_pr", "area"), }, "provider": { "type": "string", "description": persona.host_tool_parameter_description("classify_pr", "provider"), }, "rationale": { "type": "string", "description": persona.host_tool_parameter_description("classify_pr", "rationale"), }, }, "required": ["rank", "type", "rationale"], "additionalProperties": False, }, execute=execute, ) def _review_comment_to_payload(comment: Any) -> dict[str, Any]: payload: dict[str, Any] = { "path": comment.path, "line": comment.line, "side": comment.side, "body": comment.body, } if comment.start_line is not None: payload["start_line"] = comment.start_line if comment.start_side is not None: payload["start_side"] = comment.start_side return payload def _build_pr_review_comment(bindings: ToolBindings) -> HostTool[Any, Any]: def execute(args: dict[str, Any], _ctx: HostToolContext[Any]) -> str: _require_review_mode(bindings, "pr_review_comment", args) path = args.get("path") line = args.get("line") body = args.get("body") if not isinstance(path, str) or not path.strip(): msg = "pr_review_comment requires a non-empty 'path'." _audit(bindings, "pr_review_comment", args, error=msg) _raise_command(msg) if not isinstance(line, int) or line <= 0: msg = "pr_review_comment requires a positive integer 'line'." _audit(bindings, "pr_review_comment", args, error=msg) _raise_command(msg) if not isinstance(body, str) or not body.strip(): msg = "pr_review_comment requires a non-empty 'body'." _audit(bindings, "pr_review_comment", args, error=msg) _raise_command(msg) side = str(args.get("side") or "RIGHT") if side not in ("RIGHT", "LEFT"): msg = "pr_review_comment 'side' must be RIGHT or LEFT." _audit(bindings, "pr_review_comment", args, error=msg) _raise_command(msg) start_line = args.get("start_line") if start_line is not None and (not isinstance(start_line, int) or start_line <= 0): msg = "pr_review_comment 'start_line' must be a positive integer when provided." _audit(bindings, "pr_review_comment", args, error=msg) _raise_command(msg) start_side_raw = args.get("start_side") start_side = str(start_side_raw) if start_side_raw is not None else None if start_side is not None and start_side not in ("RIGHT", "LEFT"): msg = "pr_review_comment 'start_side' must be RIGHT or LEFT when provided." _audit(bindings, "pr_review_comment", args, error=msg) _raise_command(msg) staged = bindings.db.stage_review_comment( issue_key=bindings.issue_key, path=path.strip(), line=line, side=side, start_line=start_line, start_side=start_side, body=body.strip(), ) count = len(bindings.db.list_staged_review_comments(bindings.issue_key)) _audit(bindings, "pr_review_comment", args, result={"id": staged.id, "staged": count}) return f"staged review comment #{staged.id}; staged_count={count}" return host_tool( name="pr_review_comment", description=persona.host_tool_description("pr_review_comment"), parameters={ "type": "object", "properties": { "path": { "type": "string", "description": persona.host_tool_parameter_description("pr_review_comment", "path"), }, "line": { "type": "integer", "description": persona.host_tool_parameter_description("pr_review_comment", "line"), }, "body": { "type": "string", "description": persona.host_tool_parameter_description("pr_review_comment", "body"), }, "side": { "type": "string", "enum": ["RIGHT", "LEFT"], "default": "RIGHT", "description": persona.host_tool_parameter_description("pr_review_comment", "side"), }, "start_line": { "type": "integer", "description": persona.host_tool_parameter_description("pr_review_comment", "start_line"), }, "start_side": { "type": "string", "enum": ["RIGHT", "LEFT"], "description": persona.host_tool_parameter_description("pr_review_comment", "start_side"), }, }, "required": ["path", "line", "body"], "additionalProperties": False, }, execute=execute, ) def _build_submit_pr_review(bindings: ToolBindings) -> HostTool[Any, Any]: def execute(args: dict[str, Any], _ctx: HostToolContext[Any]) -> str: _require_review_mode(bindings, "submit_pr_review", args) body = args.get("body") if not isinstance(body, str) or not body.strip(): msg = "submit_pr_review requires a non-empty 'body'." _audit(bindings, "submit_pr_review", args, error=msg) _raise_command(msg) staged = bindings.db.list_staged_review_comments(bindings.issue_key) comments = [_review_comment_to_payload(comment) for comment in staged] try: review = _run_coro( bindings.loop, bindings.github.submit_pr_review( repo=bindings.repo.full_name, pr_number=bindings.default_comment_number, body=body.strip(), event="COMMENT", comments=comments, ), ) except GitHubError as exc: _audit(bindings, "submit_pr_review", args, error=str(exc)) _raise_command(f"GitHub rejected PR review: {exc.status} {exc.message}") cleared = bindings.db.clear_staged_review_comments(bindings.issue_key) _audit( bindings, "submit_pr_review", args, result={"review_id": review.id, "comments": len(comments), "cleared": cleared, "event": "COMMENT"}, ) return f"submitted PR review id={review.id}; comments={len(comments)}" return host_tool( name="submit_pr_review", description=persona.host_tool_description("submit_pr_review"), parameters={ "type": "object", "properties": { "body": { "type": "string", "description": persona.host_tool_parameter_description("submit_pr_review", "body"), }, "event": { "type": "string", "enum": ["COMMENT"], "default": "COMMENT", "description": persona.host_tool_parameter_description("submit_pr_review", "event"), }, }, "required": ["body"], "additionalProperties": False, }, execute=execute, ) def _build_set_issue_labels(bindings: ToolBindings) -> HostTool[Any, Any]: """Append labels to the originating issue (or PR).""" def execute(args: dict[str, Any], _ctx: HostToolContext[Any]) -> str: if bindings.inbound_is_pr: _audit(bindings, "set_issue_labels", args, result={"skipped": "pr_thread"}) return ( "no-op: set_issue_labels is not applicable on PR threads — PR labels are " "not used for triage. Proceed with the requested change." ) labels = args.get("labels") if not isinstance(labels, list) or not labels: _raise_command("set_issue_labels requires a non-empty 'labels' array.") cleaned = [str(lbl).strip() for lbl in labels if isinstance(lbl, str) and lbl.strip()] if not cleaned: _raise_command("set_issue_labels requires at least one non-empty label.") target_number = bindings.issue.number if isinstance(args.get("number"), int): target_number = int(args["number"]) try: applied = _run_coro( bindings.loop, bindings.github.add_issue_labels(bindings.repo.full_name, target_number, cleaned), ) except GitHubError as exc: _audit(bindings, "set_issue_labels", args, error=str(exc)) _raise_command(f"GitHub rejected labels: {exc.status} {exc.message}") _audit(bindings, "set_issue_labels", args, result={"labels": list(applied)}) return f"labels now: {', '.join(applied)}" return host_tool( name="set_issue_labels", description=persona.host_tool_description("set_issue_labels"), parameters={ "type": "object", "properties": { "labels": {"type": "array", "items": {"type": "string"}}, "number": { "type": "integer", "description": persona.host_tool_parameter_description("set_issue_labels", "number"), }, }, "required": ["labels"], "additionalProperties": False, }, execute=execute, ) def _build_classify_issue(bindings: ToolBindings) -> HostTool[Any, Any]: """Triage step. Pick a primary type, optional priority/functional/provider/platform, apply labels on GitHub, persist the primary type in sqlite, and signal which workflow branch the agent should follow.""" def execute(args: dict[str, Any], _ctx: HostToolContext[Any]) -> str: existing = bindings.db.get_issue(bindings.issue_key) if bindings.inbound_is_pr: note = ( f"no-op: classify_issue is not applicable on PR threads. " f"Issue #{bindings.issue.number} is already classified" ) if existing is not None and existing.classification: note += f" as {existing.classification!r}" note += ". Proceed with the requested change (amend the branch and push, or post a comment)." _audit(bindings, "classify_issue", args, result={"skipped": "pr_thread"}) return note if existing is not None and existing.classification: _audit(bindings, "classify_issue", args, result={"skipped": "already_classified"}) return ( f"no-op: issue #{bindings.issue.number} is already classified as " f"{existing.classification!r}. Continue with that workflow; do not re-classify." ) primary = args.get("primary") if primary not in _PRIMARY_TYPES: msg = f"classify_issue 'primary' must be one of {_PRIMARY_TYPES}; got {primary!r}." _audit(bindings, "classify_issue", args, error=msg) _raise_command(msg) rationale = args.get("rationale") if not isinstance(rationale, str) or not rationale.strip(): msg = "classify_issue requires a one-sentence 'rationale'." _audit(bindings, "classify_issue", args, error=msg) _raise_command(msg) priority = args.get("priority") if primary == "bug": if priority not in _PRIORITIES: msg = f"classify_issue requires 'priority' in {_PRIORITIES} when primary=='bug'." _audit(bindings, "classify_issue", args, error=msg) _raise_command(msg) else: # Non-bug primaries: silently drop any priority the model included # rather than rejecting the call. Some models (notably gpt-5.5 over # OpenAI Completions) treat every property as required and loop # forever when a non-empty optional value triggers a hard error. priority = None branch_slug = args.get("branch_slug") if isinstance(branch_slug, str) and branch_slug.strip(): try: branch_slug = validate_branch_slug(branch_slug) except ValueError as exc: msg = f"classify_issue rejected branch_slug: {exc}" _audit(bindings, "classify_issue", args, error=msg) _raise_command(msg) else: branch_slug = None labels: list[str] = [primary] if primary == "bug" and isinstance(priority, str): labels.append(priority) for fn in args.get("functional") or (): # Unknown functional tags are dropped silently — they aren't worth # rejecting the whole classification over. if isinstance(fn, str) and fn in _FUNCTIONAL: labels.append(fn) provider = args.get("provider") if isinstance(provider, str) and provider.strip() and provider.startswith("provider:"): labels.append("providers") labels.append(provider) platform = args.get("platform") if isinstance(platform, str) and platform in _PLATFORMS: labels.append(platform) labels.append("triaged") renamed_to: str | None = None if branch_slug: try: renamed_to = rename_workspace_branch( bindings.workspace, branch_slug, pr_number=existing.pr_number if existing is not None else None, slot_uid=bindings.slot_uid, ) except ValueError as exc: _audit(bindings, "classify_issue", args, error=str(exc)) _raise_command(f"classify_issue rejected branch_slug: {exc}") except GitCommandError as exc: _audit(bindings, "classify_issue", args, error=str(exc)) _raise_command(f"classify_issue could not rename branch: {exc}") if renamed_to != bindings.workspace.branch: # rename_workspace_branch already mutated workspace.branch on # success; this branch is purely defensive — kept so a future # refactor of that helper still surfaces the mismatch. _raise_command("classify_issue internal: branch rename inconsistent.") bindings.db.set_issue_branch(bindings.issue_key, renamed_to) try: applied = _run_coro( bindings.loop, bindings.github.add_issue_labels( bindings.repo.full_name, bindings.issue.number, labels, ), ) except GitHubError as exc: _audit(bindings, "classify_issue", args, error=str(exc)) _raise_command(f"GitHub rejected labels: {exc.status} {exc.message}") bindings.db.set_issue_classification(bindings.issue_key, primary) _audit( bindings, "classify_issue", args, result={ "primary": primary, "labels": list(applied), "rationale": rationale, "branch": renamed_to, }, ) # Echo back the workflow the agent should now follow. The persona prompt # already describes each branch; the tool result reminds it. next_step = persona.classify_next_step(str(primary)) suffix = f" Branch renamed to `{renamed_to}`." if renamed_to else "" return f"classified as {primary}; labels applied: {', '.join(applied)}.{suffix} Next: {next_step}." return host_tool( name="classify_issue", description=persona.host_tool_description("classify_issue"), parameters={ "type": "object", "properties": { "primary": { "type": "string", "enum": list(_PRIMARY_TYPES), "description": persona.host_tool_parameter_description("classify_issue", "primary"), }, "priority": { "type": "string", "enum": list(_PRIORITIES), "description": persona.host_tool_parameter_description("classify_issue", "priority"), }, "functional": { "type": "array", "items": {"type": "string", "enum": list(_FUNCTIONAL)}, "description": persona.host_tool_parameter_description("classify_issue", "functional"), }, "provider": { "type": "string", "description": persona.host_tool_parameter_description("classify_issue", "provider"), }, "platform": { "type": "string", "enum": list(_PLATFORMS), "description": persona.host_tool_parameter_description("classify_issue", "platform"), }, "rationale": { "type": "string", "description": persona.host_tool_parameter_description("classify_issue", "rationale"), }, "branch_slug": { "type": "string", "description": persona.host_tool_parameter_description("classify_issue", "branch_slug"), }, }, "required": ["primary", "rationale"], "additionalProperties": False, }, execute=execute, ) def build(bindings: ToolBindings) -> tuple[HostTool[Any, Any], ...]: """Return the full set of host tools bound to one task's context. The toolset is intentionally identical across all task kinds so the LLM prompt cache stays warm across triage → follow-up → PR-conversation transitions. Triage tools (`classify_issue`, `set_issue_labels`) enforce their own scope at execution time — see the `inbound_is_pr` and already-classified guards inside `_build_classify_issue` / `_build_set_issue_labels`. """ return ( _build_classify_issue(bindings), _build_set_issue_labels(bindings), _build_fetch_pr(bindings), _build_classify_pr(bindings), _build_pr_review_comment(bindings), _build_submit_pr_review(bindings), _build_post_comment(bindings), _build_push_branch(bindings), _build_open_pr(bindings), _build_request_review(bindings), _build_repro_record(bindings), _build_mark_unable(bindings), _build_abort_task(bindings), _build_fetch_thread(bindings), ) __all__ = ["AbortController", "ToolBindings", "build"]