From ad58a9207ac4a2f8d222963dc28822a487e610fe Mon Sep 17 00:00:00 2001 From: kappaseijin4codex Date: Sun, 13 Sep 2026 15:34:26 +0900 Subject: [PATCH 1/2] feat(g4): N1 input preconditions: prewrite, OAuth token, ready screen (#444) Implements the design in docs/decisions/2026-09-13T151535_issue-444-n1- input-preconditions.md. Without any of the three, the case is unknown and no prompt is sent. - Prewrite: before every fresh/resume launch, merge hasCompletedOnboarding, theme=dark and projects..hasTrustDialogAccepted into $GATE_CLAUDE_CONFIG/.claude.json. Only a canonical directory inside RUN_ROOT, never through a symlink, other keys kept, atomic 0600 write, read back; the record holds the path, the keys and the read-back only. - Token: the runner only looks at CLAUDE_CODE_OAUTH_TOKEN in the environment and lets the pilot inherit it; ANTHROPIC_API_KEY and ANTHROPIC_AUTH_TOKEN are unset. Unset, empty or control characters: neither case starts a launcher (auth_token_absent). N1/auth.json and capture-environment record presence and names only. - Ready: after the binding checks, pty.raw with ANSI escapes removed must show "? for shortcuts" and never "Select login method" or "Quick safety check", within 30 s, on a CLI version the verifier has observed (2.1.268). No key is sent to get past a screen. - pilot-pty.py sets the child terminal to 120x40. Tests N1P-01..15 with synthetic credential values only. Refs #444 #434 #407 Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Gd6MbRAYGpkDzHAk1XZroZ --- scripts/lib/pilot-gate-isolation.py | 278 ++++++++++++++++++++++++++++ scripts/lib/pilot-pty.py | 14 ++ scripts/pilot-gate-runner.sh | 104 ++++++++++- tests/test_pilot_gate_isolation.py | 231 +++++++++++++++++++++++ tests/test_pilot_gate_runner.bats | 256 +++++++++++++++++++++++++ tests/test_pilot_pty.py | 11 ++ 6 files changed, 892 insertions(+), 2 deletions(-) diff --git a/scripts/lib/pilot-gate-isolation.py b/scripts/lib/pilot-gate-isolation.py index 91f9dd0f..ad369072 100755 --- a/scripts/lib/pilot-gate-isolation.py +++ b/scripts/lib/pilot-gate-isolation.py @@ -70,6 +70,30 @@ "GITHUB_ENTERPRISE_TOKEN", ) +# #444: the native pilot's credential and the credentials it must not use. +# Only presence is ever recorded; values never leave the environment. +N1_OAUTH_TOKEN_ENV = "CLAUDE_CODE_OAUTH_TOKEN" +N1_COMPETING_AUTH_ENV_KEYS = ( + "ANTHROPIC_API_KEY", + "ANTHROPIC_AUTH_TOKEN", +) +CLAUDE_AUTH_ENV_KEYS = ( + N1_OAUTH_TOKEN_ENV, + *N1_COMPETING_AUTH_ENV_KEYS, +) + +# #444 section 5.2: screens that mean the prewrite did not take effect, and +# the one text that means the prompt input is ready. +N1_BLOCKING_SCREENS = ( + ("Select login method", "login_method_screen"), + ("Quick safety check", "trust_dialog_screen"), +) +N1_READY_TEXT = "? for shortcuts" +ANSI_ESCAPE_RE = re.compile( + r"\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)" # OSC ... BEL or ST + r"|\x1b\[[0-?]*[ -/]*[@-~]" # CSI +) + UUID_RE = re.compile( r"^[0-9a-f]{8}-" r"[0-9a-f]{4}-" @@ -1120,6 +1144,12 @@ def command_capture_environment( in GITHUB_CREDENTIAL_ENV_KEYS } + # #444: presence only, as for the GitHub credentials above. + claude_auth_presence = { + key: bool(os.environ.get(key)) + for key in CLAUDE_AUTH_ENV_KEYS + } + git_version = run_command( [ "git", @@ -1168,6 +1198,8 @@ def command_capture_environment( # Presence only. Secret values are deliberately not recorded. "githubCredentialEnvironmentPresent": sensitive_presence, + "claudeAuthEnvironmentPresent": + claude_auth_presence, } write_json( @@ -3726,6 +3758,250 @@ def command_write_n1_result( return 0 +# --- #444: N1 input preconditions --------------------------------------------- + + +def n1_oauth_token_usable(value: str | None) -> bool: + if not value: + return False + return not any(ord(ch) < 0x20 or ord(ch) == 0x7F for ch in value) + + +def command_n1_auth( + args: argparse.Namespace, +) -> int: + """Record whether the pilot's OAuth token is usable (#444 section 4). + + Reads only the environment. Writes presence and the names of the + competing variables the runner unsets, never a value, length or digest. + """ + present = n1_oauth_token_usable(os.environ.get(N1_OAUTH_TOKEN_ENV)) + write_json( + args.output, + { + "schemaVersion": 1, + "oauthTokenEnv": "present" if present else "absent", + "unsetCompetingEnv": list(N1_COMPETING_AUTH_ENV_KEYS), + }, + ) + return 0 if present else 2 + + +def _inside(path: pathlib.Path, root: pathlib.Path) -> bool: + return path != root and root in path.parents + + +def command_prewrite_claude_config( + args: argparse.Namespace, +) -> int: + """Merge the three onboarding/trust keys into the gate's .claude.json. + + #444 section 3: only inside RUN_ROOT, never through a symlink, other keys + kept, atomic 0600 write, read back. Exit 0 when verified, 2 (unknown) + otherwise; the record names the reason. + """ + config_dir = pathlib.Path(args.config_dir) + target = config_dir / ".claude.json" + gate_repo = args.gate_repo + wanted = { + "hasCompletedOnboarding": True, + "theme": "dark", + "projects": {gate_repo: {"hasTrustDialogAccepted": True}}, + } + record: dict[str, Any] = { + "schemaVersion": 1, + "path": str(target), + "set": wanted, + "written": False, + "readBack": "not_attempted", + "verdict": "unknown", + "reason": None, + } + + def finish(reason: str | None) -> int: + record["reason"] = reason + record["verdict"] = "pass" if reason is None else "unknown" + write_json(args.output, record) + return 0 if reason is None else 2 + + try: + run_root = pathlib.Path(os.path.realpath(args.run_root)) + resolved_dir = pathlib.Path(os.path.realpath(config_dir)) + except OSError: + return finish("claude_config_outside_run_root") + if ( + not config_dir.is_absolute() + or str(resolved_dir) != str(config_dir) + or str(run_root) != args.run_root + or not _inside(resolved_dir, run_root) + ): + return finish("claude_config_outside_run_root") + + try: + config_dir.mkdir(parents=True, exist_ok=True) + if config_dir.is_symlink() or not config_dir.is_dir(): + return finish("claude_config_outside_run_root") + + data: dict[str, Any] = {} + try: + info = os.lstat(target) + except FileNotFoundError: + info = None + if info is not None: + if not stat.S_ISREG(info.st_mode): + return finish("claude_config_unreadable") + try: + with open(target, "r", encoding="utf-8") as fh: + loaded = json.load(fh) + except (OSError, UnicodeError, ValueError): + return finish("claude_config_unreadable") + if not isinstance(loaded, dict): + return finish("claude_config_unreadable") + data = loaded + + projects = data.get("projects", {}) + if not isinstance(projects, dict): + return finish("claude_config_unreadable") + project = projects.get(gate_repo, {}) + if not isinstance(project, dict): + return finish("claude_config_unreadable") + + data["hasCompletedOnboarding"] = True + data["theme"] = "dark" + project["hasTrustDialogAccepted"] = True + projects[gate_repo] = project + data["projects"] = projects + + temporary = config_dir / f".claude.json.{os.getpid()}.tmp" + fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + try: + os.fchmod(fd, 0o600) + with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as fh: + json.dump(data, fh, ensure_ascii=False, indent=2) + fh.write("\n") + fh.flush() + os.fsync(fh.fileno()) + except BaseException: + pathlib.Path(temporary).unlink(missing_ok=True) + raise + os.replace(temporary, target) + record["written"] = True + except OSError: + return finish("claude_config_unreadable") + + try: + with open(target, "r", encoding="utf-8") as fh: + back = json.load(fh) + verified = ( + isinstance(back, dict) + and back.get("hasCompletedOnboarding") is True + and back.get("theme") == "dark" + and isinstance(back.get("projects"), dict) + and isinstance(back["projects"].get(gate_repo), dict) + and back["projects"][gate_repo].get("hasTrustDialogAccepted") is True + ) + except (OSError, UnicodeError, ValueError): + verified = False + record["readBack"] = "verified" if verified else "mismatch" + if not verified: + return finish("claude_config_prewrite_unverified") + return finish(None) + + +def strip_terminal_escapes(text: str) -> str: + return ANSI_ESCAPE_RE.sub("", text) + + +def evaluate_n1_screen(text: str) -> tuple[str, str | None]: + """Classify the whole terminal output seen so far (#444 section 5.2). + + A blocking screen wins even if the ready text appears later. + Returns ("unknown", reason), ("ready", None) or ("waiting", None). + """ + plain = strip_terminal_escapes(text) + for needle, reason in N1_BLOCKING_SCREENS: + if needle in plain: + return "unknown", reason + if N1_READY_TEXT in plain: + return "ready", None + return "waiting", None + + +def command_wait_n1_ready( + args: argparse.Namespace, +) -> int: + import time + + record: dict[str, Any] = { + "schemaVersion": 1, + "cliVersion": args.cli_version, + "verifiedCliVersions": list(args.verified_cli_version), + "timeoutSeconds": args.timeout, + "verdict": "unknown", + "reason": None, + } + + def finish(state: str, reason: str | None) -> int: + record["verdict"] = "ready" if state == "ready" else "unknown" + record["reason"] = reason + write_json(args.output, record) + return 0 if state == "ready" else 2 + + version = args.cli_version.split()[0] if args.cli_version.split() else "" + if version not in args.verified_cli_version: + return finish("unknown", "ready_patterns_unverified_for_cli_version") + + log = pathlib.Path(args.log) + deadline = time.monotonic() + args.timeout + + def read_screen() -> str | None: + try: + return log.read_bytes().decode("utf-8", errors="replace") + except FileNotFoundError: + return "" + except OSError: + return None + + while True: + text = read_screen() + if text is None: + return finish("unknown", "ready_observation_failed") + state, reason = evaluate_n1_screen(text) + if state != "waiting": + return finish(state, reason) + try: + os.kill(args.pid, 0) + except ProcessLookupError: + return finish("unknown", "ready_observation_failed") + except PermissionError: + pass + if time.monotonic() >= deadline: + return finish("unknown", "ready_not_observed") + time.sleep(0.2) + + +def add_n1_input_precondition_commands(subparsers: Any) -> None: + command = subparsers.add_parser("n1-auth") + command.add_argument("--output", required=True) + command.set_defaults(handler=command_n1_auth) + + command = subparsers.add_parser("prewrite-claude-config") + command.add_argument("--config-dir", required=True) + command.add_argument("--run-root", required=True) + command.add_argument("--gate-repo", required=True) + command.add_argument("--output", required=True) + command.set_defaults(handler=command_prewrite_claude_config) + + command = subparsers.add_parser("wait-n1-ready") + command.add_argument("--log", required=True) + command.add_argument("--pid", required=True, type=int) + command.add_argument("--cli-version", required=True) + command.add_argument("--verified-cli-version", action="append", default=[]) + command.add_argument("--timeout", required=True, type=float) + command.add_argument("--output", required=True) + command.set_defaults(handler=command_wait_n1_ready) + + def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description=( @@ -4386,6 +4662,8 @@ def build_parser() -> argparse.ArgumentParser: handler=command_write_n1_result ) + add_n1_input_precondition_commands(subparsers) + return parser diff --git a/scripts/lib/pilot-pty.py b/scripts/lib/pilot-pty.py index 182be53d..c69692ae 100755 --- a/scripts/lib/pilot-pty.py +++ b/scripts/lib/pilot-pty.py @@ -25,6 +25,7 @@ from __future__ import annotations import argparse +import fcntl import os import pathlib import pty @@ -32,10 +33,18 @@ import signal import socket import subprocess +import struct import sys +import termios from typing import Sequence +# #444: the size the verifier observed the native screens at. Without it the +# pty reports 0x0 and the CLI may wrap or cut the texts the runner looks for. +TERMINAL_ROWS = 40 +TERMINAL_COLUMNS = 120 + + def spawn( argv: Sequence[str], *, @@ -50,6 +59,11 @@ def spawn( """ master, slave = pty.openpty() try: + fcntl.ioctl( + slave, + termios.TIOCSWINSZ, + struct.pack("HHHH", TERMINAL_ROWS, TERMINAL_COLUMNS, 0, 0), + ) proc = subprocess.Popen( list(argv), cwd=str(cwd), diff --git a/scripts/pilot-gate-runner.sh b/scripts/pilot-gate-runner.sh index 7a53c7eb..1a38e4b1 100755 --- a/scripts/pilot-gate-runner.sh +++ b/scripts/pilot-gate-runner.sh @@ -44,6 +44,11 @@ readonly DEFAULT_COLLECTOR_CUTOFF_SECONDS=180 readonly N1_START_TIMEOUT_SECONDS=30 readonly N1_TRANSCRIPT_TIMEOUT_SECONDS=20 readonly N1_EXIT_GRACE_SECONDS=5 +# #444 section 5: how long to wait for the native prompt input, and the CLI +# versions whose ready/blocking screen texts the verifier has observed. Add a +# version only after the verifier has measured it. +readonly N1_READY_TIMEOUT_SECONDS=30 +readonly N1_READY_VERIFIED_CLI_VERSIONS="2.1.268" SCRIPT_DIR="$( cd "$(dirname "$0")" && @@ -642,6 +647,10 @@ export_isolated_environment() { export XDG_DATA_HOME="$GATE_XDG_DATA" export XDG_STATE_HOME="$GATE_XDG_STATE" export CLAUDE_CONFIG_DIR="$GATE_CLAUDE_CONFIG" + # #444: the pilot authenticates only with CLAUDE_CODE_OAUTH_TOKEN, which is + # inherited untouched. Competing credentials would make it unclear which + # one the native CLI used. + unset ANTHROPIC_API_KEY ANTHROPIC_AUTH_TOKEN } phase_p0_environment_capture() { @@ -1086,6 +1095,55 @@ stop_native_case() { fi } +# --- #444: N1 input preconditions ------------------------------------------- + +# Merge the onboarding/trust keys into the gate's .claude.json and read them +# back. Runs before every N1 launch: the CLI rewrites the file during fresh. +prewrite_claude_config() { + local case_dir="$1" + + python3 "$ISOLATION_HELPER" prewrite-claude-config \ + --config-dir "$GATE_CLAUDE_CONFIG" \ + --run-root "$RUN_ROOT" \ + --gate-repo "$GATE_REPO" \ + --output "$case_dir/claude-config-prewrite.json" +} + +# Wait until the native screen shows the prompt input and no blocking screen. +# Sends nothing to the terminal. +wait_for_n1_ready() { + local case_dir="$1" + local pid="$2" + local version + local -a verified=() + + for version in $N1_READY_VERIFIED_CLI_VERSIONS; do + verified+=(--verified-cli-version "$version") + done + + python3 "$ISOLATION_HELPER" wait-n1-ready \ + --log "$case_dir/pty.raw" \ + --pid "$pid" \ + --cli-version "$CLAUDE_VERSION" \ + "${verified[@]}" \ + --timeout "$N1_READY_TIMEOUT_SECONDS" \ + --output "$case_dir/ready.json" +} + +# The reason recorded by a precondition helper, or the given fallback. +n1_record_reason() { + local record="$1" + local fallback="$2" + local reason + + reason="$( + python3 -c 'import json,sys; r=json.load(open(sys.argv[1])).get("reason"); print(r if isinstance(r, str) and r else "")' \ + "$record" 2>/dev/null + )" || reason="" + + printf '%s\n' "${reason:-$fallback}" +} + launch_n1_case() { local mode="$1" local expected_generation="$2" @@ -1107,11 +1165,22 @@ launch_n1_case() { CASE_TRANSCRIPT="" CASE_PROCESS_COMMAND="" CASE_STATUS="$EX_GATE_UNKNOWN" + CASE_REASON="" mkdir -p "$case_dir" || internal_error \ "cannot create N1/$mode artifact directory" + # #444 section 3: onboarding and folder trust are prewritten, never passed + # by pressing keys. Without a verified prewrite the launcher is not started. + if ! prewrite_claude_config "$case_dir"; then + CASE_REASON="$(n1_record_reason "$case_dir/claude-config-prewrite.json" claude_config_prewrite_unverified)" + log "N1/$mode: .claude.json prewrite not verified: $CASE_REASON" + printf '%s\n' "unknown" > "$case_dir/verdict" + printf '%s\n' "$CASE_REASON" > "$case_dir/reason" + return "$EX_GATE_UNKNOWN" + fi + # Start the launcher inside a pseudo terminal (#426). With a FIFO or a # file as stdin, Claude Code runs in --print mode and exits at once # ("Input must be provided either through stdin or as a prompt argument @@ -1289,6 +1358,19 @@ launch_n1_case() { esac fi + # #444 section 5: prompt only a screen that is positively ready and never + # showed a blocking screen. No key is ever sent to get past one. + if ! wait_for_n1_ready "$case_dir" "$launcher_pid"; then + CASE_REASON="$(n1_record_reason "$case_dir/ready.json" ready_observation_failed)" + log "N1/$mode: native prompt input not ready: $CASE_REASON" + printf '%s\n' "unknown" > "$case_dir/verdict" + printf '%s\n' "$CASE_REASON" > "$case_dir/reason" + CASE_STATUS="$EX_GATE_UNKNOWN" + stop_native_case "$launcher_pid" "$pty_helper_pid" + CURRENT_NATIVE_PID="" + return "$EX_GATE_UNKNOWN" + fi + marker="AGMSG_N1_TRANSCRIPT_MARKER_${RUN_ID}_${mode}_$(openssl rand -hex 16)" || { printf '%s\n' "unknown" > "$case_dir/verdict" CASE_STATUS="$EX_GATE_UNKNOWN" @@ -1361,6 +1443,24 @@ phase_n1() { mkdir -p "$ARTIFACT_DIR/N1" + # #444 section 4: without a usable CLAUDE_CODE_OAUTH_TOKEN in the + # environment neither case starts a launcher. The token is never read from + # a file and no other credential is tried. + if ! python3 "$ISOLATION_HELPER" n1-auth --output "$ARTIFACT_DIR/N1/auth.json"; then + log "N1: CLAUDE_CODE_OAUTH_TOKEN is absent; no native pilot is started" + local absent_mode + for absent_mode in fresh resume; do + mkdir -p "$ARTIFACT_DIR/N1/$absent_mode" + printf '%s\n' "unknown" > "$ARTIFACT_DIR/N1/$absent_mode/verdict" + printf '%s\n' "auth_token_absent" > "$ARTIFACT_DIR/N1/$absent_mode/reason" + done + python3 "$ISOLATION_HELPER" write-n1-result \ + --output "$ARTIFACT_DIR/N1/result.json" \ + --verdict unknown \ + --reason auth_token_absent + return "$EX_GATE_UNKNOWN" + fi + fresh_status=0 launch_n1_case "fresh" "1" || fresh_status="$?" @@ -1378,7 +1478,7 @@ phase_n1() { python3 "$ISOLATION_HELPER" write-n1-result \ --output "$ARTIFACT_DIR/N1/result.json" \ --verdict unknown \ - --reason fresh_unobservable + --reason "${CASE_REASON:-fresh_unobservable}" return "$EX_GATE_UNKNOWN" ;; esac @@ -1415,7 +1515,7 @@ phase_n1() { python3 "$ISOLATION_HELPER" write-n1-result \ --output "$ARTIFACT_DIR/N1/result.json" \ --verdict unknown \ - --reason resume_unobservable + --reason "${CASE_REASON:-resume_unobservable}" return "$EX_GATE_UNKNOWN" ;; esac diff --git a/tests/test_pilot_gate_isolation.py b/tests/test_pilot_gate_isolation.py index 08414553..d39048a2 100644 --- a/tests/test_pilot_gate_isolation.py +++ b/tests/test_pilot_gate_isolation.py @@ -7398,3 +7398,234 @@ def test_write_n1_result_handler_defensively_rejects_invalid_verdict_when_called if __name__ == "__main__": unittest.main() + + +class PilotGateIsolationN1InputPreconditionTests(unittest.TestCase): + """#444 design section 7.1 (N1P) for the isolation helper commands.""" + + def run_cli(self, *args: str, env: dict[str, str] | None = None, timeout: float = 60): + return subprocess.run( + [sys.executable, str(HELPER), *args], + stdin=subprocess.DEVNULL, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + env=env, + timeout=timeout, + check=False, + ) + + def setUp(self) -> None: + self._temp = tempfile.TemporaryDirectory() + self.root = Path(os.path.realpath(self._temp.name)) + self.run_root = self.root / "run" + self.config = self.run_root / "claude" + self.repo = str(self.run_root / "repo") + self.config.mkdir(parents=True) + self.record = self.root / "prewrite.json" + + def tearDown(self) -> None: + self._temp.cleanup() + + def prewrite(self, config: Path | None = None): + return self.run_cli( + "prewrite-claude-config", + "--config-dir", str(config or self.config), + "--run-root", str(self.run_root), + "--gate-repo", self.repo, + "--output", str(self.record), + ) + + def load(self, path: Path): + return json.loads(path.read_text(encoding="utf-8")) + + # --- prewrite ---------------------------------------------------------- + + def test_n1p_prewrite_writes_the_three_keys_0600_and_reads_them_back(self): + result = self.prewrite() + self.assertEqual(result.returncode, 0, result.stderr) + target = self.config / ".claude.json" + self.assertEqual( + self.load(target), + { + "hasCompletedOnboarding": True, + "theme": "dark", + "projects": {self.repo: {"hasTrustDialogAccepted": True}}, + }, + ) + self.assertEqual(target.stat().st_mode & 0o777, 0o600) + record = self.load(self.record) + self.assertEqual((record["verdict"], record["readBack"], record["reason"]), ("pass", "verified", None)) + self.assertEqual(sorted(p.name for p in self.config.iterdir()), [".claude.json"]) + + def test_n1p08_other_keys_and_other_projects_are_kept(self): + target = self.config / ".claude.json" + target.write_text(json.dumps({ + "numStartups": 3, + "theme": "light", + "projects": { + "/elsewhere": {"hasTrustDialogAccepted": False, "allowedTools": ["x"]}, + self.repo: {"history": [1]}, + }, + }), encoding="utf-8") + result = self.prewrite() + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual( + self.load(target), + { + "numStartups": 3, + "theme": "dark", + "hasCompletedOnboarding": True, + "projects": { + "/elsewhere": {"hasTrustDialogAccepted": False, "allowedTools": ["x"]}, + self.repo: {"history": [1], "hasTrustDialogAccepted": True}, + }, + }, + ) + + def test_n1p09_a_symlinked_config_file_is_not_written_through(self): + outside = self.root / "real-home-claude.json" + outside.write_text('{"keep": true}\n', encoding="utf-8") + (self.config / ".claude.json").symlink_to(outside) + result = self.prewrite() + self.assertEqual(result.returncode, 2) + self.assertEqual(self.load(self.record)["reason"], "claude_config_unreadable") + self.assertEqual(outside.read_text(encoding="utf-8"), '{"keep": true}\n') + self.assertTrue((self.config / ".claude.json").is_symlink()) + + def test_n1p09_invalid_json_or_non_object_is_not_overwritten(self): + target = self.config / ".claude.json" + for content in ("{not json", "[1, 2]", '{"projects": []}', '{"projects": {"%s": 1}}' % self.repo): + with self.subTest(content=content): + target.write_text(content, encoding="utf-8") + result = self.prewrite() + self.assertEqual(result.returncode, 2) + self.assertEqual(self.load(self.record)["reason"], "claude_config_unreadable") + self.assertEqual(target.read_text(encoding="utf-8"), content) + + def test_n1p10_a_config_directory_outside_run_root_is_not_written(self): + outside = self.root / "outside" + outside.mkdir() + linked = self.run_root / "linked-claude" + linked.symlink_to(outside) + for config in (outside, linked, self.run_root, Path("relative/claude")): + with self.subTest(config=str(config)): + result = self.prewrite(config) + self.assertEqual(result.returncode, 2) + self.assertEqual(self.load(self.record)["reason"], "claude_config_outside_run_root") + self.assertFalse((outside / ".claude.json").exists()) + + def test_n1p11_a_second_prewrite_restores_a_removed_trust(self): + self.assertEqual(self.prewrite().returncode, 0) + target = self.config / ".claude.json" + data = self.load(target) + del data["projects"][self.repo]["hasTrustDialogAccepted"] + data["hasCompletedOnboarding"] = False + target.write_text(json.dumps(data), encoding="utf-8") + result = self.prewrite() + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(self.load(self.record)["readBack"], "verified") + data = self.load(target) + self.assertIs(data["projects"][self.repo]["hasTrustDialogAccepted"], True) + self.assertIs(data["hasCompletedOnboarding"], True) + + # --- readiness --------------------------------------------------------- + + def wait_ready(self, screen: bytes, *, version: str = "2.1.268 (Claude Code)", timeout: float = 1.5, pid: int | None = None): + log = self.root / "pty.raw" + log.write_bytes(screen) + output = self.root / "ready.json" + result = self.run_cli( + "wait-n1-ready", + "--log", str(log), + "--pid", str(pid or os.getpid()), + "--cli-version", version, + "--verified-cli-version", "2.1.268", + "--timeout", str(timeout), + "--output", str(output), + ) + return result.returncode, self.load(output)["reason"] + + def test_n1p_ready_text_behind_terminal_escapes_is_ready(self): + screen = b"\x1b]0;claude\x07\x1b[2m\xe2\x8f\xb8 manual mode on \xc2\xb7 ? for\x1b[0m shortcuts" + # The ready text is split by an escape: it only matches once stripped. + self.assertNotIn(b"? for shortcuts", screen) + self.assertEqual(self.wait_ready(screen), (0, None)) + + def test_n1p04_n1p05_a_blocking_screen_wins_over_later_ready_text(self): + for text, reason in ( + (b"Select login method:\r\n", "login_method_screen"), + (b"Quick safety check: Is this a project you created or one you trust?\r\n", "trust_dialog_screen"), + ): + with self.subTest(reason=reason): + self.assertEqual(self.wait_ready(text + b"\x1b[2J? for shortcuts"), (2, reason)) + + def test_n1p06_nothing_on_screen_times_out_as_not_observed(self): + self.assertEqual(self.wait_ready(b"Welcome\r\n", timeout=0.6), (2, "ready_not_observed")) + + def test_n1p07_an_unverified_cli_version_is_not_judged(self): + for version in ("2.1.269 (Claude Code)", "", "2.1.2680"): + with self.subTest(version=version): + self.assertEqual( + self.wait_ready(b"? for shortcuts", version=version), + (2, "ready_patterns_unverified_for_cli_version"), + ) + + def test_n1p_an_exited_child_is_an_observation_failure(self): + child = subprocess.Popen([sys.executable, "-c", "pass"]) + child.wait() + self.assertEqual(self.wait_ready(b"loading", pid=child.pid, timeout=5), (2, "ready_observation_failed")) + + # --- auth and environment --------------------------------------------- + + def test_n1p02_n1p03_token_presence_only(self): + base = {k: v for k, v in os.environ.items() if k not in ISOLATION.CLAUDE_AUTH_ENV_KEYS} + output = self.root / "auth.json" + token = "qqzz-synthetic-" + os.urandom(8).hex() + for value, expected, rc in ( + (None, "absent", 2), + ("", "absent", 2), + ("qqzz\x1bsynthetic", "absent", 2), + (token, "present", 0), + ): + with self.subTest(value=value): + env = dict(base) + if value is not None: + env["CLAUDE_CODE_OAUTH_TOKEN"] = value + result = self.run_cli("n1-auth", "--output", str(output), env=env) + self.assertEqual(result.returncode, rc, result.stderr) + text = output.read_text(encoding="utf-8") + self.assertEqual( + json.loads(text), + { + "schemaVersion": 1, + "oauthTokenEnv": expected, + "unsetCompetingEnv": ["ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN"], + }, + ) + self.assertNotIn(token, text + result.stdout + result.stderr) + + def test_n1p14_capture_environment_records_no_credential_value(self): + values = {key: f"qqzz-synthetic-{key}-{os.urandom(6).hex()}" for key in ISOLATION.CLAUDE_AUTH_ENV_KEYS} + env = {**os.environ, **values} + output = self.root / "environment.json" + result = self.run_cli( + "capture-environment", + "--output", str(output), "--run-id", "run", "--source", str(self.root), + "--source-head", "0" * 40, "--live-skill-dir", str(self.root), + "--artifact-dir", str(self.root), "--run-root", str(self.run_root), + "--gate-team", "team", "--claude-bin", "claude", "--claude-resolved", "claude", + "--claude-version", "2.1.268", "--claude-digest", "0" * 64, + "--collector-cutoff-seconds", "180", + env=env, + ) + self.assertEqual(result.returncode, 0, result.stderr) + text = output.read_text(encoding="utf-8") + for value in values.values(): + self.assertNotIn(value, text + result.stdout + result.stderr) + # Positive control: presence is recorded for all three names. + self.assertEqual( + json.loads(text)["claudeAuthEnvironmentPresent"], + {key: True for key in ISOLATION.CLAUDE_AUTH_ENV_KEYS}, + ) diff --git a/tests/test_pilot_gate_runner.bats b/tests/test_pilot_gate_runner.bats index dbd3b40c..e8be2b94 100644 --- a/tests/test_pilot_gate_runner.bats +++ b/tests/test_pilot_gate_runner.bats @@ -1388,6 +1388,8 @@ EOF PTY_HELPER="$SCRIPTS/lib/pilot-pty.py" export_isolated_environment() { :; } + # #444 preconditions are covered by their own tests below. + prewrite_claude_config() { :; } # Observe the launcher 8 seconds after start instead of a real binding. wait_for_binding() { sleep 8 @@ -1439,6 +1441,7 @@ EOF chmod +x "$DUMMY_HELPERS/pilot-pty.py" PTY_HELPER="$DUMMY_HELPERS/pilot-pty.py" export_isolated_environment() { :; } + prewrite_claude_config() { :; } wait_for_binding() { printf '%s\n' called >> "$CALL_LOG"; return 1; } local case_status=0 @@ -1464,6 +1467,9 @@ prepare_n1_marker_case() { printf '%s\n' 'raise SystemExit(0)' > "$ISOLATION_HELPER" export_isolated_environment() { :; } + # #444 preconditions are covered by their own tests below. + prewrite_claude_config() { :; } + wait_for_n1_ready() { :; } wait_for_binding() { sleep 1; printf '%s\n' "$TEST_ROOT/binding.json"; } json_field() { printf '%s\n' "123e4567-e89b-42d3-a456-426614174000"; } record_process_command() { printf '%s\n' "claude" > "$2"; } @@ -1520,3 +1526,253 @@ EOF [ "$(wc -l < "$TEST_ROOT/send-calls" | tr -d ' ')" = "1" ] [ ! -e "$TEST_ROOT/transcript-calls" ] } + +# --- #444: N1 input preconditions ------------------------------------------- + +# A gate run layout with canonical paths, the real isolation helper for the +# #444 commands (every other isolation check accepts), a real pty helper that +# counts prompt sends, and a stand-in launcher whose screen is chosen per test. +prepare_n1_precondition_case() { + local screen="$1" + local root + + root="$(cd "$TEST_ROOT" && pwd -P)" + RUN_ROOT="$root/run" + GATE_REPO="$RUN_ROOT/repo" + GATE_CLAUDE_CONFIG="$RUN_ROOT/claude" + GATE_HOME="$RUN_ROOT/home" + GATE_XDG_CONFIG="$RUN_ROOT/xdg/config" + GATE_XDG_CACHE="$RUN_ROOT/xdg/cache" + GATE_XDG_DATA="$RUN_ROOT/xdg/data" + GATE_XDG_STATE="$RUN_ROOT/xdg/state" + ARTIFACT_DIR="$RUN_ROOT/artifacts" + GATE_TEAM="gate-team" + RUN_ID="run444" + CLAUDE_VERSION="2.1.268 (Claude Code)" + mkdir -p "$GATE_REPO/scripts" "$GATE_CLAUDE_CONFIG" "$GATE_HOME" "$ARTIFACT_DIR" + printf '%s\n' "$screen" > "$GATE_REPO/screen" + printf '%s\n' '{}' > "$TEST_ROOT/binding.json" + + N1P_TOKEN="qqzz-synthetic-$(openssl rand -hex 12)" + export CLAUDE_CODE_OAUTH_TOKEN="$N1P_TOKEN" + + local real_isolation="$SCRIPTS/lib/pilot-gate-isolation.py" + ISOLATION_HELPER="$DUMMY_HELPERS/isolation-444.py" + cat > "$ISOLATION_HELPER" < "$PTY_HELPER" < "$GATE_REPO/scripts/pilot-launcher.sh" <<'EOF' +#!/usr/bin/env bash +repo="$(cd "$(dirname "$0")/.." && pwd)" +printf '%s\n' started >> "$repo/launcher-starts" +stty size > "$repo/launcher-tty-size" 2>/dev/null +[ -n "${ANTHROPIC_API_KEY+set}${ANTHROPIC_AUTH_TOKEN+set}" ] && + printf '%s\n' competing > "$repo/launcher-competing-auth" +[ -n "${CLAUDE_CODE_OAUTH_TOKEN:-}" ] && + printf '%s\n' present > "$repo/launcher-oauth" +stty raw -echo 2>/dev/null +case "$(cat "$repo/screen")" in + ready) + printf '\033[2m\342\217\270 manual mode on \302\267 ? for shortcuts\033[0m\r\n' ;; + login) + printf 'Select login method:\r\n'; sleep 1; printf '? for shortcuts\r\n' ;; + trust) + printf 'Quick safety check: Is this a project you created or one you trust?\r\n' + sleep 1; printf '? for shortcuts\r\n' ;; + strip-trust) + python3 - "$CLAUDE_CONFIG_DIR/.claude.json" <<'PY' +import json, sys +path = sys.argv[1] +data = json.load(open(path)) +for project in data.get("projects", {}).values(): + project.pop("hasTrustDialogAccepted", None) +json.dump(data, open(path, "w")) +PY + printf '? for shortcuts\r\n' ;; + none) ;; +esac +# Record every byte this terminal receives (prompts and any keystroke). +exec python3 -c ' +import os, sys, time +end = time.time() + 60 +with open(sys.argv[1], "ab") as out: + while time.time() < end: + data = os.read(0, 4096) + if not data: + break + out.write(data); out.flush() +' "$repo/terminal-input" +EOF + chmod +x "$GATE_REPO/scripts/pilot-launcher.sh" + + wait_for_binding() { sleep 1; printf '%s\n' "$TEST_ROOT/binding.json"; } + json_field() { printf '%s\n' "123e4567-e89b-42d3-a456-426614174000"; } + record_process_command() { printf '%s\n' "claude" > "$2"; } + wait_for_transcript() { printf '%s\n' "$2" >> "$TEST_ROOT/transcript-calls"; return 1; } +} + +n1p_count() { + if [ -f "$1" ]; then wc -l < "$1" | tr -d ' '; else printf '0\n'; fi +} + +@test "#444 N1P-01/13/15: a ready screen is prompted once, in a 120x40 terminal, without competing credentials" { + prepare_n1_precondition_case ready + export ANTHROPIC_API_KEY="qqzz-competing-$(openssl rand -hex 6)" + + local case_status=0 + launch_n1_case fresh 1 || case_status="$?" + + local case_dir="$ARTIFACT_DIR/N1/fresh" + [ "$(n1p_count "$TEST_ROOT/send-calls")" = "1" ] || + { cat "$case_dir/ready.json" "$case_dir/stderr.raw" >&2; return 1; } + [ "$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["verdict"])' "$case_dir/ready.json")" = "ready" ] + grep -qF "AGMSG_N1_TRANSCRIPT_MARKER_run444_fresh_" "$GATE_REPO/terminal-input" || + { echo "prompt did not reach the terminal" >&2; return 1; } + [ "$(cat "$GATE_REPO/launcher-tty-size")" = "40 120" ] || + { echo "terminal size: $(cat "$GATE_REPO/launcher-tty-size")" >&2; return 1; } + [ "$(cat "$GATE_REPO/launcher-oauth")" = "present" ] + [ ! -e "$GATE_REPO/launcher-competing-auth" ] || + { echo "a competing credential reached the pilot" >&2; return 1; } + # N1P-12 on the launched path: the token reached the pilot only through its + # environment, so no file in the run tree holds it. Positive control below. + [ -z "$(grep -rlF -e "$N1P_TOKEN" -e "$ANTHROPIC_API_KEY" "$RUN_ROOT" "$TEST_ROOT/send-calls" 2>/dev/null)" ] || + { echo "credential value written to the run tree" >&2; return 1; } + printf '%s\n' "$N1P_TOKEN" > "$ARTIFACT_DIR/positive-control" + [ "$(grep -rlF -e "$N1P_TOKEN" "$RUN_ROOT" | wc -l | tr -d ' ')" = "1" ] + # No transcript is observed in this fixture: unknown, not pass. + [ "$case_status" -eq "$EX_GATE_UNKNOWN" ] +} + +@test "#444 N1P-02/03/12/13: without a usable token no launcher starts, and only presence and names are recorded" { + prepare_n1_precondition_case ready + launch_n1_case() { printf '%s\n' "$1" >> "$TEST_ROOT/launch-calls"; return 0; } + export ANTHROPIC_API_KEY="qqzz-competing-$(openssl rand -hex 6)" + local competing="$ANTHROPIC_API_KEY" + + local token_state + for token_state in unset empty control; do + rm -rf "$ARTIFACT_DIR/N1" + case "$token_state" in + unset) unset CLAUDE_CODE_OAUTH_TOKEN ;; + empty) export CLAUDE_CODE_OAUTH_TOKEN="" ;; + control) export CLAUDE_CODE_OAUTH_TOKEN="$(printf 'qqzz-synthetic\nline')" ;; + esac + + local n1_status=0 + phase_n1 || n1_status="$?" + + [ "$n1_status" -eq "$EX_GATE_UNKNOWN" ] || + { echo "$token_state: status $n1_status" >&2; return 1; } + [ "$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1]))["reason"])' "$ARTIFACT_DIR/N1/result.json")" = "auth_token_absent" ] + [ "$(cat "$ARTIFACT_DIR/N1/fresh/reason")" = "auth_token_absent" ] + [ "$(cat "$ARTIFACT_DIR/N1/resume/reason")" = "auth_token_absent" ] + [ "$(python3 -c 'import json,sys; r=json.load(open(sys.argv[1])); print(r["oauthTokenEnv"], ",".join(r["unsetCompetingEnv"]), sorted(r))' "$ARTIFACT_DIR/N1/auth.json")" = "absent ANTHROPIC_API_KEY,ANTHROPIC_AUTH_TOKEN ['oauthTokenEnv', 'schemaVersion', 'unsetCompetingEnv']" ] || + { cat "$ARTIFACT_DIR/N1/auth.json" >&2; return 1; } + done + [ "$(n1p_count "$TEST_ROOT/launch-calls")" = "0" ] + [ "$(n1p_count "$TEST_ROOT/send-calls")" = "0" ] + + # N1P-12: no credential value anywhere in the run tree. Positive control: + # the same search finds a file that does contain the value. + [ -z "$(grep -rlF -e "$N1P_TOKEN" -e "$competing" "$RUN_ROOT" "$TEST_ROOT/send-calls" 2>/dev/null)" ] || + { echo "credential value written to the run tree" >&2; return 1; } + printf '%s\n' "$N1P_TOKEN" > "$RUN_ROOT/positive-control" + [ "$(grep -rlF -e "$N1P_TOKEN" "$RUN_ROOT" | wc -l | tr -d ' ')" = "1" ] +} + +@test "#444 N1P-04: a login screen is never prompted or keyed through, even when ready text follows" { + prepare_n1_precondition_case login + + local case_status=0 + launch_n1_case fresh 1 || case_status="$?" + + local case_dir="$ARTIFACT_DIR/N1/fresh" + [ "$case_status" -eq "$EX_GATE_UNKNOWN" ] + [ "$(cat "$case_dir/reason")" = "login_method_screen" ] || + { cat "$case_dir/ready.json" >&2; return 1; } + [ "$(n1p_count "$TEST_ROOT/send-calls")" = "0" ] + [ ! -s "$GATE_REPO/terminal-input" ] || + { echo "bytes were sent to the terminal" >&2; return 1; } +} + +@test "#444 N1P-05: a trust dialog is never prompted or keyed through, even when ready text follows" { + prepare_n1_precondition_case trust + + local case_status=0 + launch_n1_case fresh 1 || case_status="$?" + + local case_dir="$ARTIFACT_DIR/N1/fresh" + [ "$case_status" -eq "$EX_GATE_UNKNOWN" ] + [ "$(cat "$case_dir/reason")" = "trust_dialog_screen" ] || + { cat "$case_dir/ready.json" >&2; return 1; } + [ "$(n1p_count "$TEST_ROOT/send-calls")" = "0" ] + [ ! -s "$GATE_REPO/terminal-input" ] +} + +@test "#444 N1P-07: an unverified CLI version is not judged ready and not prompted" { + prepare_n1_precondition_case ready + CLAUDE_VERSION="9.9.9 (Claude Code)" + + local case_status=0 + launch_n1_case fresh 1 || case_status="$?" + + [ "$case_status" -eq "$EX_GATE_UNKNOWN" ] + [ "$(cat "$ARTIFACT_DIR/N1/fresh/reason")" = "ready_patterns_unverified_for_cli_version" ] + [ "$(n1p_count "$TEST_ROOT/send-calls")" = "0" ] +} + +@test "#444 N1P-10: a claude config directory outside RUN_ROOT is not written and starts no launcher" { + prepare_n1_precondition_case ready + local outside + outside="$(cd "$TEST_ROOT" && pwd -P)/outside-claude" + mkdir -p "$outside" + GATE_CLAUDE_CONFIG="$outside" + + local case_status=0 + launch_n1_case fresh 1 || case_status="$?" + + [ "$case_status" -eq "$EX_GATE_UNKNOWN" ] + [ "$(cat "$ARTIFACT_DIR/N1/fresh/reason")" = "claude_config_outside_run_root" ] + [ ! -e "$outside/.claude.json" ] + [ "$(n1p_count "$GATE_REPO/launcher-starts")" = "0" ] +} + +@test "#444 N1P-11: resume rewrites the trust the fresh session removed and reads it back" { + prepare_n1_precondition_case strip-trust + + local fresh_status=0 + launch_n1_case fresh 1 || fresh_status="$?" + python3 -c ' +import json, sys +data = json.load(open(sys.argv[1])) +assert "hasTrustDialogAccepted" not in data["projects"][sys.argv[2]], data +' "$GATE_CLAUDE_CONFIG/.claude.json" "$GATE_REPO" || + { echo "the fixture did not remove the trust" >&2; return 1; } + + printf '%s\n' ready > "$GATE_REPO/screen" + local resume_status=0 + launch_n1_case resume 2 "123e4567-e89b-42d3-a456-426614174000" || resume_status="$?" + + local record="$ARTIFACT_DIR/N1/resume/claude-config-prewrite.json" + [ "$(python3 -c 'import json,sys; r=json.load(open(sys.argv[1])); print(r["verdict"], r["readBack"])' "$record")" = "pass verified" ] || + { cat "$record" >&2; return 1; } + python3 -c ' +import json, sys +data = json.load(open(sys.argv[1])) +assert data["projects"][sys.argv[2]]["hasTrustDialogAccepted"] is True, data +' "$GATE_CLAUDE_CONFIG/.claude.json" "$GATE_REPO" +} diff --git a/tests/test_pilot_pty.py b/tests/test_pilot_pty.py index 3f5172de..3e06ec7a 100644 --- a/tests/test_pilot_pty.py +++ b/tests/test_pilot_pty.py @@ -95,6 +95,17 @@ def test_child_has_a_terminal_on_all_three_streams_and_its_own_session(self) -> self.assertEqual(proc.returncode, 0) self.assertIn("tty:111 sid:1", self.log.read_text(errors="replace")) + def test_n1p15_the_child_terminal_is_120_columns_by_40_rows(self) -> None: + # #444: the size the native screens were observed at. + report = [sys.executable, "-c", "import os; s = os.get_terminal_size(0); print('size=%dx%d' % (s.columns, s.lines))"] + proc, master = PTY.spawn(report, cwd=self.tmp, env=dict(os.environ)) + try: + self.drain(master, proc) + proc.wait(timeout=10) + finally: + os.close(master) + self.assertIn("size=120x40", self.log.read_text(errors="replace")) + def test_without_a_terminal_the_stand_in_takes_the_print_path(self) -> None: # Control for the stand-in itself: the FIFO/pipe start of the old # runner gives exactly the #426 failure. From c9312aad2004e98253137a7bc14af4fe590246e8 Mon Sep 17 00:00:00 2001 From: kappaseijin4codex Date: Sun, 13 Sep 2026 16:06:48 +0900 Subject: [PATCH 2/2] test(g4): make the N1P-04/05 blocking-screen control deterministic (#444) The stand-in launcher printed the blocking screen, slept one second, then printed the ready text. A readiness poll inside that second saw only the blocking screen, so a mutant that checks the ready text first was caught only when the poll came later: the reviewer measured 5/10 and 6/10. Both texts now arrive in one write. The priority mutant is killed 10/10 for each test and the unmutated code passes 10/10. Refs #444 Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01Gd6MbRAYGpkDzHAk1XZroZ --- tests/test_pilot_gate_runner.bats | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/tests/test_pilot_gate_runner.bats b/tests/test_pilot_gate_runner.bats index e8be2b94..e5013cca 100644 --- a/tests/test_pilot_gate_runner.bats +++ b/tests/test_pilot_gate_runner.bats @@ -1588,11 +1588,13 @@ stty raw -echo 2>/dev/null case "$(cat "$repo/screen")" in ready) printf '\033[2m\342\217\270 manual mode on \302\267 ? for shortcuts\033[0m\r\n' ;; + # The blocking screen and the ready text arrive in one write, so the + # first observation already holds both: only the blocking-first priority + # can make this unknown, whatever the polling timing. login) - printf 'Select login method:\r\n'; sleep 1; printf '? for shortcuts\r\n' ;; + printf 'Select login method:\r\n? for shortcuts\r\n' ;; trust) - printf 'Quick safety check: Is this a project you created or one you trust?\r\n' - sleep 1; printf '? for shortcuts\r\n' ;; + printf 'Quick safety check: Is this a project you created or one you trust?\r\n? for shortcuts\r\n' ;; strip-trust) python3 - "$CLAUDE_CONFIG_DIR/.claude.json" <<'PY' import json, sys @@ -1694,7 +1696,7 @@ n1p_count() { [ "$(grep -rlF -e "$N1P_TOKEN" "$RUN_ROOT" | wc -l | tr -d ' ')" = "1" ] } -@test "#444 N1P-04: a login screen is never prompted or keyed through, even when ready text follows" { +@test "#444 N1P-04: a login screen shown with the ready text is never prompted or keyed through" { prepare_n1_precondition_case login local case_status=0 @@ -1709,7 +1711,7 @@ n1p_count() { { echo "bytes were sent to the terminal" >&2; return 1; } } -@test "#444 N1P-05: a trust dialog is never prompted or keyed through, even when ready text follows" { +@test "#444 N1P-05: a trust dialog shown with the ready text is never prompted or keyed through" { prepare_n1_precondition_case trust local case_status=0