"""Batch (multi-question) clarify panel state machine — CLI side. Drives ``_clarify_callback`` with a ``questions`` list on a background thread (the way the agent thread calls it) and simulates the keybinding handlers by calling the same helper methods they call (``_clarify_batch_set_active`` for Tab, ``_clarify_batch_enter`` for Enter, ``_clarify_batch_lock`` for the freetext submit path). No real terminal needed — mirrors tests/hermes_cli/test_cli_approval_ui.py. """ import json import threading import time from unittest.mock import MagicMock, patch from cli import HermesCLI from agent.i18n import t def _make_cli_stub(): cli = HermesCLI.__new__(HermesCLI) cli._clarify_state = None cli._clarify_freetext = False cli._clarify_multi_base = None cli._clarify_prefill = "" cli._clarify_deadline = None cli._paint_now = MagicMock() cli._persist_prompt_summary = MagicMock() return cli def _q(index, question, choices=None, multi_select=False): """One normalized batch entry, shaped like _normalize_questions output.""" return { "qid": f"q{index}", "question": question, "choices": list(choices) if choices else None, "choices_offered": list(choices) if choices else None, "multi_select": bool(multi_select) and bool(choices), } def _start_batch(cli, questions): """Run the batch callback on a thread; wait for the panel state.""" result = {} def _run(): result["value"] = cli._clarify_callback(questions) thread = threading.Thread(target=_run, daemon=True) thread.start() deadline = time.time() + 2 while cli._clarify_state is None and time.time() < deadline: time.sleep(0.01) assert cli._clarify_state is not None return thread, result class TestClarifyBatchPanel: def test_all_locked_returns_answers_dict_keyed_by_qid(self): cli = _make_cli_stub() questions = [ _q(0, "Color?", ["red", "blue"]), _q(1, "Size?", ["small", "large"]), ] thread, result = _start_batch(cli, questions) state = cli._clarify_state assert state["active"] == 0 assert state["choices"] == ["red", "blue"] # Enter locks the active question's highlighted choice, then the # cursor advances to the next unanswered question. cli._clarify_batch_enter(state) assert state["answers"] == {"q0": "red"} assert state["active"] == 1 state["selected"] = 1 cli._clarify_batch_enter(state) thread.join(timeout=2) assert result["value"] == {"answers": {"q0": "red", "q1": "large"}, "outcome": "submitted"} assert cli._clarify_state is None def test_any_order_answering_via_tab_cycle(self): cli = _make_cli_stub() questions = [ _q(0, "First?", ["a", "b"]), _q(1, "Second?", ["c", "d"]), _q(2, "Third?", ["e", "f"]), ] thread, result = _start_batch(cli, questions) state = cli._clarify_state # Tab twice: q0 -> q1 -> q2 (what the tab keybinding does). cli._clarify_batch_set_active(state, (state["active"] + 1) % 3) cli._clarify_batch_set_active(state, (state["active"] + 1) % 3) assert state["active"] == 2 cli._clarify_batch_enter(state) # lock q2 = "e" # Advance wraps to the next unanswered question (q0). assert state["active"] == 0 state["selected"] = 1 cli._clarify_batch_enter(state) # lock q0 = "b" assert state["active"] == 1 cli._clarify_batch_enter(state) # lock q1 = "c" thread.join(timeout=2) assert result["value"] == { "answers": {"q0": "b", "q1": "c", "q2": "e"}, "outcome": "submitted" } def test_reanswer_overwrites_before_completion(self): cli = _make_cli_stub() questions = [ _q(0, "Approach?", ["quick", "thorough"]), _q(1, "Scope?", ["narrow", "wide"]), ] thread, result = _start_batch(cli, questions) state = cli._clarify_state cli._clarify_batch_enter(state) # lock q0 = "quick" assert state["answers"]["q0"] == "quick" assert state["active"] == 1 # Tab back to the answered question and change the answer. cli._clarify_batch_set_active(state, 0) state["selected"] = 1 cli._clarify_batch_enter(state) # overwrite q0 = "thorough" assert state["answers"]["q0"] == "thorough" # Advance lands on the still-unanswered q1. assert state["active"] == 1 cli._clarify_batch_enter(state) # lock q1 = "narrow" thread.join(timeout=2) assert result["value"] == { "answers": {"q0": "thorough", "q1": "narrow"}, "outcome": "submitted" } def test_timeout_returns_partials_with_timed_out_outcome(self): cli = _make_cli_stub() questions = [ _q(0, "Answered?", ["yes", "no"]), _q(1, "Never answered?", ["x", "y"]), ] with patch( "tools.clarify_gateway.resolve_clarify_timeout", return_value=1 ): thread, result = _start_batch(cli, questions) state = cli._clarify_state cli._clarify_batch_enter(state) # lock q0 only thread.join(timeout=5) assert not thread.is_alive() assert result["value"] == {"answers": {"q0": "yes"}, "outcome": "timed_out"} assert cli._clarify_state is None def test_multi_select_lock_produces_json_array_string(self): cli = _make_cli_stub() questions = [ _q(0, "Toppings?", ["ham", "olives", "basil"], multi_select=True), ] thread, result = _start_batch(cli, questions) state = cli._clarify_state assert state["multi_select"] is True state["selected_indices"].update({0, 2}) cli._clarify_batch_enter(state) thread.join(timeout=2) answer = result["value"]["answers"]["q0"] assert isinstance(answer, str) assert json.loads(answer) == ["ham", "basil"] def test_open_ended_question_locks_typed_answer(self): cli = _make_cli_stub() questions = [ _q(0, "Anything else?"), _q(1, "Pick one", ["a", "b"]), ] thread, result = _start_batch(cli, questions) state = cli._clarify_state # Open-ended active question drops straight into freetext. assert cli._clarify_freetext is True # The Enter freetext submit path locks the typed text. cli._clarify_freetext = False cli._clarify_batch_lock(state, "custom words") assert state["active"] == 1 cli._clarify_batch_enter(state) thread.join(timeout=2) assert result["value"] == { "answers": {"q0": "custom words", "q1": "a"}, "outcome": "submitted" } def test_locked_question_persists_scrollback_summary(self): cli = _make_cli_stub() questions = [ _q(0, "Color?", ["red", "blue"]), _q(1, "Size?", ["small", "large"]), ] thread, result = _start_batch(cli, questions) state = cli._clarify_state cli._clarify_batch_enter(state) cli._clarify_batch_enter(state) thread.join(timeout=2) calls = cli._persist_prompt_summary.call_args_list assert len(calls) == 2 assert calls[0].args == ("?", "Clarify", "Color?", "red") assert calls[1].args == ("?", "Clarify", "Size?", "small") def test_connection_callback_masks_secret_and_submits_env(self): cli = _make_cli_stub() cli._connection_state = None cli._capture_modal_input_snapshot = MagicMock() cli._restore_modal_input_snapshot = MagicMock() cli._ring_bell = MagicMock() cli.session_id = "session" operation = MagicMock(op_id="op") payload = { "op_id": "op", "targets": [{ "name": "asana", "state": "pending", "instructions": "Use an Asana app.", "required_env": [ {"name": "CLIENT_ID", "prompt": "Client ID", "required": True, "default": "default-id"}, {"name": "CLIENT_SECRET", "prompt": "Client secret", "required": True, "secret": True}, ], }], } connection_result = {} with patch.object(cli, "_connection_operation", return_value=operation), patch( "tools.connectors.mcp.apply_answer" ) as apply_answer: connection_thread = threading.Thread( target=lambda: connection_result.setdefault("value", cli._connection_callback(payload)), daemon=True ) connection_thread.start() connection_thread.join(timeout=2) assert not connection_thread.is_alive() assert connection_result["value"] is None state = cli._connection_state state["drafts"]["asana"]["CLIENT_SECRET"] = "never-render-this" assert "never-render-this" not in "\n".join(cli._connection_render_lines()) assert f"Client secret*: {t('cli.connect.secret_set')}" in cli._connection_render_lines() cli._connection_answer(approve=True) sent = json.loads(apply_answer.call_args.args[1]) assert sent == {"targets": [{ "name": "asana", "status": "approved", "env": {"CLIENT_ID": "default-id", "CLIENT_SECRET": "never-render-this"}, }]} cli._connection_close() class TestClarifyBatchNavigation: """Shift-Tab, answer restore on re-visit, and Other edit-prefill.""" def test_backward_navigation_wraps(self): cli = _make_cli_stub() questions = [ _q(0, "Color?", ["red", "blue"]), _q(1, "Size?", ["small", "large"]), _q(2, "Speed?", ["slow", "fast"]), ] thread, result = _start_batch(cli, questions) state = cli._clarify_state # Shift-Tab from question 0 wraps to the last question. cli._clarify_batch_set_active(state, (state["active"] - 1) % 3) assert state["active"] == 2 cli._clarify_batch_set_active(state, (state["active"] - 1) % 3) assert state["active"] == 1 state["response_queue"].put(None) thread.join(timeout=2) def test_revisit_choice_answer_restores_cursor(self): cli = _make_cli_stub() questions = [ _q(0, "Color?", ["red", "blue"]), _q(1, "Size?", ["small", "large"]), ] thread, result = _start_batch(cli, questions) state = cli._clarify_state # Lock "blue" (index 1) on q0; the cursor advances to q1. state["selected"] = 1 cli._clarify_batch_enter(state) assert state["active"] == 1 # Tab back to q0: the cursor sits on the earlier answer, not row 0. cli._clarify_batch_set_active(state, 0) assert state["selected"] == 1 state["response_queue"].put(None) thread.join(timeout=2) def test_revisit_other_answer_highlights_other_and_prefills_edit(self): cli = _make_cli_stub() questions = [ _q(0, "Color?", ["red", "blue"]), _q(1, "Size?", ["small", "large"]), ] thread, result = _start_batch(cli, questions) state = cli._clarify_state # Answer q0 via Other: select the Other row, then the freetext # submit path locks the typed text with its meta. state["selected"] = 2 cli._clarify_batch_enter(state) assert cli._clarify_freetext is True cli._clarify_freetext = False cli._clarify_batch_lock( state, "chartreuse", meta={"kind": "other", "other_text": "chartreuse"} ) assert state["active"] == 1 # Tab back to q0: the cursor highlights the Other row. cli._clarify_batch_set_active(state, 0) assert state["selected"] == 2 # Enter on the answered Other switches to freetext and prefills the # earlier text for editing. cli._clarify_batch_enter(state) assert cli._clarify_freetext is True assert cli._clarify_prefill == "chartreuse" state["response_queue"].put(None) thread.join(timeout=2) def test_reanswer_overwrites_and_updates_meta(self): cli = _make_cli_stub() questions = [ _q(0, "Color?", ["red", "blue"]), _q(1, "Size?", ["small", "large"]), ] thread, result = _start_batch(cli, questions) state = cli._clarify_state # First answer via Other. cli._clarify_batch_lock( state, "teal", meta={"kind": "other", "other_text": "teal"} ) # Re-visit and overwrite with a plain choice. cli._clarify_batch_set_active(state, 0) assert state["selected"] == 2 state["selected"] = 0 cli._clarify_batch_enter(state) assert state["answers"]["q0"] == "red" assert state["answer_meta"]["q0"] == {"kind": "choice"} # Finish q1 so the batch resolves with the overwritten answer. cli._clarify_batch_set_active(state, 1) cli._clarify_batch_enter(state) thread.join(timeout=2) assert result["value"] == {"answers": {"q0": "red", "q1": "small"}, "outcome": "submitted"} class TestClarifyBellOnPrompt: """display.bell_on_prompt rings BEL when a clarify modal opens; off is silent.""" @staticmethod def _run_clarify(bell_on_prompt): import io cli = _make_cli_stub() cli.bell_on_prompt = bell_on_prompt out = io.StringIO() with patch("cli.sys.stdout", out), patch( "tools.clarify_gateway.resolve_clarify_timeout", return_value=60 ): thread = threading.Thread( target=cli._clarify_callback, args=([_q(0, "Color?", ["red", "blue"])],), daemon=True ) thread.start() deadline = time.time() + 2 while cli._clarify_state is None and time.time() < deadline: time.sleep(0.01) assert cli._clarify_state is not None cli._clarify_state["response_queue"].put(None) thread.join(timeout=2) return out.getvalue() def test_bell_on_prompt_rings_and_off_is_silent(self): assert "\a" in self._run_clarify(True) assert "\a" not in self._run_clarify(False)