openworker/tests/test_run_cli.py

"""`openworker run` as a person uses it: this computer's settings are read and never
changed, the session is saved under the state folder by workspace path, approvals and
questions are asked in the terminal, and the exit code says whether the task finished.

The sealed benchmark run (`--isolated`) is covered in test_headless_run.py."""

from __future__ import annotations

import asyncio
import json
import os
import select
import subprocess
import sys
import time
from pathlib import Path

import pytest

from coworker.engine import ApprovalOutcome, PermissionRequest
from coworker.headless.runner import allowed_sites, parse_args, workspace_folder_name
from coworker.headless.terminal import Terminal

ROOT = Path(__file__).resolve().parents[1]

WRITE_THEN_DONE = [
    {"text": "I will create the file.", "tool_calls": [{"name": "write_file", "arguments": {"path": "hello.txt", "content": "hello"}}]},
    {"text": "The file is written. Done."},
]


def _env(machine: Path) -> dict:
    env = dict(os.environ)
    env["PYTHONPATH"] = str(ROOT) + os.pathsep + env.get("PYTHONPATH", "")
    for name in ("ANTHROPIC_API_KEY", "OPENAI_API_KEY", "COWORKER_SCRATCH_BASE", "OPENWORKER_BASE_DIR"):
        env.pop(name, None)
    env["COWORKER_STATE_DIR"] = str(machine)  # "this computer's" state folder
    return env


def _command(tmp_path: Path, script: list[dict], *extra: str) -> tuple[list[str], Path]:
    ws = tmp_path / "ws"
    ws.mkdir(exist_ok=True)
    path = tmp_path / "script.json"
    path.write_text(json.dumps(script), encoding="utf-8")
    return [
        sys.executable, "-m", "coworker.cli", "run", "--prompt", "Create hello.txt.",
        "--workspace", str(ws), "--scripted", str(path), "--timeout-seconds", "120", *extra,
    ], ws


def _run(tmp_path: Path, script: list[dict], *extra: str, machine: Path | None = None):
    machine = machine or tmp_path / "machine"
    machine.mkdir(exist_ok=True)
    cmd, ws = _command(tmp_path, script, *extra)
    proc = subprocess.run(cmd, env=_env(machine), capture_output=True, text=True, timeout=300)
    return proc, ws, machine


def _session_folders(machine: Path, ws: Path) -> list[Path]:
    base = machine / "sessions" / workspace_folder_name(ws.resolve())
    return sorted(p for p in base.iterdir()) if base.is_dir() else []


# -- the saved session ------------------------------------------------------------------


def test_a_run_is_saved_under_the_state_folder_by_workspace_path(tmp_path: Path) -> None:
    proc, ws, machine = _run(tmp_path, WRITE_THEN_DONE, "--model", "anthropic/claude-sonnet-5", "--approval-mode", "bypass-approvals")
    assert proc.returncode == 0, proc.stderr
    assert (ws / "hello.txt").read_text(encoding="utf-8") == "hello"
    (folder,) = _session_folders(machine, ws)
    summary = json.loads((folder / "summary.json").read_text(encoding="utf-8"))
    assert summary["outcome"] == "completed" and summary["args"]["isolated"] is False
    assert folder.name == summary["run_id"]
    for name in ("events.jsonl", "messages.json", "trajectory.json", "audit.db"):
        assert (folder / name).exists(), name
    # Both exist from the start: a sandbox is handed the run's folders when it is made.
    assert (folder / "scratch").is_dir() and (folder / "tool-output").is_dir()
    # Standard output is the coworker's answer and nothing else.
    assert proc.stdout == "The file is written. Done.\n"
    assert "outcome=completed" in proc.stderr and "write_file: hello.txt" in proc.stderr


def test_the_workspace_folder_name_is_the_path_with_dashes() -> None:
    assert workspace_folder_name(Path("/Users/me/src/my app")) == "-Users-me-src-my-app"
    long = workspace_folder_name(Path("/" + "/".join(["folder"] * 60)))
    assert len(long) < 200 and long != workspace_folder_name(Path("/" + "/".join(["folder"] * 61)))


