HERMES-GIELIE/tests/test_1062_busy_input_modes.py

"""Regression tests for busy_input_mode (PR #1062, closes #720).

Pins the wiring for the three modes (queue / interrupt / steer):
- The setting key + default + enum validation in api/config.py
- Three slash commands registered in static/commands.js
- send()'s busy branch reads window._busyInputMode and dispatches
- Boot initializes window._busyInputMode from settings
- 17 new i18n keys present in all 6 locale blocks

Issue: #720 (configurable busy-input behaviour)
"""
from pathlib import Path

ROOT = Path(__file__).parent.parent
CONFIG_PY = (ROOT / "api" / "config.py").read_text(encoding="utf-8")
COMMANDS_JS = (ROOT / "static" / "commands.js").read_text(encoding="utf-8")
MESSAGES_JS = (ROOT / "static" / "messages.js").read_text(encoding="utf-8")
BOOT_JS = (ROOT / "static" / "boot.js").read_text(encoding="utf-8")
PANELS_JS = (ROOT / "static" / "panels.js").read_text(encoding="utf-8")
INDEX_HTML = (ROOT / "static" / "index.html").read_text(encoding="utf-8")
I18N_JS = (ROOT / "static" / "i18n.js").read_text(encoding="utf-8")


# ── Backend: setting registration + enum validation ─────────────────────

class TestBusyInputModeSetting:
    """The new setting key must be registered with a default and enum validator."""

    def test_default_is_queue(self):
        """Default value preserves existing queue behaviour for users who don't touch the setting."""
        assert '"busy_input_mode": "queue"' in CONFIG_PY, (
            "_DEFAULT_SETTINGS must include busy_input_mode='queue' so existing users see no change"
        )

    def test_enum_validator_present(self):
        """_SETTINGS_ENUM_KEYS must validate busy_input_mode against {queue, interrupt, steer}."""
        # Find the entry inside the enum dict (a set literal as the value)
        idx = CONFIG_PY.find('"busy_input_mode": {')
        assert idx >= 0, "busy_input_mode entry missing from _SETTINGS_ENUM_KEYS"
        block = CONFIG_PY[idx:idx + 200]
        assert '"queue"' in block and '"interrupt"' in block and '"steer"' in block, (
            "busy_input_mode enum must contain {queue, interrupt, steer}"
        )


# ── Frontend: slash commands ─────────────────────────────────────────────

class TestSlashCommandRegistration:
    """The three new slash commands must be registered in COMMANDS array."""

    def test_queue_command_registered(self):
        assert "name:'queue'" in COMMANDS_JS and "fn:cmdQueue" in COMMANDS_JS

    def test_interrupt_command_registered(self):
        assert "name:'interrupt'" in COMMANDS_JS and "fn:cmdInterrupt" in COMMANDS_JS

    def test_steer_command_registered(self):
        assert "name:'steer'" in COMMANDS_JS and "fn:cmdSteer" in COMMANDS_JS

    def test_all_three_busy_commands_are_no_echo(self):
        """All three busy commands must set noEcho:true so the slash invocation
        is not echoed as a visible user bubble.  Without noEcho, /queue causes a
        double-bubble: the raw slash text appears, then the queued message appears
        again when the drain fires.
        """
        for name in ("queue", "interrupt", "steer"):
            idx = COMMANDS_JS.find(f"name:'{name}'")
            assert idx >= 0, f"{name} not registered"
            block = COMMANDS_JS[idx:idx + 250]
            assert "noEcho:true" in block, (
                f"/{name} registration must set noEcho:true — "
                "without it the command text is echoed as a user bubble, causing duplicates"
            )


class TestSlashCommandHandlers:
    """The three handler functions must guard properly and call cancelStream where appropriate."""

    def test_cmd_queue_requires_busy(self):
        """/queue while not busy is a usage error — user should send normally."""
        idx = COMMANDS_JS.find("async function cmdQueue(")
        assert idx >= 0
        body = COMMANDS_JS[idx:idx + 600]
        assert "if(!S.busy)" in body, "/queue must check S.busy and reject if idle"

    def test_cmd_interrupt_calls_cancel_stream(self):
        idx = COMMANDS_JS.find("async function cmdInterrupt(")
        assert idx >= 0
        body = COMMANDS_JS[idx:idx + 800]
        assert "queueSessionMessage" in body, "/interrupt must queue the new message before cancelling"
        assert "cancelStream" in body, "/interrupt must call cancelStream() so the drain re-sends"

    def test_cmd_steer_delegates_to_try_steer(self):
        """/steer delegates to _trySteer which calls /api/chat/steer with
        a queue+cancel fallback. The fallback path is exercised by tests
        in test_real_steer.py — this test just pins the delegation."""
        idx = COMMANDS_JS.find("async function cmdSteer(")
        assert idx >= 0
        body = COMMANDS_JS[idx:idx + 800]
        # cmdSteer now delegates to _trySteer; the fallback (queueSessionMessage
        # + cancelStream) lives inside _trySteer.
        assert "_trySteer" in body, "cmdSteer must call _trySteer to use the real /api/chat/steer endpoint"
        # The shared helper must contain the fallback path
        helper_idx = COMMANDS_JS.find("async function _trySteer(")
        assert helper_idx >= 0, "_trySteer helper must exist"
        helper_body = COMMANDS_JS[helper_idx:helper_idx + 1500]
        assert "queueSessionMessage" in helper_body
        assert "cancelStream" in helper_body
        # Toast should differ from interrupt to signal it's the steer path
        assert "cmd_steer_fallback" in helper_body or "steer_fallback" in helper_body


