import os import time from controllers.types import Agents, Environments, Notices, Nudges from engine.record import Record from features.permission_prompts.commands import AnswerPermission from overview.summary import summarize from resources.types import EnvironmentKind from engine.sessions import Sessions from runner.hooks import answer from tests.kit import handle from providers import DRIVERS, PROVIDERS from resources.base import SYSTEM from tests.conftest import fresh, refused def test_a_permission_the_agent_waits_on_is_shown_in_the_chat_until_it_is_answered(): record = fresh() provider = PROVIDERS["claude"]() hook = {"session_id": "claude-1", "tool_name": "Bash", "tool_input": {"command": "git push"}} def waiting(): return [n.title for n in Notices(record, actor=SYSTEM).rows.standing() if n.data.get("action") == "permission"] handle(provider, record.root, record.env, {**hook, "hook_event_name": "PreToolUse"}) assert waiting() == [], "a tool use alone asks for nothing" handle(provider, record.root, record.env, {**hook, "hook_event_name": "PermissionRequest"}) assert waiting() == ["Waiting for permission - Bash git push"], "the prompt is shown, naming the call" assert Agents(record, actor=SYSTEM).by_session("claude-1").asking["tool"] == "Bash" handle(provider, record.root, record.env, {**hook, "hook_event_name": "PostToolUse"}) assert waiting() == [], "the call ran: the notice goes" def test_the_skip_switch_restarts_in_the_same_conversation_with_the_flag(monkeypatch): claude = DRIVERS["claude"] assert claude.resumed(claude.skipping(["-c", "--model", "opus"], True), "abc") == \ ["--dangerously-skip-permissions", "--model", "opus", "--resume", "abc"], "skip on, resumed in place of continue" assert claude.skipping(["--dangerously-skip-permissions", "x"], False) == ["x"], "skip off drops the flag" from features.permission_prompts.skipping import launch_args typed = fresh() typed.set_setting("permission_prompts", {"skip": False}) assert (launch_args(typed, "claude", ["--dangerously-skip-permissions"]), typed.setting("permission_prompts", {}).get("skip")) == \ (["--dangerously-skip-permissions"], True), "a flag typed at launch passes through and turns the switch back on" assert launch_args(typed, "claude", []) == ["--dangerously-skip-permissions"], "so the next launch carries it by itself" assert launch_args(fresh(), "claude", []) == ["--dangerously-skip-permissions"], "a project that never set the switch runs without prompts" off = fresh() off.set_setting("features", {"work_tracking.auto": False}) off.set_setting("permission_prompts", {"skip": False}) assert launch_args(off, "claude", ["--model", "opus"]) == ["--model", "opus"], "switched off in Settings, no flag is added" codex = DRIVERS["codex"] assert codex.skipping(["x"], True) == ["--dangerously-bypass-approvals-and-sandbox", "x"], "Codex runs its commands without asking" assert codex.skipping(["--approve-for-me", "x"], True) == ["--dangerously-bypass-approvals-and-sandbox", "x"], \ "Codex refuses approve-for-me beside the bypass, so skipping drops it" asked = b"\x1b[2m> Ask Codex to do anything\x1b[0m\r\nDo you trust the contents of this directory?\r\n\xe2\x80\xba 1. Yes, continue\r\n 2. No, quit" assert (codex.consent(asked), codex.opening(asked)) == (b"1\r", ""), "Codex's trust question is answered Yes by number, before the opening" assert codex.opening(b"\x1b[2m> Ask Codex to do anything\x1b[0m") == codex.OPENING, "at its empty prompt Codex is given the journal's opening line" assert codex.consent(asked + b"\r\n> Ask Codex to do anything") == b"", "once Codex is at its prompt, nothing more is typed into the question" folder = b"Folder access /p Trust this folder?\r\nCodex can read, edit, and run files here.\r\n\xe2\x80\xba 1. Trust and continue\r\n 2. Quit" assert codex.consent(folder) == b"1\r", "Codex 0.160 words its trust question differently, and it is answered the same way" screen = ("Running printf 'hi' > hello.txt\r\nWould you like to run the following command?\r\nReason: May I create hello.txt?\r\n" "$ printf 'hi' > hello.txt\r\n\u203a 1. Yes, proceed (y)\r\n 2. Yes, and don't ask again (p)\r\n 3. No, and tell Codex what to do differently (esc)") driver = codex(fresh(), "codex-ask") driver.printed.parent.mkdir(parents=True, exist_ok=True) driver.printed.write_bytes(f"\x1b[2m> Ask Codex to do anything\x1b[0m\r\nesc to interrupt\r\n{screen}".encode()) assert (driver.asked().tool, driver.asked().call, codex.ALLOW) == ("exec_command", "printf 'hi' > hello.txt", b"y"), \ "Codex's approval prompt is read off its screen with its command, and Allow presses y" assert "Would you like to run" in driver.screen(driver.PROMPT_TAIL), "a driver with no suggestion pattern of its own still reads its screen, so the menu funnel never crashes a Codex session" driver.last_report = lambda: None driver.send("todo 5 next") assert driver.held == ["todo 5 next"], "nothing is typed into Codex while its approval prompt waits on the user; the line waits too" del driver.last_report driver.printed.write_bytes(f"{screen}\r\n\x1b[2m> Ask Codex to do anything\x1b[0m".encode()) assert driver.asked() is None, "once the prompt is gone, nothing is asked" driver.QUIET = 0.0 assert driver.at_prompt() is True, "an agent whose screen ends at its empty prompt is at the prompt" driver.printed.write_bytes(b"esc to interrupt\r\n\x1b[2m> Ask Codex to do anything\x1b[0m\r\nesc to interrupt") assert driver.at_prompt() is False, "and one still busy under it is not" assert launch_args(fresh(), "nobody-we-know", ["--x"]) == ["--x"], "arguments for an agent the journal does not know are left as they are" from pathlib import Path assert (codex.command([], Path.cwd())[:2], codex.command([codex.TRUSTS_HOOKS])[0:2]) == (["codex", codex.TRUSTS_HOOKS], ["codex", codex.TRUSTS_HOOKS]), \ "Codex is started trusting the journal's hooks, once, and the folder it is started in" assert any("trust_level" in word for word in codex.command([], Path.cwd())), "and trusting the folder it starts in" assert codex.resumed(["--model", "x"], "") == ["--model", "x"], "with no conversation to carry on, the arguments are as they were" from runner.engine import Engine from tests.kit import report record = driver.record report(record, "working", "PreToolUse", session="codex-ask", provider="codex") engine, row = Engine(record, driver), lambda: Agents(record, actor=SYSTEM).by_session("codex-ask") calls = [] for command in ("printf 'hi' > hello.txt", "rm hello.txt"): asking_screen = screen.replace("printf 'hi' > hello.txt", command) driver.printed.write_bytes(f"esc to interrupt\r\n{asking_screen}".encode()) driver.reported = (float("-inf"), None) engine.screen_asks() calls.append(row().asking["call"]) assert calls == ["printf 'hi' > hello.txt", "rm hello.txt"], "a second approval right after the first replaces the command shown in the chat" claude = DRIVERS["claude"] danger = ("⏺ Bash(rm -rf $DIR/)\r\nBash command\r\n\r\n rm -rf $DIR/\r\n Remove the directory\r\n\r\n Dangerous rm operation on possibly-empty variable path\r\n" " Do you want to proceed?\r\n ❯ 1. Yes\r\n 2. No\r\n") asker = claude(fresh(), "claude-danger") asker.printed.parent.mkdir(parents=True, exist_ok=True) asker.printed.write_bytes(f"esc to interrupt\r\n{danger}".encode()) assert (asker.asking(), asker.asked().tool, "Dangerous rm operation" in asker.asked().call) == (True, "Bash", True), \ "Claude's own safety question, which sends no permission request, is read off its screen with the command it names" asker.printed.write_bytes(f"{danger}\r\nesc to interrupt".encode()) assert asker.asked() is None, "once the command runs again, nothing is asked" pressed, slept = [], [] asker.press_raw = asker._wrote = pressed.append monkeypatch.setattr(time, "sleep", slept.append) asker.stop_turn() asker.permit(False) assert (pressed, len(slept), slept[-1] > 0.5) == ([b"\x1b", b"\x1b"], 2, True), \ "the journal spaces its Escape presses so two never open Claude Code's Rewind menu" asker._screen_file().parent.mkdir(parents=True, exist_ok=True) asker._screen_file().write_bytes("Conversation compacted\r\n❯\xa0\r\nChecking for updates\r\n❯ [journal] ticket 5: build the plan\r\n".encode()) typed_note = "[journal] ticket 5: build the plan" assert asker._taken(typed_note, 0) is False, "a note still sitting in the input box has not been taken, whatever status line is on screen" asker._screen_file().write_bytes("❯ [journal] ticket 5: build the plan\r\n✻ Thinking… (esc to interrupt)\r\n❯ \r\n".encode()) assert asker._taken(typed_note, 0) is True, "a note after which the agent shows it is at work and the box is empty was taken, though its hook has not reported yet" warning = "WARNING: Loading development channels\r\n--dangerously-load-development-channels is for local channel development only.\r\n" \ "❯ 1. I am using this for local development\r\n 2. Exit\r\nEnter to confirm".encode() assert (claude.consent(warning), claude.consent(warning + "\r\n❯ ".encode())) == (b"\r", b""), \ "Claude's development channels warning is answered with Enter while its menu is the last thing on screen, never once the prompt is back" from providers.dialogs import Menu sharing = Menu.on("Blender MCP wants to share data to improve the product. Allow?\r\n❯ 1. Yes, allow\r\n 2. No, do not share\r\nEnter to select · Esc to cancel") trusting = Menu.on("Do you trust the files in this folder?\r\n❯ 1. Yes, I trust this folder\r\n 2. No, exit\r\nEnter to confirm · Esc to cancel") unknown = Menu.on("Pick a colour\r\n❯ 1. Red\r\n 2. Blue\r\nEnter to select") permission = Menu.on("Do you want to proceed?\r\n❯ 1. Yes\r\n 2. No\r\nEnter to select") assert (sharing.choice().keys, trusting.choice().keys, unknown.choice().keys, permission.foreign(), Menu.on("no menu here\r\n❯ ")) == \ (b"\x1b[B\r", b"\r", b"\x1b", False, None), \ "a menu is recognised by its shape and chosen by its words: data sharing is declined by moving to the label, a folder the journal launched the agent into is trusted, anything unknown is closed, and the agent's own permission question is left to its asks" rewind = Menu.on("Rewind\r\n❯ 1. fix the login\r\n 2. add the page\r\n 3. (current)\r\nEnter to continue · Esc to cancel") assert rewind.foreign() is False, "Claude's Rewind list, which has no decline label, is a list of past messages you opened to read, so the journal never closes it" from types import SimpleNamespace from runner.worker import Dialogs shown = "Blender MCP wants to share data. Allow?\r\n❯ 1. Yes, allow\r\n 2. No, do not share\r\nEnter to select · Esc to cancel" pressed = [] stub = SimpleNamespace(PROMPT_TAIL=1, name="claude", record=fresh(), screen=lambda size: shown, last_report=lambda: None, quiet_for=lambda: 9.0, press_raw=pressed.append, keyed_at=lambda: 0.0) own = Dialogs(stub) own.tick() assert pressed == [b"\x1b[B\r"], "a menu that came up with no key pressed is answered" opened = SimpleNamespace(**{**vars(stub), "press_raw": pressed.append, "keyed_at": lambda: time.time()}) pressed.clear() mine = Dialogs(opened) mine.tick() assert pressed == [], "a menu that came up just after a key you pressed is left to you, however quiet the terminal is" update = b"\x1b[2m> Ask Codex to do anything\x1b[0m\r\nUpdate available 0.159.3 \xe2\x86\x92 0.160.0\r\n\xe2\x80\xba 1. Update now\r\n 2. Skip\r\n 3. Skip until next version" assert (codex.consent(update), codex.opening(update)) == (b"2\r", ""), "Codex's update question is skipped, and the opening waits until it is gone" assert (codex.carried_on(["continue"]), codex.carried_on(["--resume", "abc"]), codex.carried_on(["-c", "k=v"])) == \ (["resume", "--last"], ["resume", "abc"], ["-c", "k=v"]), "Codex continues and resumes with its resume subcommand, and -c stays its config flag" assert codex.carried_on(codex.resumed(codex.skipping(["continue"], True), "abc")) == \ ["resume", "abc", "--dangerously-bypass-approvals-and-sandbox"], "Codex restarts in the same conversation" def test_every_flag_typed_at_launch_reaches_the_agent_whatever_the_switch_says(): from providers import DRIVERS from features.permission_prompts.skipping import launch_args for name, driver in DRIVERS.items(): typed = ["--model", "opus", "--some-flag", "value", *driver.SKIP_ARGS] for skip in (True, False): record = fresh() record.set_setting("permission_prompts", {"skip": skip}) given = launch_args(record, name, typed) assert all(arg in given for arg in typed), f"{name}, switch {'on' if skip else 'off'}: every typed flag is passed on" assert given[given.index("--model"):given.index("--model") + 4] == typed[:4], f"{name}: in the order they were typed" def test_the_start_asks_about_permission_prompts_only_when_the_flag_is_not_typed(): from tests.kit import asked_prompts record = fresh() asked = [] asked_prompts(record, "claude", ["--dangerously-skip-permissions"], ask=lambda _: asked.append(1) or "", answering=True) assert asked == [], "a typed flag is the answer already" asked_prompts(record, "claude", [], ask=lambda _: "2", answering=True) assert record.setting("permission_prompts", {}).get("skip") is False, "No keeps the prompts, and is remembered" def closed(_=""): raise EOFError asked_prompts(record, "claude", [], ask=closed, answering=True) assert record.setting("permission_prompts", {}).get("skip") is False, "input that ends before an answer changes nothing" def test_claude_is_kept_out_of_the_record_files_but_not_their_attachments(tmp_path): from providers import PROVIDERS from providers.base import HookCommand from providers.claude import RECORD_FILES claude = PROVIDERS["claude"]() claude.save(tmp_path, {"permissions": {"deny": ["Read(./.env)"]}}) hook = HookCommand(tmp_path / ".journal" / "src" / "hook.sh", "claude", tmp_path / ".journal") claude.wire(tmp_path, hook) claude.wire(tmp_path, hook) deny = claude.settings(tmp_path)["permissions"]["deny"] assert deny == ["Read(./.env)", *RECORD_FILES], "rows are denied once, beside what the project already denied" assert all("*/*.md" in rule for rule in RECORD_FILES), "only the row files: an attached picture a folder deeper stays readable" def test_auto_mode_launches_each_agent_in_its_own_approval_mode(): from features.permission_prompts.skipping import launch_args record = fresh() record.features = {**record.features, "work_tracking.auto": True} record.set_setting("permission_prompts", {"skip": False}) assert launch_args(record, "claude", ["--model", "sonnet"]) == ["--dangerously-skip-permissions", "--model", "sonnet"], \ "in auto mode Claude never stops at a permission prompt" assert launch_args(record, "codex", ["--model", "gpt-5"]) == ["--dangerously-bypass-approvals-and-sandbox", "--model", "gpt-5"], \ "in auto mode Codex never stops for an approval" assert launch_args(record, "claude", ["--permission-mode=dontAsk"]) == ["--permission-mode=dontAsk"], \ "an explicit Claude permission choice wins" assert launch_args(record, "codex", ["--ask-for-approval", "never"]) == ["--ask-for-approval", "never"], \ "an explicit Codex approval choice wins" def test_an_agents_controls_are_pressed_from_the_viewer_only_for_a_session_that_is_online(monkeypatch): import time import features from commands.http import dispatch from controllers.types import Agents, Notices from engine import runtime from engine.stored import write_json from resources.base import SYSTEM from tests.conftest import fresh features.load() record = fresh() seat = runtime.session_file(record.root, "claude-7", "seat.json") write_json(seat, {"at": time.time(), "agent": "claude", "env": record.env, "report": {"title": "claude-7", "provider": "claude", "model": "opus"}, "reported": {"title": "claude-7", "provider": "claude", "model": "opus"}}) from engine.sessions import Sessions Sessions(record.root).bind("claude-7", record.env, provider="claude") Agents(record, actor=SYSTEM).create("claude-7", provider="claude", status="idle") post = lambda action, body=None, session="claude-7", env=None: dispatch("POST", f"/api/{env or record.env}/agent/{session}/{action}", record.root, {}, body or {}) for action in ("force", "pause", "resume"): assert post(action).code == 200, f"{action} queues for a live session" assert post("permit", {"allow": True}).code == 200 and post("permit", {"allow": False}).code == 200, "a permission is allowed or denied from the viewer" assert post("shell", {"command": " "}).code == 400, "an empty shell line is refused" assert post("shell", {"command": "ls", "now": True}).code == 200, "a shell line can be run at once" assert post("pause", session="claude-404").code == 400, "a session that is not online is refused" assert post("pause", env="elsewhere").code == 400, "a session of another environment is refused" assert (post("control", {"action": "model", "value": "opus"}).body["label"], post("control", {"action": "effort", "value": "high"}).body["label"]) == ("Opus", "High"), "a model or an effort chosen in the viewer is queued for the agent" assert "does not support" in post("control", {"action": "effort", "value": "bogus"}).body["error"], "a choice the agent does not offer is refused" controls = dispatch("GET", "/api/agent-controls/claude", record.root, {"model": "opus", "effort": "high"}, {}) assert controls.code == 200 and controls.body["provider"] == "claude", "the controls offered are the provider's" assert dispatch("GET", "/api/agent-controls/nobody", record.root, {}, {}).body["groups"] == [], "an agent with no controls offers none" def test_a_burst_of_tool_uses_writes_the_session_file_once_not_on_every_hook(monkeypatch): record = fresh() provider = PROVIDERS["claude"]() hook = {"hook_event_name": "PreToolUse", "session_id": "claude-9", "tool_name": "Read", "tool_input": {"file_path": "/tmp/x"}} answer(provider, record.root, hook, os.getpid(), record.env) first = Sessions(record.root).read("claude-9").seen answer(provider, record.root, hook, os.getpid(), record.env) assert Sessions(record.root).read("claude-9").seen == first, "a second tool use within seconds leaves the session file as it was" later = time.time() + 60 monkeypatch.setattr(time, "time", lambda: later) answer(provider, record.root, hook, os.getpid(), record.env) assert Sessions(record.root).read("claude-9").seen == later, "a minute on, the hook is heard again" def asked(env: Record): provider = PROVIDERS["claude"]() hook = {"hook_event_name": "PermissionRequest", "session_id": f"claude-{env.env}", "tool_name": "Bash", "tool_input": {"command": "git push"}} handle(provider, env.root, env.env, hook) return [n for n in Notices(env, actor=SYSTEM).rows.standing() if n.data.get("action") == "permission"] def helper_of(main: Record) -> Record: Environments(main, actor=SYSTEM).create("helper-ada", owner="helper:1", launched_from=main.env, kind=EnvironmentKind.HELPER) return Record(main.root, "helper-ada") def test_with_auto_and_orchestrator_on_a_helpers_permission_goes_to_the_orchestrating_agent_not_the_user(): main = fresh() main.set_setting("work_modes", {"mode": "orchestrator"}) Agents(main, actor=SYSTEM).create("claude-main") helper = helper_of(main) waiting = asked(helper) assert [n.data["to"] for n in waiting] == ["orchestrator"], "the request is marked as the orchestrator's" lines = [n.title for n in Nudges(main, actor=SYSTEM).rows.every()] assert "helper-ada waits for permission for Bash git push" in lines, "the orchestrating agent is told which agent, tool and call" cell = next(e for e in summarize(helper.root)["helpers"] if e["name"] == "helper-ada")["counts"] assert (cell["prompts"], cell["routed"]) == (0, 1), "nothing waits on the user, and the cell counts it as the orchestrator's" assert "not yours" in refused(lambda: AnswerPermission().run(None, Agents(helper, actor=SYSTEM), "helper-ada", "allow")), "only the launching environment answers it" assert "no agent is running" in refused(lambda: AnswerPermission().run(None, Agents(main, actor=SYSTEM), "helper-ada", "allow")), "the launching environment's agent may, once the agent runs" def test_outside_auto_and_orchestrator_mode_a_permission_request_stays_the_users(): main = fresh() Agents(main, actor=SYSTEM).create("claude-main") helper = helper_of(main) assert [n.data.get("to", "") for n in asked(helper)] == [""], "in builder mode nothing is routed" assert [n.title for n in Nudges(main, actor=SYSTEM).rows.every()] == [], "and the main agent is told nothing" assert "not yours" in refused(lambda: AnswerPermission().run(None, Agents(main, actor=SYSTEM), "helper-ada", "allow")), "its agent may not answer it"