def test_this_computers_settings_are_read_and_not_changed(tmp_path: Path) -> None:
    machine = tmp_path / "machine"
    machine.mkdir()
    (machine / "config.toml").write_text('model = "anthropic:claude-sonnet-5"\n', encoding="utf-8")
    before = {p.name: p.read_bytes() for p in machine.iterdir() if p.is_file()}

    proc, ws, _ = _run(tmp_path, WRITE_THEN_DONE, "--approval-mode", "bypass-approvals", machine=machine)  # no --model
    assert proc.returncode == 0, proc.stderr
    (folder,) = _session_folders(machine, ws)
    assert json.loads((folder / "summary.json").read_text(encoding="utf-8"))["args"]["model"] == "anthropic:claude-sonnet-5"
    assert {p.name: p.read_bytes() for p in machine.iterdir() if p.is_file()} == before
    # Beside the settings, the run left only its own session.
    assert sorted(p.name for p in machine.iterdir() if p.is_dir()) == ["sessions"]


def test_a_run_has_the_connectors_this_computer_has_connected(tmp_path: Path) -> None:
    """The write is refused here (mode ask, nobody to ask), which is the proof the
    connector's tool was there to be called; nothing leaves the computer."""
    machine = tmp_path / "machine"
    machine.mkdir()
    (machine / "secrets.json").write_text(json.dumps({"github:default": {"token": "ghp_test", "enabled": True}}), encoding="utf-8")
    script = [
        {"text": "", "tool_calls": [{"name": "github_create_issue", "arguments": {"owner": "acme", "repo": "app", "title": "A title", "body": "A body"}}]},
        {"text": "It was refused."},
    ]
    proc, ws, _ = _run(tmp_path, script, "--model", "anthropic/claude-sonnet-5", machine=machine)
    assert proc.returncode == 0, proc.stderr
    (folder,) = _session_folders(machine, ws)
    answers = json.loads((folder / "answers.json").read_text(encoding="utf-8"))
    assert [row["tool"] for row in answers["rows"]] == ["github_create_issue"]
    assert json.loads((machine / "secrets.json").read_text(encoding="utf-8")) == {"github:default": {"token": "ghp_test", "enabled": True}}


def test_two_runs_in_one_workspace_are_two_sessions(tmp_path: Path) -> None:
    for _ in range(2):
        proc, ws, machine = _run(tmp_path, WRITE_THEN_DONE, "--model", "anthropic/claude-sonnet-5", "--approval-mode", "bypass-approvals")
        assert proc.returncode == 0, proc.stderr
    assert len(_session_folders(machine, ws)) == 2


# -- nobody at a terminal ---------------------------------------------------------------


def test_the_default_mode_asks_and_with_no_terminal_the_card_is_refused(tmp_path: Path) -> None:
    proc, ws, machine = _run(tmp_path, WRITE_THEN_DONE, "--model", "anthropic/claude-sonnet-5")
    assert proc.returncode == 0, proc.stderr  # the model's turn finished; the write was refused
    assert not (ws / "hello.txt").exists()
    assert "no terminal is attached" in proc.stderr and "--approval-mode auto-approve" in proc.stderr
    (folder,) = _session_folders(machine, ws)
    summary = json.loads((folder / "summary.json").read_text(encoding="utf-8"))
    assert summary["args"]["approval_mode"] == "ask" and summary["args"]["auto_answer"] is True
    assert summary["answers"]["cards_refused"] == 1


def test_with_no_terminal_a_question_is_answered_by_rule(tmp_path: Path) -> None:
    script = [
        {"text": "", "tool_calls": [{"name": "ask_user", "arguments": {"question": "Which colour?", "options": ["red", "blue"]}}]},
        {"text": "Going with the safe one."},
    ]
    proc, ws, machine = _run(tmp_path, script, "--model", "anthropic/claude-sonnet-5", "--approval-mode", "bypass-approvals")
    assert proc.returncode == 0, proc.stderr
    (folder,) = _session_folders(machine, ws)
    assert json.loads((folder / "summary.json").read_text(encoding="utf-8"))["answers"]["questions_answered"] == 1