# ── send() busy branch ───────────────────────────────────────────────────

    def test_slash_commands_clear_pending_files(self):
        """All three busy command handlers must clear S.pendingFiles (directly
        or via _trySteer) after enqueuing, so staged files are not duplicated.

        cmdQueue and cmdInterrupt call queueSessionMessage themselves and clear
        S.pendingFiles directly.  cmdSteer delegates to _trySteer.  The fallback/interrupt path clears
        S.pendingFiles inside _trySteer; the success path returns early and
        send() handles the post-await clear.  Either way files are not
        duplicated — we verify by checking _trySteer body for the clearing.
        """
        # cmdQueue and cmdInterrupt clear pendingFiles directly
        for fn_name in ("cmdQueue", "cmdInterrupt"):
            idx = COMMANDS_JS.find(f"function {fn_name}(")
            assert idx >= 0, f"{fn_name} not found"
            body = COMMANDS_JS[idx:idx + 800]
            assert "S.pendingFiles=[]" in body, (
                f"{fn_name} must clear S.pendingFiles after queueSessionMessage"
            )
            assert "renderTray()" in body, (
                f"{fn_name} must call renderTray() after clearing pendingFiles"
            )
        # cmdSteer delegates to _trySteer; that helper clears pendingFiles
        idx_try = COMMANDS_JS.find("function _trySteer(")
        assert idx_try >= 0, "_trySteer not found"
        try_body = COMMANDS_JS[idx_try:idx_try + 1200]
        assert "S.pendingFiles=[]" in try_body, (
            "_trySteer must clear S.pendingFiles in its fallback path — "
            "without this, files are lost on steer→interrupt fallback"
        )
        assert "renderTray()" in try_body, (
            "_trySteer must call renderTray() after clearing pendingFiles"
        )


class TestSendBusyBranchDispatch:
    """send()'s busy block must read window._busyInputMode and branch accordingly."""

    def test_send_reads_busy_input_mode(self):
        # The send() function should read window._busyInputMode in the busy block
        send_idx = MESSAGES_JS.find("async function send(")
        assert send_idx >= 0
        # Look in the first ~3000 chars of send() for the busy mode read
        send_body = MESSAGES_JS[send_idx:send_idx + 3000]
        assert "_busyInputMode" in send_body, (
            "send() must read window._busyInputMode in the S.busy branch"
        )

    def test_send_calls_cancel_stream_on_interrupt(self):
        send_idx = MESSAGES_JS.find("async function send(")
        send_body = MESSAGES_JS[send_idx:send_idx + 3000]
        # The interrupt branch must call cancelStream
        assert "cancelStream" in send_body
        # And queue before cancel (otherwise the drain has nothing to pick up)
        # Verify the order textually: queueSessionMessage appears before cancelStream
        # within the busy block's interrupt branch
        cancel_idx = send_body.find("cancelStream")
        queue_idx = send_body.find("queueSessionMessage")
        assert queue_idx >= 0 and cancel_idx >= 0
        assert queue_idx < cancel_idx, (
            "queueSessionMessage must run before cancelStream so the drain "
            "after setBusy(false) picks up the queued message"
        )


    def test_slash_commands_intercepted_before_busymode_routing(self):
        """The three busy-control slash commands (/steer /interrupt /queue) must be
        intercepted at the TOP of the busy block — before the busyMode routing — so
        they execute immediately while the agent is running.

        Without this intercept, typing /steer while busy queues the text as a plain
        message.  When it drains after the turn ends there is no active stream, so
        cmdSteer says "No active task to stop." and the steer is lost entirely.
        """
        send_idx = MESSAGES_JS.find("async function send(")
        assert send_idx >= 0, "send() not found"
        # Look in the first 500 chars of the busy block for the intercept
        busy_start = MESSAGES_JS.find("S.busy||compressionRunning", send_idx)
        assert busy_start >= 0, "busy block not found"
        # The intercept must appear BEFORE the busyMode assignment
        intercept_idx = MESSAGES_JS.find("'steer','interrupt','queue'", busy_start)
        busymode_idx = MESSAGES_JS.find("_busyInputMode||'queue'", busy_start)
        assert intercept_idx >= 0, (
            "send() must intercept /steer /interrupt /queue before the busyMode "
            "routing block — otherwise they queue instead of executing immediately"
        )
        assert intercept_idx < busymode_idx, (
            "The slash-command intercept must come BEFORE the busyMode routing "
            "so /steer executes while the agent is running, not after the turn ends"
        )

    def test_steer_intercept_calls_handler_directly(self):
        """The busy-intercept must dispatch via _bc.fn(_pc.args), not queue the text."""
        send_idx = MESSAGES_JS.find("async function send(")
        busy_start = MESSAGES_JS.find("S.busy||compressionRunning", send_idx)
        intercept_idx = MESSAGES_JS.find("'steer','interrupt','queue'", busy_start)
        assert intercept_idx >= 0
        # Get the intercept block (up to the next busyMode assignment)
        busymode_idx = MESSAGES_JS.find("_busyInputMode||'queue'", busy_start)
        intercept_block = MESSAGES_JS[intercept_idx:busymode_idx]
        assert "_bc.fn(_pc.args)" in intercept_block, (
            "The intercept must call the command handler directly via _bc.fn(_pc.args)"
        )
        assert "return;" in intercept_block, (
            "The intercept must return after dispatching so send() does not also queue"
        )

    def test_steer_intercept_clears_input_before_await(self):
        """The intercept must clear $('msg').value BEFORE awaiting the handler.

        Without the sync clear, the input field still shows '/steer foo' after
        the steer fires. If the user presses Enter again (a common reflex while
        waiting for the toast), send() re-runs and either re-fires the command
        or — once the turn ended — drops a confusing 'No active task to stop.'
        """
        send_idx = MESSAGES_JS.find("async function send(")
        busy_start = MESSAGES_JS.find("S.busy||compressionRunning", send_idx)
        intercept_idx = MESSAGES_JS.find("'steer','interrupt','queue'", busy_start)
        busymode_idx = MESSAGES_JS.find("_busyInputMode||'queue'", busy_start)
        intercept_block = MESSAGES_JS[intercept_idx:busymode_idx]
        clear_idx = intercept_block.find("$('msg').value=''")
        await_idx = intercept_block.find("await _bc.fn")
        assert clear_idx >= 0, (
            "The intercept must clear $('msg').value (so the field doesn't keep "
            "showing /steer foo after the command fires)"
        )
        assert await_idx >= 0, "await _bc.fn(...) must be present in the intercept"
        assert clear_idx < await_idx, (
            "$('msg').value='' must be cleared BEFORE awaiting the handler — "
            "otherwise a reflexive Enter press during the await re-fires the command"
        )


