(magpie) 03/13: feat(adversarial-review): read-only backend adapters for codex, copilot, gemini, claude

[po...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

potiuk pushed a commit to branch feat/adversarial-review-tool
in repository https://gitbox.apache.org/repos/asf/magpie.git

commit e82801e9dcb367597961469093922bc1fcaa8658
Author: Jarek Potiuk <po...@apache.org>
AuthorDate: Thu Sep 24 00:19:39 2026 +0200

    feat(adversarial-review): read-only backend adapters for codex, copilot, gemini, claude
    
    Generated-by: Claude Opus 5
---
 .../src/adversarial_review/backends.py             | 188 +++++++++++++++++++++
 tools/adversarial-review/tests/test_backends.py    | 175 +++++++++++++++++++
 2 files changed, 363 insertions(+)

diff --git a/tools/adversarial-review/src/adversarial_review/backends.py b/tools/adversarial-review/src/adversarial_review/backends.py
new file mode 100644
index 00000000..da9f2067
--- /dev/null
+++ b/tools/adversarial-review/src/adversarial_review/backends.py
@@ -0,0 +1,188 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""
+One adapter per reviewer CLI: its headless, read-only command line, how the
+harness it belongs to is recognised, and where its final answer sits in its output.
+
+The command lines are this package's security surface. Each one keeps the
+reviewer read-only, and none carries the prompt in argv: the diff can exceed
+ARG_MAX, so the prompt goes on stdin, or through a brief file for Copilot, whose
+`-p` takes text only. `tests/test_backends.py` snapshots every argv and rejects
+known write-granting flags, so a regression that drops a read-only flag fails.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Callable
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+COPILOT_INSTRUCTION = (
+    "Read the review brief at {brief} and follow it exactly. Change no file and run no command."
+)
+STDIN_INSTRUCTION = "Follow the review brief given on standard input exactly. Change no file."
+CLAUDE_DENIED_TOOLS = "Bash,Edit,Write,NotebookEdit,WebFetch,WebSearch"
+
+
+class BackendOutputError(ValueError):
+    """The CLI exited 0, but its output carries no usable final answer."""
+
+
+@dataclass(frozen=True)
+class RunContext:
+    repo_dir: Path
+    prompt: str
+    brief_path: Path
+    schema_path: Path
+    last_message_path: Path
+    model: str | None = None
+
+
+@dataclass(frozen=True)
+class Invocation:
+    argv: list[str]
+    stdin: str | None
+
+
+@dataclass(frozen=True)
+class Backend:
+    name: str
+    # (environment variable, required value — None means any non-empty value)
+    self_markers: tuple[tuple[str, str | None], ...]
+    build: Callable[[RunContext], Invocation]
+    extract: Callable[[str, RunContext], str]
+
+
+def _model(flag: str, ctx: RunContext) -> list[str]:
+    return [flag, ctx.model] if ctx.model else []
+
+
+def _json_envelope(stdout: str, what: str) -> dict[str, Any]:
+    try:
+        data = json.loads(stdout)
+    except ValueError as exc:
+        raise BackendOutputError(f"{what} output is not JSON: {exc}") from None
+    if not isinstance(data, dict):
+        raise BackendOutputError(f"{what} output is not a JSON object")
+    return data
+
+
+def _codex(ctx: RunContext) -> Invocation:
+    argv = [
+        "codex",
+        "exec",
+        "-s",
+        "read-only",
+        "--ephemeral",
+        "--skip-git-repo-check",
+        "-C",
+        str(ctx.repo_dir),
+        "--output-schema",
+        str(ctx.schema_path),
+        "-o",
+        str(ctx.last_message_path),
+        *_model("-m", ctx),
+        "-",
+    ]
+    return Invocation(argv, stdin=ctx.prompt)
+
+
+def _codex_extract(stdout: str, ctx: RunContext) -> str:
+    try:
+        text = ctx.last_message_path.read_text(encoding="utf-8")
+    except FileNotFoundError:
+        raise BackendOutputError("codex wrote no final message") from None
+    if not text.strip():
+        raise BackendOutputError("codex's final message is empty")
+    return text
+
+
+def _copilot(ctx: RunContext) -> Invocation:
+    argv = [
+        "copilot",
+        "-p",
+        COPILOT_INSTRUCTION.format(brief=ctx.brief_path),
+        "--add-dir",
+        str(ctx.brief_path.parent),
+        "--deny-tool",
+        "shell",
+        "--deny-tool",
+        "write",
+        "--no-color",
+        "--log-level",
+        "none",
+        *_model("--model", ctx),
+    ]
+    return Invocation(argv, stdin=None)
+
+
+def _plain_extract(stdout: str, ctx: RunContext) -> str:
+    if not stdout.strip():
+        raise BackendOutputError("empty reply")
+    return stdout
+
+
+def _gemini(ctx: RunContext) -> Invocation:
+    argv = ["gemini", "--approval-mode", "plan", "-o", "json", *_model("-m", ctx), "-p", STDIN_INSTRUCTION]
+    return Invocation(argv, stdin=ctx.prompt)
+
+
+def _gemini_extract(stdout: str, ctx: RunContext) -> str:
+    data = _json_envelope(stdout, "gemini")
+    if data.get("error"):
+        raise BackendOutputError(f"gemini reported an error: {data['error']}")
+    response = data.get("response")
+    if not isinstance(response, str) or not response.strip():
+        raise BackendOutputError("gemini output has no `response` text")
+    return response
+
+
+def _claude(ctx: RunContext) -> Invocation:
+    argv = [
+        "claude",
+        "-p",
+        "--output-format",
+        "json",
+        "--disallowedTools",
+        CLAUDE_DENIED_TOOLS,
+        *_model("--model", ctx),
+    ]
+    return Invocation(argv, stdin=ctx.prompt)
+
+
+def _claude_extract(stdout: str, ctx: RunContext) -> str:
+    data = _json_envelope(stdout, "claude")
+    result = data.get("result")
+    if data.get("is_error"):
+        raise BackendOutputError(f"claude reported an error: {result}")
+    if not isinstance(result, str) or not result.strip():
+        raise BackendOutputError("claude output has no `result` text")
+    return result
+
+
+# Order matters for self-detection: a harness started from inside another
+# inherits the outer one's variables, so the innermost candidates are checked
+# first and Claude Code's widely inherited CLAUDECODE comes last.
+BACKENDS: dict[str, Backend] = {
+    "codex": Backend("codex", (("CODEX_SANDBOX", None), ("CODEX_THREAD_ID", None)), _codex, _codex_extract),
+    "copilot": Backend("copilot", (("COPILOT_CLI", None),), _copilot, _plain_extract),
+    "gemini": Backend("gemini", (("GEMINI_CLI", "1"),), _gemini, _gemini_extract),
+    "claude": Backend("claude", (("CLAUDECODE", "1"),), _claude, _claude_extract),
+}
diff --git a/tools/adversarial-review/tests/test_backends.py b/tools/adversarial-review/tests/test_backends.py
new file mode 100644
index 00000000..f2af7a6e
--- /dev/null
+++ b/tools/adversarial-review/tests/test_backends.py
@@ -0,0 +1,175 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from adversarial_review.backends import BACKENDS, BackendOutputError, RunContext
+
+CTX = RunContext(
+    repo_dir=Path("/repo"),
+    prompt="PROMPT",
+    brief_path=Path("/t/brief.md"),
+    schema_path=Path("/t/findings.schema.json"),
+    last_message_path=Path("/t/codex-last-message.json"),
+)
+WRITE_GRANTING = {
+    "--allow-all-tools",
+    "--allow-all-paths",
+    "yolo",
+    "--yolo",
+    "auto_edit",
+    "danger-full-access",
+    "workspace-write",
+    "--full-auto",
+    "--dangerously-bypass-approvals-and-sandbox",
+    "--dangerously-skip-permissions",
+    "bypassPermissions",
+}
+
+
+def test_backend_set_and_order():
+    assert list(BACKENDS) == ["codex", "copilot", "gemini", "claude"]
+
+
+def test_codex_argv():
+    inv = BACKENDS["codex"].build(CTX)
+    assert inv.argv == [
+        "codex",
+        "exec",
+        "-s",
+        "read-only",
+        "--ephemeral",
+        "--skip-git-repo-check",
+        "-C",
+        "/repo",
+        "--output-schema",
+        "/t/findings.schema.json",
+        "-o",
+        "/t/codex-last-message.json",
+        "-",
+    ]
+    assert inv.stdin == "PROMPT"
+
+
+def test_copilot_argv():
+    inv = BACKENDS["copilot"].build(CTX)
+    assert inv.argv == [
+        "copilot",
+        "-p",
+        "Read the review brief at /t/brief.md and follow it exactly. Change no file and run no command.",
+        "--add-dir",
+        "/t",
+        "--deny-tool",
+        "shell",
+        "--deny-tool",
+        "write",
+        "--no-color",
+        "--log-level",
+        "none",
+    ]
+    assert inv.stdin is None
+
+
+def test_gemini_argv():
+    inv = BACKENDS["gemini"].build(CTX)
+    assert inv.argv == [
+        "gemini",
+        "--approval-mode",
+        "plan",
+        "-o",
+        "json",
+        "-p",
+        "Follow the review brief given on standard input exactly. Change no file.",
+    ]
+    assert inv.stdin == "PROMPT"
+
+
+def test_claude_argv():
+    inv = BACKENDS["claude"].build(CTX)
+    assert inv.argv == [
+        "claude",
+        "-p",
+        "--output-format",
+        "json",
+        "--disallowedTools",
+        "Bash,Edit,Write,NotebookEdit,WebFetch,WebSearch",
+    ]
+    assert inv.stdin == "PROMPT"
+
+
+@pytest.mark.parametrize(
+    ("name", "flag"), [("codex", "-m"), ("copilot", "--model"), ("gemini", "-m"), ("claude", "--model")]
+)
+def test_model_override_is_passed(name, flag):
+    ctx = RunContext(**{**CTX.__dict__, "model": "some-model"})
+    argv = BACKENDS[name].build(ctx).argv
+    assert argv[argv.index(flag) + 1] == "some-model"
+
+
+@pytest.mark.parametrize("name", list(BACKENDS))
+def test_no_backend_grants_writes_or_carries_the_prompt_in_argv(name):
+    ctx = RunContext(**{**CTX.__dict__, "prompt": "DIFF-BODY-MUST-NOT-BE-IN-ARGV"})
+    argv = BACKENDS[name].build(ctx).argv
+    assert not WRITE_GRANTING & set(argv)
+    assert not any("DIFF-BODY-MUST-NOT-BE-IN-ARGV" in a for a in argv)
+
+
+def test_codex_extract_reads_the_last_message_file(tmp_path):
+    ctx = RunContext(**{**CTX.__dict__, "last_message_path": tmp_path / "last.json"})
+    (tmp_path / "last.json").write_text('{"findings": []}', encoding="utf-8")
+    assert BACKENDS["codex"].extract("ignored stdout", ctx) == '{"findings": []}'
+
+
+def test_codex_extract_without_file_is_an_output_error(tmp_path):
+    ctx = RunContext(**{**CTX.__dict__, "last_message_path": tmp_path / "missing.json"})
+    with pytest.raises(BackendOutputError, match="no final message"):
+        BACKENDS["codex"].extract("", ctx)
+
+
+def test_gemini_extract_unwraps_response():
+    assert BACKENDS["gemini"].extract(json.dumps({"response": "R"}), CTX) == "R"
+
+
+def test_gemini_extract_error_envelope():
+    with pytest.raises(BackendOutputError, match="gemini reported an error"):
+        BACKENDS["gemini"].extract(json.dumps({"error": {"message": "quota"}}), CTX)
+
+
+def test_claude_extract_unwraps_result():
+    assert BACKENDS["claude"].extract(json.dumps({"result": "R", "is_error": False}), CTX) == "R"
+
+
+def test_claude_extract_is_error():
+    with pytest.raises(BackendOutputError, match="claude reported an error"):
+        BACKENDS["claude"].extract(json.dumps({"result": "Invalid API key", "is_error": True}), CTX)
+
+
+@pytest.mark.parametrize("name", ["gemini", "claude"])
+def test_json_envelope_backends_reject_non_json(name):
+    with pytest.raises(BackendOutputError, match="not JSON"):
+        BACKENDS[name].extract("plain text", CTX)
+
+
+def test_copilot_extract_is_plain_stdout_but_not_empty():
+    assert BACKENDS["copilot"].extract("text", CTX) == "text"
+    with pytest.raises(BackendOutputError, match="empty"):
+        BACKENDS["copilot"].extract("  \n", CTX)