def test_a_run_that_stops_early_exits_3_and_says_why(tmp_path: Path) -> None:
    fail = {"error": "Invalid API key provided", "error_type": "AuthenticationError"}
    proc, ws, machine = _run(tmp_path, [fail], "--model", "anthropic/claude-sonnet-5", "--approval-mode", "bypass-approvals")
    assert proc.returncode == 3
    assert "outcome=model_error" in proc.stderr and "Invalid API key provided" in proc.stderr
    assert proc.stdout == ""  # no answer was given
    (folder,) = _session_folders(machine, ws)
    assert (folder / "trajectory.json").exists()  # the record is written all the same


# -- usage errors -----------------------------------------------------------------------


@pytest.mark.parametrize(
    "extra, said",
    [
        (["--isolated", "--model", "anthropic/claude-sonnet-5"], "--isolated needs --out"),
        (["--isolated", "--out", "OUT"], "--isolated needs --model"),
        (["--model", "anthropic/claude-sonnet-5", "--allow-site", "*.example.com"], "not a site name"),
        (["--model", "anthropic/claude-sonnet-5", "--allow-site", "10.0.0.1"], "not a site name"),
        (["--model", "anthropic/claude-sonnet-5", "--allow-sites-file", "missing.txt"], "cannot read --allow-sites-file"),
    ],
)
def test_a_command_line_that_cannot_run_stops_with_exit_code_2(tmp_path: Path, extra: list[str], said: str) -> None:
    extra = [str(tmp_path / "out") if x == "OUT" else x for x in extra]
    proc, ws, machine = _run(tmp_path, WRITE_THEN_DONE, *extra)
    assert proc.returncode == 2
    assert said in proc.stderr and proc.stdout == ""
    assert not (machine / "sessions").exists()  # nothing was started, nothing is left behind


def test_allowed_sites_come_from_the_flag_and_the_file(tmp_path: Path) -> None:
    listed = tmp_path / "sites.txt"
    listed.write_text("# package registries\nregistry.npmjs.org\n\nfiles.pythonhosted.org:443  # wheels\ninternal.example.com:8443\n", encoding="utf-8")
    args = parse_args(["--prompt", "x", "--allow-site", "example.com", "--allow-site", "registry.npmjs.org", "--allow-sites-file", str(listed)])
    assert allowed_sites(args) == [
        "example.com:443", "registry.npmjs.org:443", "files.pythonhosted.org:443", "internal.example.com:8443",
    ]
    assert allowed_sites(parse_args(["--prompt", "x"])) == []


def test_allow_site_says_so_when_no_list_holds_the_run(tmp_path: Path) -> None:
    proc, _ws, _machine = _run(
        tmp_path, WRITE_THEN_DONE, "--model", "anthropic/claude-sonnet-5", "--approval-mode", "bypass-approvals", "--allow-site", "example.com",
    )
    assert proc.returncode == 0, proc.stderr
    assert "--allow-site has no effect here" in proc.stderr


# -- the prompts ------------------------------------------------------------------------


class _Keys:
    """A terminal for the prompts: the lines the person types, and what was printed."""

    def __init__(self, *typed: str) -> None:
        self.typed = list(typed)
        self.printed = ""

    def terminal(self, cwd: Path | None = None) -> Terminal:
        return Terminal(lambda: self.typed.pop(0) if self.typed else None, self._write, cwd=cwd)

    def _write(self, text: str) -> None:
        self.printed += text


def _request(tool: str, arguments: dict, reason: str = "requires approval") -> PermissionRequest:
    return PermissionRequest(tool_name=tool, arguments=arguments, metadata=None, reason=reason)