# ── Boot init + settings panel wiring ───────────────────────────────────

class TestBootAndPanelsWiring:
    def test_boot_init_default_path(self):
        """Boot success path initialises window._busyInputMode from settings."""
        assert "window._busyInputMode=(s.busy_input_mode||'queue')" in BOOT_JS

    def test_boot_init_fallback_path(self):
        """Boot fallback path (settings load failed) initialises to safe default."""
        # The fallback should set window._busyInputMode='queue'
        assert "window._busyInputMode='queue'" in BOOT_JS

    def test_panels_load_save_apply(self):
        assert "settingsBusyInputMode" in PANELS_JS, "panels.js must load the setting"
        assert "body.busy_input_mode" in PANELS_JS, "saveSettings must include busy_input_mode in body"
        assert "window._busyInputMode=body.busy_input_mode" in PANELS_JS, (
            "_applySavedSettingsUi must propagate busy_input_mode to the global"
        )

    def test_index_html_dropdown_has_three_options(self):
        idx = INDEX_HTML.find('id="settingsBusyInputMode"')
        assert idx >= 0
        block = INDEX_HTML[idx:idx + 800]
        assert 'value="queue"' in block
        assert 'value="interrupt"' in block
        assert 'value="steer"' in block


# ── i18n locale coverage ─────────────────────────────────────────────────

class TestI18nKeys:
    """All 17 new keys must appear in each of the 6 locale blocks."""

    REQUIRED_KEYS = [
        "cmd_queue",
        "cmd_interrupt",
        "cmd_steer",
        "cmd_queue_no_msg",
        "cmd_queue_not_busy",
        "cmd_queue_confirm",
        "cmd_interrupt_no_msg",
        "cmd_interrupt_confirm",
        "cmd_steer_no_msg",
        "cmd_steer_fallback",
        "busy_steer_fallback",
        "busy_interrupt_confirm",
        "settings_label_busy_input_mode",
        "settings_desc_busy_input_mode",
        "settings_busy_input_mode_queue",
        "settings_busy_input_mode_interrupt",
        "settings_busy_input_mode_steer",
    ]

    def test_each_key_appears_at_least_six_times(self):
        """Each key should appear once per locale (en, ru, es, de, zh, zh-Hant) = 6 occurrences minimum."""
        for key in self.REQUIRED_KEYS:
            count = I18N_JS.count(f"{key}:")
            assert count >= 6, (
                f"i18n key {key!r} appears {count} times; expected ≥6 (one per locale block)"
            )

    def test_key_count_total(self):
        """17 keys × 6 locales = 102 minimum occurrences across the file."""
        total = sum(I18N_JS.count(f"{key}:") for key in self.REQUIRED_KEYS)
        assert total >= 17 * 6, (
            f"Total i18n occurrences = {total}; expected ≥ {17*6}"
        )