"""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, 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") _REPO_COMMAND_SCRUBBED_ENV_KEYS: tuple[str, ...] = ( "GITHUB_TOKEN", "GITHUB_WEBHOOK_SECRET", "ROBOMP_REPLAY_TOKEN", "ROBOMP_GH_PROXY_HMAC_KEY", ) _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 _PRE_PR_FIX_COMMIT_SUBJECT = "style: bun run fix" @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 _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, ) -> subprocess.CompletedProcess[str]: """Run a repo-local command with agent-equivalent permissions and env.""" return subprocess.run( list(cmd), cwd=str(bindings.workspace.repo_dir), check=False, capture_output=True, text=True, timeout=timeout, env=_repo_command_env(bindings), **_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 _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 commit any working-tree diff as the bot. Silently no-ops when the repository does not define a `scripts.fix` entry. Anything the formatter touches gets folded into a fresh `style: bun run fix` commit so the downstream cleanliness gate sees a pristine worktree. `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 `style: bun run fix` commit 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 folded into the `style: bun run fix` " "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 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", "-c", f"user.email={bindings.author_email}", "-c", f"user.name={bindings.author_name}", "commit", "-m", _PRE_PR_FIX_COMMIT_SUBJECT, ], ) if commit.returncode != 0: err = (commit.stderr or commit.stdout).strip() msg = f"refusing to {stage}: failed to commit `bun run fix` changes: {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", "abandoned"}) def _should_schedule_autoclose(bindings: ToolBindings, target_number: int) -> float | None: """Return the configured close window (hours) when this comment should schedule an auto-close; ``None`` otherwise. Conditions: feature enabled in `Settings`, the comment lands on the originating issue (not a different number, not a PR thread), the issue is classified as `question`, and the issue is not already in a terminal state (closed/merged/abandoned). """ 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 _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]) 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 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") 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", ) _audit(bindings, "gh_open_pr", args, result={"pr_number": pr.number, "url": pr.html_url}) 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) _audit(bindings, "repro_record", args, result={"path": str(target.relative_to(bindings.workspace.root))}) 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}") bindings.db.set_issue_state(bindings.issue_key, "abandoned") _audit(bindings, "mark_unable_to_reproduce", args, result={"comment_id": comment.id}) return f"posted abandonment 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 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"]