def test_a_command_card_offers_once_this_session_and_no() -> None:
    home = Path.home()
    for typed, outcome in (("y", ApprovalOutcome.ONCE), ("s", ApprovalOutcome.ALWAYS_COMMAND), ("n", ApprovalOutcome.DENY)):
        keys = _Keys(typed)
        got = asyncio.run(keys.terminal(cwd=home / "src" / "project").approve(_request("run_shell", {"command": "npm install"})))
        assert got is outcome
        assert "Approval needed: run a command\n  npm install\n  in ~/src/project\n" in keys.printed
        assert "[y] yes, once   [s] yes, for this session   [n] no" in keys.printed
        assert "requires approval" not in keys.printed  # the engine's filler reason is not shown


def test_an_unknown_key_asks_again_and_a_closed_terminal_is_a_no() -> None:
    keys = _Keys("maybe", "", "Y")
    assert asyncio.run(keys.terminal().approve(_request("run_shell", {"command": "ls"}))) is ApprovalOutcome.ONCE
    assert keys.printed.count("Type y, s or n.") == 2

    assert asyncio.run(_Keys().terminal().approve(_request("run_shell", {"command": "ls"}))) is ApprovalOutcome.DENY


def test_a_file_write_card_shows_its_arguments_and_the_reason() -> None:
    keys = _Keys("s")
    request = _request("write_file", {"path": "notes.md", "content": "x" * 900}, reason="writes outside the workspace")
    assert asyncio.run(keys.terminal().approve(request)) is ApprovalOutcome.ALWAYS_TOOL
    assert "Approval needed: use write_file" in keys.printed and "path: notes.md" in keys.printed
    assert "Why it asks: writes outside the workspace" in keys.printed
    assert "x" * 300 not in keys.printed  # a long value is cut


def test_the_network_card_offers_this_run_only() -> None:
    keys = _Keys("y")
    term = keys.terminal()
    term.card = {
        "network_request": {
            "reason": "npm install needs the package registry.",
            "evidence": True,
            "hosts": [{"host": "registry.npmjs.org:443", "blocked_seconds_ago": 4}],
        }
    }
    request = _request("request_network_access", {"hosts": ["registry.npmjs.org"], "reason": "…"})
    assert asyncio.run(term.approve(request)) is ApprovalOutcome.ALWAYS_DOMAIN  # this run; never "always"
    assert "Approval needed: let commands reach registry.npmjs.org\n" in keys.printed
    assert "Reason: npm install needs the package registry." in keys.printed
    assert "The sandbox blocked this site 4 seconds ago." in keys.printed
    assert "[s] yes, for this run   [n] no" in keys.printed and "once" not in keys.printed

    keys = _Keys("n")
    term = keys.terminal()
    term.card = {"network_request": {"reason": "", "evidence": True, "hosts": [{"host": "example.org:8443", "blocked_seconds_ago": None}]}}
    assert asyncio.run(term.approve(request)) is ApprovalOutcome.DENY
    assert "let commands reach example.org:8443" in keys.printed and "No command has tried this site yet." in keys.printed


def test_a_question_takes_a_number_or_typed_text() -> None:
    question = {"question": "Which database should the migration target?", "options": ["staging", "production"]}
    keys = _Keys("2")
    assert asyncio.run(keys.terminal().ask(dict(question))) == {"answer": "production"}
    assert "The coworker asks:\n  Which database should the migration target?\n    1. staging\n    2. production\n" in keys.printed
    assert "Type a number, or your own answer:" in keys.printed

    assert asyncio.run(_Keys("", "neither, use a copy").terminal().ask(dict(question))) == {"answer": "neither, use a copy"}
    assert asyncio.run(_Keys("the blue one").terminal().ask({"question": "Which colour?"})) == {"answer": "the blue one"}

    keys = _Keys("other", "1")
    assert asyncio.run(keys.terminal().ask({**question, "allow_text": False})) == {"answer": "staging"}
    assert "Type a number from 1 to 2." in keys.printed


