chore: import upstream snapshot with attribution
This commit is contained in:
wehub-resource-sync
@@ -0,0 +1,212 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sys
|
||||
from typing import Any, TextIO
|
||||
|
||||
import aisuite as ai
|
||||
|
||||
|
||||
class ApprovalController:
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
input_stream: TextIO = sys.stdin,
|
||||
output_stream: TextIO = sys.stdout,
|
||||
):
|
||||
self.input_stream = input_stream
|
||||
self.output_stream = output_stream
|
||||
self.always_allow_tools: set[str] = set()
|
||||
self.always_allow_commands: set[str] = set()
|
||||
|
||||
def evaluate(self, context: ai.ToolPolicyContext) -> ai.ToolPolicyDecision:
|
||||
metadata = context.tool_metadata
|
||||
command = _exact_command(context)
|
||||
if command and command in self.always_allow_commands:
|
||||
return ai.ToolPolicyDecision(
|
||||
allowed=True,
|
||||
reason="command allowed for session",
|
||||
)
|
||||
if context.tool_name in self.always_allow_tools:
|
||||
return ai.ToolPolicyDecision(
|
||||
allowed=True,
|
||||
reason="tool allowed for session",
|
||||
)
|
||||
if metadata is None or not metadata.requires_approval:
|
||||
return ai.ToolPolicyDecision(allowed=True, reason="low risk")
|
||||
|
||||
self._print_approval_request(context)
|
||||
choice = self._read_choice()
|
||||
if choice == "c" and command:
|
||||
self.always_allow_commands.add(command)
|
||||
return ai.ToolPolicyDecision(
|
||||
allowed=True,
|
||||
reason="command allowed for session",
|
||||
)
|
||||
if choice == "a":
|
||||
self.always_allow_tools.add(context.tool_name)
|
||||
return ai.ToolPolicyDecision(
|
||||
allowed=True,
|
||||
reason="tool allowed for session",
|
||||
)
|
||||
if choice == "y":
|
||||
return ai.ToolPolicyDecision(allowed=True, reason="approved by user")
|
||||
return ai.ToolPolicyDecision(allowed=False, reason="denied by user")
|
||||
|
||||
def _print_approval_request(self, context: ai.ToolPolicyContext) -> None:
|
||||
metadata = context.tool_metadata
|
||||
action = _approval_action(context)
|
||||
effect = _approval_effect(context)
|
||||
preview = _approval_preview(context)
|
||||
risk = metadata.risk_level if metadata else "unknown"
|
||||
category = metadata.category if metadata and metadata.category else "-"
|
||||
|
||||
print("\nPermission required", file=self.output_stream)
|
||||
print(" Action", file=self.output_stream)
|
||||
print(f" {action}", file=self.output_stream)
|
||||
print(" Risk", file=self.output_stream)
|
||||
print(f" {risk} · {category}", file=self.output_stream)
|
||||
print(" Effect", file=self.output_stream)
|
||||
print(f" {effect}", file=self.output_stream)
|
||||
print(" Preview", file=self.output_stream)
|
||||
for line in preview:
|
||||
print(f" {line}", file=self.output_stream)
|
||||
prompt = " Allow? [y] once [n] deny [a] always this tool"
|
||||
if _exact_command(context):
|
||||
prompt += " [c] always this command"
|
||||
print(prompt, file=self.output_stream)
|
||||
|
||||
def _read_choice(self) -> str:
|
||||
print("> ", end="", file=self.output_stream, flush=True)
|
||||
choice = self.input_stream.readline().strip().lower()
|
||||
if choice in {"y", "yes"}:
|
||||
return "y"
|
||||
if choice in {"a", "always"}:
|
||||
return "a"
|
||||
if choice in {"c", "command"}:
|
||||
return "c"
|
||||
return "n"
|
||||
|
||||
|
||||
def _approval_action(context: ai.ToolPolicyContext) -> str:
|
||||
args = context.arguments or {}
|
||||
if context.tool_name == "run_shell":
|
||||
command = args.get("command") or "-"
|
||||
return f"run shell command: {_single_line(str(command), 140)}"
|
||||
if context.tool_name == "write_file":
|
||||
return f"write file: {args.get('path', '-')}"
|
||||
if context.tool_name == "replace_in_file":
|
||||
return f"edit file: {args.get('path', '-')}"
|
||||
if context.tool_name == "apply_patch":
|
||||
return "apply Codex-style patch"
|
||||
if context.tool_name == "apply_unified_diff":
|
||||
return "apply unified diff"
|
||||
if context.tool_name == "review_changes":
|
||||
return "invoke reviewer subagent"
|
||||
return f"run tool: {context.tool_name}"
|
||||
|
||||
|
||||
def _approval_effect(context: ai.ToolPolicyContext) -> str:
|
||||
args = context.arguments or {}
|
||||
if context.tool_name == "run_shell":
|
||||
return "Executes a command in the configured workspace."
|
||||
if context.tool_name == "write_file":
|
||||
overwrite = args.get("overwrite", True)
|
||||
mode = "create or overwrite" if overwrite else "create if missing"
|
||||
return f"May {mode} a file under the configured workspace."
|
||||
if context.tool_name == "replace_in_file":
|
||||
expected = args.get("expected_replacements", 1)
|
||||
return f"Replaces exact text in one file; expected replacements: {expected}."
|
||||
if context.tool_name == "apply_patch":
|
||||
return "Applies one or more targeted file edits from a Codex-style patch."
|
||||
if context.tool_name == "apply_unified_diff":
|
||||
return "Applies one or more file edits from a unified diff."
|
||||
if context.tool_name == "review_changes":
|
||||
return "Runs a read-only reviewer subagent and records a child trace."
|
||||
return "Runs the tool with the summarized arguments below."
|
||||
|
||||
|
||||
def _approval_preview(context: ai.ToolPolicyContext) -> list[str]:
|
||||
args = context.arguments or {}
|
||||
if context.tool_name == "run_shell":
|
||||
return [f"command: {_summarize_argument(args.get('command'), max_chars=260)}"]
|
||||
if context.tool_name == "write_file":
|
||||
return [
|
||||
f"path: {args.get('path', '-')}",
|
||||
f"content: {_text_stats(args.get('content'))}",
|
||||
]
|
||||
if context.tool_name == "replace_in_file":
|
||||
return [
|
||||
f"path: {args.get('path', '-')}",
|
||||
f"old: {_text_stats(args.get('old'))}",
|
||||
f"new: {_text_stats(args.get('new'))}",
|
||||
]
|
||||
if context.tool_name == "apply_patch":
|
||||
return [f"patch: {_text_stats(args.get('patch'))}"]
|
||||
if context.tool_name == "apply_unified_diff":
|
||||
return [f"diff: {_text_stats(args.get('diff'))}"]
|
||||
if context.tool_name == "review_changes":
|
||||
return [f"input: {_summarize_argument(args.get('input'), max_chars=260)}"]
|
||||
if not args:
|
||||
return ["no arguments"]
|
||||
return [f"{key}: {_summarize_argument(value)}" for key, value in args.items()]
|
||||
|
||||
|
||||
def _summarize_argument(value: Any, max_chars: int = 700) -> str:
|
||||
if _is_artifact_ref(value):
|
||||
ref = value.get("artifact_ref") or {}
|
||||
preview = value.get("preview")
|
||||
pieces = [
|
||||
f"artifact {ref.get('artifact_id')}",
|
||||
(
|
||||
f"{ref.get('size_bytes')} bytes"
|
||||
if ref.get("size_bytes") is not None
|
||||
else None
|
||||
),
|
||||
]
|
||||
if preview:
|
||||
pieces.append(f"preview={_single_line(str(preview), 180)}")
|
||||
return " · ".join(piece for piece in pieces if piece)
|
||||
if isinstance(value, str):
|
||||
lines = len(value.splitlines())
|
||||
if len(value) <= max_chars:
|
||||
return value
|
||||
return (
|
||||
f"{len(value)} chars · {lines} lines · "
|
||||
f"preview={_single_line(value, 360)}"
|
||||
)
|
||||
try:
|
||||
rendered = json.dumps(value, sort_keys=True)
|
||||
except TypeError:
|
||||
rendered = str(value)
|
||||
if len(rendered) <= max_chars:
|
||||
return rendered
|
||||
return f"{len(rendered)} chars · preview={_single_line(rendered, 360)}"
|
||||
|
||||
|
||||
def _is_artifact_ref(value: Any) -> bool:
|
||||
return (
|
||||
isinstance(value, dict)
|
||||
and value.get("type") == "artifact_ref"
|
||||
and isinstance(value.get("artifact_ref"), dict)
|
||||
)
|
||||
|
||||
|
||||
def _single_line(value: str, max_chars: int) -> str:
|
||||
rendered = value.replace("\n", "\\n")
|
||||
if len(rendered) <= max_chars:
|
||||
return rendered
|
||||
return rendered[: max_chars - 3] + "..."
|
||||
|
||||
|
||||
def _exact_command(context: ai.ToolPolicyContext) -> str | None:
|
||||
if context.tool_name != "run_shell":
|
||||
return None
|
||||
command = context.arguments.get("command") if context.arguments else None
|
||||
return command if isinstance(command, str) and command else None
|
||||
|
||||
|
||||
def _text_stats(value: Any) -> str:
|
||||
if not isinstance(value, str):
|
||||
return "-"
|
||||
return f"{len(value)} chars · {len(value.splitlines())} lines"
|
||||
Reference in New Issue
Block a user