def test_grouped_questions_are_asked_in_turn_and_keyed_by_header() -> None:
    keys = _Keys("1, 2", "weekly")
    got = asyncio.run(
        keys.terminal().ask(
            {
                "questions": [
                    {"header": "Regions", "question": "Which regions?", "options": [{"label": "EU"}, {"label": "US"}], "multi": True},
                    {"question": "How often?"},
                ]
            }
        )
    )
    assert got == {"answers": {"Regions": "EU, US", "How often?": "weekly"}}


def test_a_folder_request_is_asked_and_yes_gives_the_folder(tmp_path: Path) -> None:
    exports = tmp_path / "exports"
    exports.mkdir()
    given: list[tuple[Path, bool]] = []

    def terminal(keys: _Keys) -> Terminal:
        term = keys.terminal()
        term.grant_folder = lambda folder, writable: given.append((folder, writable)) or folder.resolve()
        return term

    request = {"path": str(exports), "writable": True, "reason": "the report reads last month's exports"}
    keys = _Keys("y")
    got = asyncio.run(terminal(keys).directory(dict(request)))
    assert got == {"granted": True, "path": str(exports.resolve()), "writable": True}
    assert f"The coworker asks for a folder to read and write: {exports}" in keys.printed
    assert "Reason: the report reads last month's exports" in keys.printed
    assert "[y] yes   [r] yes, read only   [n] no   or type another folder" in keys.printed

    # Read only, although it asked to write.
    assert asyncio.run(terminal(_Keys("r")).directory(dict(request)))["writable"] is False
    # A different folder, typed; one that does not exist is asked again.
    other = tmp_path / "other"
    other.mkdir()
    keys = _Keys(str(tmp_path / "missing"), str(other))
    assert asyncio.run(terminal(keys).directory(dict(request)))["path"] == str(other.resolve())
    assert "is not a folder" in keys.printed
    assert given == [(exports, True), (exports, False), (other, True)]

    # No, and a closed terminal, both decline and give nothing.
    for keys in (_Keys("n"), _Keys()):
        assert asyncio.run(terminal(keys).directory(dict(request))) == {"granted": False, "reason": "the user declined the request"}
    assert len(given) == 3

    # A read request has no read-only choice; a request with no path asks for one.
    keys = _Keys("n")
    asyncio.run(terminal(keys).directory({"path": str(exports)}))
    assert "to read:" in keys.printed and "[r]" not in keys.printed
    keys = _Keys("y", str(exports))
    assert asyncio.run(terminal(keys).directory({"reason": "needs the data"}))["granted"] is True
    assert "Type the folder, or n for no" in keys.printed


def test_a_tool_install_is_asked_and_yes_installs_the_pinned_build(monkeypatch: pytest.MonkeyPatch) -> None:
    from coworker import toolchain

    monkeypatch.setattr(toolchain, "describe", lambda name: {"version": "1.2.3", "summary": "a scanner"})
    installed: list[str] = []
    monkeypatch.setattr(toolchain, "install", lambda name: installed.append(name) or f"/tools/{name}")

    keys = _Keys("what", "y")
    got = asyncio.run(keys.terminal().tool({"name": "scanner", "reason": "to check the lockfile"}))
    assert got == {"installed": True, "path": "/tools/scanner", "version": "1.2.3"} and installed == ["scanner"]
    assert "The coworker asks to install a tool: scanner 1.2.3" in keys.printed and "[y] install it   [n] no" in keys.printed

    got = asyncio.run(_Keys("n").terminal().tool({"name": "scanner"}))
    assert got["installed"] is False and installed == ["scanner"]


# -- approvals are separate from questions --------------------------------------------


def test_with_auto_answer_the_cards_still_reach_the_person() -> None:
    from coworker.engine import TurnEngine

    engine = TurnEngine.__new__(TurnEngine)
    engine.attendance = lambda: "auto"
    engine.cards_reach_person = None
    assert engine._auto_answering() and engine._nobody_answers_cards()  # the app's "auto": both
    engine.cards_reach_person = lambda: True
    assert engine._auto_answering() and not engine._nobody_answers_cards()
    engine.attendance = lambda: "attended"
    engine.cards_reach_person = None
    assert not engine._nobody_answers_cards()


# -- at a real terminal -----------------------------------------------------------------


def _at_a_terminal(tmp_path: Path, script: list[dict], typed: list[tuple[str, str]], *extra: str):
    """Run with a terminal on standard input and standard error (standard output stays a
    pipe). `typed`: (text to wait for, line to type) in order. Returns (exit code,
    what the terminal showed, standard output, workspace)."""
    import pty

    machine = tmp_path / "machine"
    machine.mkdir(exist_ok=True)
    cmd, ws = _command(tmp_path, script, "--model", "anthropic/claude-sonnet-5", *extra)
    master, slave = pty.openpty()
    proc = subprocess.Popen(cmd, env=_env(machine), stdin=slave, stderr=slave, stdout=subprocess.PIPE)
    os.close(slave)
    shown = ""
    waiting = list(typed)
    deadline = time.time() + 120
    try:
        while time.time() < deadline:
            ready, _, _ = select.select([master], [], [], 0.2)
            if ready:
                try:
                    chunk = os.read(master, 4096)
                except OSError:  # the run closed its end
                    break
                if not chunk:
                    break
                shown += chunk.decode("utf-8", errors="replace")
            if waiting and waiting[0][0] in shown and shown.rstrip().endswith(">"):
                os.write(master, (waiting.pop(0)[1] + "\n").encode("utf-8"))
            if proc.poll() is not None and not ready:
                break
        out = proc.stdout.read().decode("utf-8") if proc.stdout else ""
        return proc.wait(timeout=30), shown, out, ws
    finally:
        if proc.poll() is None:
            proc.kill()
        os.close(master)


@pytest.mark.skipif(sys.platform == "win32", reason="needs a pseudo-terminal")
def test_at_a_terminal_the_approval_is_asked_and_yes_runs_the_call(tmp_path: Path) -> None:
    code, shown, out, ws = _at_a_terminal(tmp_path, WRITE_THEN_DONE, [("Approval needed: use write_file", "y")])
    assert code == 0, shown
    assert (ws / "hello.txt").read_text(encoding="utf-8") == "hello"
    assert "path: hello.txt" in shown and "no terminal is attached" not in shown
    assert out == "The file is written. Done.\n"


@pytest.mark.skipif(sys.platform == "win32", reason="needs a pseudo-terminal")
def test_at_a_terminal_no_refuses_the_call(tmp_path: Path) -> None:
    code, shown, _out, ws = _at_a_terminal(tmp_path, WRITE_THEN_DONE, [("Approval needed: use write_file", "n")])
    assert code == 0, shown
    assert not (ws / "hello.txt").exists()


QUESTION_THEN_WRITE = [
    {"text": "", "tool_calls": [{"name": "ask_user", "arguments": {"question": "Which colour?", "options": ["red", "blue"]}}]},
    {"text": "", "tool_calls": [{"name": "write_file", "arguments": {"path": "hello.txt", "content": "hello"}}]},
    {"text": "Done."},
]


@pytest.mark.skipif(sys.platform == "win32", reason="needs a pseudo-terminal")
def test_at_a_terminal_a_question_is_asked_there(tmp_path: Path) -> None:
    code, shown, _out, ws = _at_a_terminal(
        tmp_path, QUESTION_THEN_WRITE, [("Which colour?", "2")], "--approval-mode", "bypass-approvals",
    )
    assert code == 0, shown
    assert "The coworker asks:" in shown and "2. blue" in shown
    machine = tmp_path / "machine"
    (folder,) = _session_folders(machine, ws)
    messages = json.loads((folder / "messages.json").read_text(encoding="utf-8"))
    assert any(m.get("role") == "tool" and "blue" in str(m.get("content")) for m in messages)
    assert json.loads((folder / "summary.json").read_text(encoding="utf-8"))["args"]["auto_answer"] is False


@pytest.mark.skipif(sys.platform == "win32", reason="needs a pseudo-terminal")
def test_auto_answer_at_a_terminal_answers_the_question_and_still_asks_the_approval(tmp_path: Path) -> None:
    code, shown, _out, ws = _at_a_terminal(
        tmp_path, QUESTION_THEN_WRITE, [("Approval needed: use write_file", "y")], "--auto-answer",
    )
    assert code == 0, shown
    assert "The coworker asks:" not in shown  # answered by rule
    assert (ws / "hello.txt").exists()  # the approval was the person's, and they said yes
    (folder,) = _session_folders(tmp_path / "machine", ws)
    summary = json.loads((folder / "summary.json").read_text(encoding="utf-8"))
    assert summary["answers"]["questions_answered"] == 1 and summary["answers"]["cards_refused"] == 0


@pytest.mark.skipif(sys.platform == "win32", reason="needs a pseudo-terminal")
def test_ctrl_c_at_a_prompt_ends_the_run_and_leaves_the_record(tmp_path: Path) -> None:
    import pty
    import signal

    machine = tmp_path / "machine"
    machine.mkdir()
    cmd, ws = _command(tmp_path, WRITE_THEN_DONE, "--model", "anthropic/claude-sonnet-5")
    master, slave = pty.openpty()
    proc = subprocess.Popen(cmd, env=_env(machine), stdin=slave, stderr=slave, stdout=subprocess.PIPE)
    os.close(slave)
    try:
        shown = ""
        deadline = time.time() + 60
        while "Approval needed" not in shown or not shown.rstrip().endswith(">"):
            assert time.time() < deadline, shown
            if select.select([master], [], [], 0.2)[0]:
                shown += os.read(master, 4096).decode("utf-8", errors="replace")
        proc.send_signal(signal.SIGINT)
        assert proc.wait(timeout=30) == 130  # without anyone pressing Enter
    finally:
        if proc.poll() is None:
            proc.kill()
        os.close(master)
    assert not (ws / "hello.txt").exists()
    (folder,) = _session_folders(machine, ws)
    assert json.loads((folder / "summary.json").read_text(encoding="utf-8"))["outcome"] == "interrupted"



FOLDER_THEN_WRITE = [
    {"text": "", "tool_calls": [{"name": "request_directory", "arguments": {"path": "EXPORTS", "writable": True, "reason": "to save the report there"}}]},
    {"text": "", "tool_calls": [{"name": "write_file", "arguments": {"path": "EXPORTS/report.txt", "content": "done"}}]},
    {"text": "Saved."},
]


def _folder_script(exports: Path) -> list[dict]:
    return json.loads(json.dumps(FOLDER_THEN_WRITE).replace("EXPORTS", str(exports)))


@pytest.mark.skipif(sys.platform == "win32", reason="needs a pseudo-terminal")
def test_at_a_terminal_a_folder_request_is_asked_and_the_folder_can_then_be_written(tmp_path: Path) -> None:
    exports = tmp_path / "exports"
    exports.mkdir()
    code, shown, _out, _ws = _at_a_terminal(
        tmp_path, _folder_script(exports), [("The coworker asks for a folder", "y")], "--approval-mode", "bypass-approvals",
    )
    assert code == 0, shown
    assert "to save the report there" in shown
    assert (exports / "report.txt").read_text(encoding="utf-8") == "done"


def test_with_auto_answer_a_folder_request_is_declined_by_rule(tmp_path: Path) -> None:
    exports = tmp_path / "exports"
    exports.mkdir()
    proc, ws, machine = _run(
        tmp_path, _folder_script(exports), "--model", "anthropic/claude-sonnet-5", "--approval-mode", "bypass-approvals", "--auto-answer",
    )
    assert proc.returncode == 0, proc.stderr
    assert not (exports / "report.txt").exists()
    (folder,) = _session_folders(machine, ws)
    assert json.loads((folder / "summary.json").read_text(encoding="utf-8"))["answers"]["directory_requests_declined"] == 1


# -- a failing model call ---------------------------------------------------------------


def test_the_root_cause_is_the_bottom_of_the_chain() -> None:
    from coworker.headless.runner import root_cause_text

    ProxyError = type("ProxyError", (Exception,), {})
    try:
        try:
            try:
                raise ProxyError("403 Forbidden")
            except Exception as low:
                raise RuntimeError("httpx failed") from low
        except Exception as mid:
            raise ConnectionError("Connection error.") from mid
    except Exception as top:
        assert root_cause_text(top) == "the proxy answered 403 Forbidden"

    try:
        try:
            raise OSError("[Errno 111] Connection refused")
        except Exception as low:
            raise ConnectionError("Connection error.") from low
    except Exception as top:
        assert root_cause_text(top) == "OSError: [Errno 111] Connection refused"

    assert root_cause_text(ValueError("nothing underneath")) == ""


REFUSED_BY_A_PROXY = {"error": "Connection error.", "error_type": "APIConnectionError", "cause": "403 Forbidden", "cause_type": "ProxyError"}


def test_the_retry_line_says_why_the_model_call_failed(tmp_path: Path) -> None:
    machine = tmp_path / "machine"
    machine.mkdir()
    cmd, ws = _command(tmp_path, [REFUSED_BY_A_PROXY, {"text": "Recovered."}], "--model", "anthropic/claude-sonnet-5")
    env = {**_env(machine), "OPENWORKER_RUN_RETRY_DELAYS": "0"}
    proc = subprocess.run(cmd, env=env, capture_output=True, text=True, timeout=300)
    assert proc.returncode == 0, proc.stderr
    assert "provider error (APIConnectionError: Connection error. Cause: the proxy answered 403 Forbidden); retry 1/1 in 0s" in proc.stderr
    (folder,) = _session_folders(machine, ws)
    events = [json.loads(line) for line in (folder / "events.jsonl").read_text(encoding="utf-8").splitlines()]
    (retry,) = [e for e in events if e["type"] == "provider_retry"]
    assert retry["data"]["cause"] == "the proxy answered 403 Forbidden"

    # When the retries run out, the closing line says why as well.
    second = tmp_path / "second"
    second.mkdir()
    cmd, _ws = _command(second, [REFUSED_BY_A_PROXY, REFUSED_BY_A_PROXY], "--model", "anthropic/claude-sonnet-5")
    proc = subprocess.run(cmd, env={**_env(machine), "OPENWORKER_RUN_RETRY_DELAYS": "0"}, capture_output=True, text=True, timeout=300)
    assert proc.returncode == 3
    assert proc.stderr.rstrip().splitlines()[-1].endswith("Cause: the proxy answered 403 Forbidden")


@pytest.mark.skipif(sys.platform == "win32", reason="sends a stop signal")
def test_a_stop_signal_ends_the_run_during_the_wait_between_retries(tmp_path: Path) -> None:
    import signal

    machine = tmp_path / "machine"
    machine.mkdir()
    cmd, ws = _command(tmp_path, [REFUSED_BY_A_PROXY, {"text": "never reached"}], "--model", "anthropic/claude-sonnet-5")
    env = {**_env(machine), "OPENWORKER_RUN_RETRY_DELAYS": "600"}
    proc = subprocess.Popen(cmd, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    try:
        shown = ""
        deadline = time.time() + 60
        while "retry 1/1 in 600s" not in shown:
            assert time.time() < deadline, shown
            if select.select([proc.stderr], [], [], 0.2)[0]:
                shown += os.read(proc.stderr.fileno(), 4096).decode("utf-8", errors="replace")
        time.sleep(0.5)  # it is now waiting out the 600 seconds
        proc.send_signal(signal.SIGTERM)
        assert proc.wait(timeout=30) == 130  # it stopped; it did not wait
    finally:
        if proc.poll() is None:
            proc.kill()
    (folder,) = _session_folders(machine, ws)
    assert json.loads((folder / "summary.json").read_text(encoding="utf-8"))["outcome"] == "interrupted"