"""Installed identity proofs at inert native supervisor executable boundaries."""
import json
import os
from pathlib import Path
import subprocess
import sys
import time

import pytest


@pytest.mark.platforms("linux")
@pytest.mark.parametrize("case,reason", [
    ("custom", "deadline"), ("named", "deadline"), ("alias", "deadline"),
    ("default", "deadline"),
    ("wrong_home", "profile_mismatch"), ("wrong_user", "service_account_mismatch"),
    ("missing_identity", "service_identity_unverified"),
    ("omitted_empty_exec_lists", "deadline"),
    ("env_file", "service_identity_unverified"), ("dynamic_user", "service_identity_unverified"),
    ("wrong_command", "service_identity_unverified"), ("command_profile", "profile_mismatch"),
    ("dropin_home", "profile_mismatch"), ("manager_home", "profile_mismatch"),
])
def test_ensure_checks_effective_service_binding_before_start(tmp_path, monkeypatch, case, reason):
    from hermes_cli import gateway_runtime as runtime
    from hermes_cli.gateway_runtime_service import service_suffix

    monkeypatch.setattr(Path, "home", lambda: tmp_path)
    home = tmp_path / (".hermes" if case == "default" else "custom root")
    if case == "named":
        home = home / "profiles" / "worker"
    home.mkdir(mode=0o700, parents=True)
    monkeypatch.setenv("HERMES_HOME", str(home))
    suffix = service_suffix(home)
    unit = tmp_path / ".config/systemd/user" / f"hermes-gateway{'-' + suffix if suffix else ''}.service"
    unit.parent.mkdir(parents=True)
    unit.write_text(f'[Service]\nExecStart={sys.executable} -m hermes_cli.main gateway run\nEnvironment="HERMES_HOME={home}"\n', encoding="utf-8")
    original = unit.read_bytes()
    configured = home
    if case in {"wrong_home", "dropin_home"}:
        configured = tmp_path / "unrequested"
    if case == "wrong_home":
        unit.write_text(f'[Service]\nEnvironment="HERMES_HOME={configured}"\n', encoding="utf-8")
        original = unit.read_bytes()
    if case == "dropin_home":
        override = unit.parent / (unit.name + '.d') / 'override.conf'
        override.parent.mkdir()
        override.write_text(f'[Service]\nEnvironment="HERMES_HOME={configured}"\n', encoding="utf-8")
    if case == "alias":
        configured = tmp_path / "alias"
        configured.symlink_to(home, target_is_directory=True)
    command = f'{sys.executable} -m hermes_cli.main gateway run'
    if case == "named":
        command += " --profile worker"
    if case == "command_profile":
        command += " --profile stranger"
    if case == "wrong_command":
        command = '/bin/echo hermes_cli.main gateway run'
    props = dict(LoadState="loaded", ActiveState="inactive", SubState="dead", UnitFileState="enabled",
                 User=str(os.getuid()), DynamicUser="no", Environment=f'"HERMES_HOME={configured}" "HERMES_SUPERVISED_CHILD=1"',
                 EnvironmentFiles="", PassEnvironment="", UnsetEnvironment="", PAMName="",
                 RootDirectory="", RootImage="", ExecStartPre="", ExecCondition="",
                 ExecStart=f'{{ path={command.split()[0]} ; argv[]={command} ; ignore_errors=no ; }}',
                 DropInPaths=str(unit.parent / (unit.name + '.d') / 'override.conf') if case == "dropin_home" else "")
    manager_env = ""
    if case == "wrong_user":
        props['User'] = str(os.getuid() + 1)
    if case == "missing_identity":
        props = {k: props[k] for k in ("LoadState", "ActiveState", "SubState", "UnitFileState")}
    if case == "omitted_empty_exec_lists":
        # systemd >= 25x prints nothing for empty exec-command lists and
        # EnvironmentFiles even under --all; absence there means empty, not unknown.
        for key in ("ExecStartPre", "ExecCondition", "EnvironmentFiles"):
            del props[key]
    if case == "env_file":
        env_file = tmp_path / "override.env"
        env_file.write_text(f'HERMES_HOME={tmp_path / "other"}\n', encoding="utf-8")
        props['EnvironmentFiles'] = f'{env_file} (ignore_errors=no)'
    if case == "dynamic_user":
        props['DynamicUser'] = 'yes'
    if case == "manager_home":
        props['Environment'] = 'HERMES_SUPERVISED_CHILD=1'
        manager_env = f'HERMES_HOME={tmp_path / "other"}\n'
    if case == "default":
        # An older default install may rely on the user manager's HOME.
        props['Environment'] = 'HERMES_SUPERVISED_CHILD=1'
        manager_env = f'HOME={tmp_path}\n'
    definition = tmp_path / "effective.json"
    definition.write_text(json.dumps({'props': props, 'system': False, 'env': manager_env}), encoding="utf-8")
    calls = tmp_path / 'calls.jsonl'
    executable = tmp_path / 'inert-supervisor'
    executable.write_text('#!' + sys.executable + '\nimport json,sys\nfrom pathlib import Path\n'
        + f'p=Path({str(tmp_path)!r})\nd=json.loads((p/"effective.json").read_text())\n'
        + 'with (p/"calls.jsonl").open("a") as f: f.write(json.dumps(sys.argv[1:])+"\\n")\n'
        + 'if "show-environment" in sys.argv: print(d["env"])\n'
        + 'elif "show" in sys.argv:\n'
        + ' if ("--user" not in sys.argv) == d["system"]: print("\\n".join(k+"="+v for k,v in d["props"].items()))\n'
        + ' else: print("LoadState=not-found")\n'
        + 'elif "start" not in sys.argv: sys.exit(91)\n', encoding='utf-8')
    executable.chmod(0o700)
    monkeypatch.setenv('PATH', str(tmp_path) + os.pathsep + os.environ['PATH'])
    from hermes_cli import gateway as gw
    monkeypatch.setattr(gw, "_systemctl_cmd", lambda system=False: [str(executable)] + ([] if system else ["--user"]))
    result = runtime.ensure_gateway_runtime(home, timeout=3.0)
    assert result.reason_code == reason
    commands = [json.loads(line) for line in calls.read_text().splitlines()]
    starts = [cmd for cmd in commands if 'start' in cmd]
    assert len(starts) == (1 if reason == 'deadline' else 0)
    assert all('--no-ask-password' in cmd for cmd in starts)
    assert unit.read_bytes() == original
    assert not (home / 'logs').exists()


@pytest.mark.parametrize("configured,command,reason", [
    ("requested", "gateway", None), ("other", "gateway", "profile_mismatch"),
    ("requested", "echo", "service_identity_unverified"),
    (None, "gateway", "service_identity_unverified"),
    ("requested", "unsupervised", "service_identity_unverified"),
    ("requested", "wrong_account", "service_account_mismatch"),
])
def test_loaded_launchd_identity_is_independent_of_disk(configured, command, reason, tmp_path):
    from hermes_cli.gateway_runtime_service_identity import verify_launchd_loaded
    home = tmp_path / "requested"
    argv = [sys.executable, "-m", "hermes_cli.main", "gateway", "run"]
    if command == "echo":
        argv = ["/bin/echo", "hermes_cli.main", "gateway", "run"]
    environment = f"HERMES_HOME => {tmp_path / configured}" if configured else ""
    output = "gui/123/ai.hermes.gateway = {\n\tstate = not running\n\tprogram = " + argv[0] + "\n\targuments = {\n" + "\n".join("\t\t" + a for a in argv) + "\n\t}\n\tenvironment = {\n\t\t" + environment + "\n\t}\n}"
    if command != "unsupervised":
        output = output.replace("\n\tenvironment = {\n", "\n\tenvironment = {\n\t\tHERMES_SUPERVISED_CHILD => 1\n")
    if command == "wrong_account":
        output = output.replace("\n\tstate", "\n\tusername = other\n\tstate")
    if reason:
        with pytest.raises(ValueError, match=reason):
            verify_launchd_loaded(output, home, uid=123, username="owner")
    else:
        verify_launchd_loaded(output, home, uid=123, username="owner")


def _launchd_gateway_command(tmp_path, monkeypatch, *, store, profile=""):
    """The gateway command generate_launchd_plist persists, from a store Python or an external one."""
    from hermes_cli import _launchers, gateway as gw
    from hermes_cli.gateway_launchd import _timestamped_stderr_gateway_command
    monkeypatch.setattr(gw, "PROJECT_ROOT", tmp_path / "hermes-agent")
    monkeypatch.setattr(gw, "get_python_path", lambda: sys.executable)
    monkeypatch.setattr(gw, "_profile_arg", lambda *a, **k: profile)
    monkeypatch.setattr(_launchers, "resolve_store_python", lambda *a, **k: Path(sys.executable) if store else None)
    return _timestamped_stderr_gateway_command(tmp_path / "logs" / "gateway.error.log", external_supervisor=True)


@pytest.mark.platforms("linux", "macos")  # launchd ProgramArguments are POSIX shell-quoted
@pytest.mark.parametrize("case,reason", [
    ("store", None), ("inline", None), ("store_profile", "profile_mismatch"),
    ("chained_shell", "service_identity_unverified"), ("substituted_shell", "service_identity_unverified"),
    ("extra_jxa", "service_identity_unverified"), ("foreign_osascript", "service_identity_unverified"),
    ("tampered_bootstrap", "service_identity_unverified"), ("arbitrary_inline", "service_identity_unverified"),
    ("other_run_module", "service_identity_unverified"),
    ("foreign_root_store", "service_identity_unverified"), ("foreign_root_inline", "service_identity_unverified"),
    ("other_home", "foreign_root"), ("osascript_non_gateway", "service_identity_unverified"),
    ("own_wrapper_foreign_inner", "service_identity_unverified"),
    ("osascript_foreign_bare", "service_identity_unverified"),
    ("osascript_legacy_module", "service_identity_unverified"),
])
def test_launchd_verifies_the_osascript_wrapped_command_the_installer_writes(case, reason, tmp_path, monkeypatch):
    """The Local Network wrapper (#71206) is unwrapped only as its exact template, and the command inside
    is held to the same gateway identity as a bare ProgramArguments, on disk and as launchd loaded it."""
    import plistlib
    from hermes_cli.gateway_launchd import launchd_program_arguments
    from hermes_cli.gateway_runtime_service_identity import verify_launchd_loaded, verify_launchd_plist
    home = tmp_path / ".hermes"
    from hermes_cli import gateway as gw
    command = _launchd_gateway_command(tmp_path, monkeypatch,
                                       store=case not in {"inline", "tampered_bootstrap", "foreign_root_inline"},
                                       profile="--profile stranger" if case == "store_profile" else "")
    if case.startswith("foreign_root"):  # the installer's own output, written for another checkout
        from hermes_cli.gateway_launchd import _timestamped_stderr_gateway_command
        root = gw.PROJECT_ROOT
        monkeypatch.setattr(gw, "PROJECT_ROOT", tmp_path / "other-checkout")
        command = _timestamped_stderr_gateway_command(tmp_path / "logs" / "gateway.error.log", external_supervisor=True)
        monkeypatch.setattr(gw, "PROJECT_ROOT", root)
    if case in {"own_wrapper_foreign_inner", "osascript_foreign_bare"}:
        # The installer's own gateway command for ANOTHER checkout: inside OUR stderr wrapper, or bare
        # inside the Local Network wrapper. Neither is a launcher this install root writes.
        from hermes_cli._launchers import installation_command
        foreign = installation_command(tmp_path / "other-checkout", ["gateway", "run", "--external-supervisor"],
                                       python=sys.executable)
        assert Path(foreign[0]).name == "hermes"
        command = foreign if case == "osascript_foreign_bare" else installation_command(
            gw.PROJECT_ROOT, ["--error-log", str(tmp_path / "logs" / "gateway.error.log"), "--", *foreign],
            module="hermes_cli.stderr_timestamp", python=sys.executable)
    if case == "osascript_legacy_module":
        # The legacy ``python -m`` form from any interpreter: accepted unwrapped, never written inside osascript.
        command = [sys.executable, "-m", "hermes_cli.stderr_timestamp", "--error-log",
                   str(tmp_path / "logs" / "gateway.error.log"), "--",
                   sys.executable, "-m", "hermes_cli.main", "gateway", "run", "--external-supervisor"]
    if case == "osascript_non_gateway":
        command = ["/bin/sh", "-c", "id"]
    if case == "tampered_bootstrap":
        tampered = command[3].replace("import hermes_bootstrap; ", "import hermes_bootstrap; print('injected'); ")
        assert tampered != command[3]
        command[3] = tampered
    if case == "arbitrary_inline":
        command = [sys.executable, "-I", "-c", "print('injected')", *command[3:]]
    if case == "other_run_module":
        command[2] = "other_module"
    argv = launchd_program_arguments(command, tmp_path / "logs" / "gateway.log", tmp_path / "logs" / "gateway.error.log")
    shell_edits = {"chained_shell": (" >> ", "; id >> "), "substituted_shell": (" gateway run", " $(id) gateway run")}
    if case in shell_edits:
        edited = argv[4].replace(*shell_edits[case], 1)
        assert edited != argv[4]
        argv[4] = edited
    if case == "extra_jxa":
        argv[4] += ' $.system("id");'
    if case == "foreign_osascript":
        argv[0] = str(tmp_path / "osascript")
    env = dict(HERMES_HOME=str(home if case != "other_home" else tmp_path / "elsewhere" / ".hermes"),
               HERMES_SUPERVISED_CHILD="1")
    definition = plistlib.loads(plistlib.dumps(dict(Label="ai.hermes.gateway", ProgramArguments=argv, EnvironmentVariables=env)))
    output = ("gui/123/ai.hermes.gateway = {\n\tstate = not running\n\tprogram = " + argv[0] + "\n\targuments = {\n"
              + "\n".join("\t\t" + arg for arg in argv) + "\n\t}\n\tenvironment = {\n"
              + "".join(f"\t\t{key} => {value}\n" for key, value in env.items()) + "\t}\n}")
    for verify in (lambda: verify_launchd_plist(definition, "ai.hermes.gateway", home),
                   lambda: verify_launchd_loaded(output, home, uid=123, username="owner")):
        if reason:
            with pytest.raises(ValueError, match=reason):
                verify()
        else:
            verify()


def _installed_vbs(home, configured):
    """The current installed launcher grammar; no script is executed."""
    script = home / 'gateway.vbs'
    script.write_text("\n".join([
        "' Hermes Agent Gateway", "Option Explicit", "Dim sh, env, existing_pp",
        'Set sh = CreateObject("WScript.Shell")', 'Set env = sh.Environment("PROCESS")',
        f'env.Item("HERMES_HOME") = "{configured}"',
        'env.Item("HERMES_SUPERVISED_CHILD") = "1"',
        'env.Item("VIRTUAL_ENV") = "C:\\Hermes"',
        'existing_pp = env.Item("PYTHONPATH")', 'If Len(existing_pp) > 0 Then',
        '  env.Item("PYTHONPATH") = "C:\\Hermes;" & existing_pp', 'Else',
        '  env.Item("PYTHONPATH") = "C:\\Hermes"', 'End If',
        f'sh.CurrentDirectory = "{home}"',
        f'sh.Run "{sys.executable} -m hermes_cli.main gateway run", 0, False',
    ]) + "\n", encoding='utf-8')
    return script


def _as_schtasks_exports(xml):
    """``schtasks /Query /XML`` omits elements holding the Windows default (seen on a real host)."""
    return xml.replace('<RunLevel>LeastPrivilege</RunLevel>', '').replace('<Enabled>true</Enabled>\n    <Hidden>', '<Hidden>')


@pytest.mark.parametrize("case,reason", [
    ("bound", None), ("utf16", None), ("wrong_home", "profile_mismatch"),
    ("wrong_user", "service_account_mismatch"), ("missing_user", "service_identity_unverified"),
    ("extra_action", "service_identity_unverified"), ("disabled", "service_disabled"),
    ("unknown_script", "service_identity_unverified"), ("malformed_xml", "service_identity_unverified"),
    ("schtasks_export", None), ("elevated", "service_identity_unverified"),
    ("empty_run_level", "service_identity_unverified"), ("empty_enabled", "service_disabled"),
])
def test_task_xml_binds_actual_action_and_principal(case, reason, tmp_path):
    from hermes_cli.gateway_runtime_service_identity import verify_windows_task
    from hermes_cli.gateway_windows import _build_scheduled_task_xml
    home = tmp_path / 'requested'
    home.mkdir()
    script = _installed_vbs(home, str(home if case != 'wrong_home' else tmp_path / 'other'))
    user = 'DOMAIN\\owner'
    xml = _build_scheduled_task_xml('Hermes_Gateway', script, user)
    if case == 'schtasks_export':
        xml = _as_schtasks_exports(xml)
    if case == 'elevated':  # an explicit non-default value is still refused
        xml = xml.replace('<RunLevel>LeastPrivilege</RunLevel>', '<RunLevel>HighestAvailable</RunLevel>')
    if case == 'empty_run_level':  # present but empty is not "absent": still refused
        xml = xml.replace('<RunLevel>LeastPrivilege</RunLevel>', '<RunLevel></RunLevel>')
    if case == 'empty_enabled':
        xml = xml.replace('<Enabled>true</Enabled>\n    <Hidden>', '<Enabled></Enabled>\n    <Hidden>')
    if case == 'wrong_user':
        xml = xml.replace(user, 'DOMAIN\\other')
    if case == 'missing_user':
        xml = xml.replace(f'<UserId>{user}</UserId>', '')
    if case == 'extra_action':
        xml = xml.replace('</Actions>', '<Exec><Command>other.exe</Command></Exec></Actions>')
    if case == 'disabled':
        xml = xml.replace('<Enabled>true</Enabled>', '<Enabled>false</Enabled>')
    if case == 'utf16':
        xml = xml.encode('utf-16')
    if case == 'malformed_xml':
        xml = '<Task>'
    if case == 'unknown_script':
        script.write_text('MsgBox "not a gateway"', encoding='utf-8')
    if reason:
        with pytest.raises(ValueError, match=reason):
            verify_windows_task(xml, home, user, 'S-1-5-21-123')
    else:
        verify_windows_task(xml, home, user, 'S-1-5-21-123')


@pytest.mark.platforms("macos")
@pytest.mark.parametrize("case,reason", [("loaded_good_disk_wrong", None), ("loaded_wrong_disk_good", "profile_mismatch"), ("bootstrap", None)])
def test_native_launchd_checks_loaded_job_before_disk(case, reason, tmp_path, monkeypatch):
    import plistlib
    import pwd
    from types import SimpleNamespace
    from hermes_cli import gateway_runtime_service as service
    home = tmp_path / 'profile'
    home.mkdir(mode=0o700)
    monkeypatch.setenv('HERMES_HOME', str(home))
    monkeypatch.setattr(pwd, 'getpwuid', lambda uid: SimpleNamespace(pw_dir=str(tmp_path), pw_name="owner"))
    label = 'ai.hermes.gateway' + ('-' + service.service_suffix(home) if service.service_suffix(home) else '')
    plist = tmp_path / 'Library/LaunchAgents' / (label + '.plist')
    plist.parent.mkdir(parents=True)
    argv = [sys.executable, '-m', 'hermes_cli.main', 'gateway', 'run']
    disk_home = tmp_path / 'wrong' if case == 'loaded_good_disk_wrong' else home
    plist.write_bytes(plistlib.dumps(dict(Label=label, ProgramArguments=argv,
        EnvironmentVariables=dict(HERMES_HOME=str(disk_home), HERMES_SUPERVISED_CHILD='1'))))
    original = plist.read_bytes()
    loaded_home = tmp_path / 'wrong' if case == 'loaded_wrong_disk_good' else home
    output = 'gui/' + str(os.getuid()) + '/' + label + ' = {\n\tstate = not running\n\tprogram = ' + argv[0] + '\n\targuments = {\n' + '\n'.join('\t\t' + arg for arg in argv) + '\n\t}\n\tenvironment = {\n\t\tHERMES_HOME => ' + str(loaded_home) + '\n\t\tHERMES_SUPERVISED_CHILD => 1\n\t}\n}'
    peer = tmp_path / 'supervisor.py'
    peer.write_text('import sys\n' + f'output={output!r}\nbootstrap={case == "bootstrap"!r}\n' +
        'if "print" in sys.argv:\n sys.exit(113) if bootstrap or any(a.startswith("user/") for a in sys.argv) else print(output)\n', encoding='utf-8')
    actual = subprocess.run
    calls = []
    def boundary(argv, **kw):
        assert argv[0] == 'launchctl'
        calls.append(argv)
        return actual([sys.executable, str(peer), *argv[1:]], **kw)
    monkeypatch.setattr(subprocess, 'run', boundary)
    if reason:
        with pytest.raises(service.RuntimeStartError, match=reason):
            service.discover_existing_gateway_service(home, deadline=time.monotonic()+5)
        assert all(a[1] == 'print' for a in calls)
    else:
        found = service.discover_existing_gateway_service(home, deadline=time.monotonic()+5)
        assert found is not None
        service.start_existing_gateway_service(found, deadline=time.monotonic()+5)
        assert calls[-1][1] == ('bootstrap' if case == 'bootstrap' else 'kickstart')
    assert plist.read_bytes() == original


@pytest.mark.platforms("windows")
@pytest.mark.parametrize("wrong", [False, True])
def test_native_task_query_uses_installed_xml_and_vendor_launcher(wrong, tmp_path, monkeypatch):
    from hermes_cli import gateway_runtime_service as service
    from hermes_cli.gateway_windows import _build_gateway_vbs_script, _build_scheduled_task_xml
    home = tmp_path / 'profile'
    home.mkdir()
    monkeypatch.setenv('HERMES_HOME', str(home))
    script = home / 'gateway.vbs'
    # newline='' mirrors the installer's _atomic_write: the template is already CRLF and
    # Windows text mode would otherwise double it to \r\r\n, which is not the vendor template.
    script.write_text(_build_gateway_vbs_script(sys.executable, str(home), str(home), ''), encoding='utf-8', newline='')
    original = script.read_bytes()
    # whoami is read-only native account evidence; no task-manager mutation.
    identity = subprocess.run(['whoami.exe', '/USER', '/FO', 'CSV', '/NH'], stdin=subprocess.DEVNULL, capture_output=True, timeout=5, check=False)
    from hermes_cli.gateway_windows import _schtasks_encoding
    import csv
    _account, sid = next(csv.reader(identity.stdout.decode(_schtasks_encoding()).splitlines()))
    suffix = service.service_suffix(home)
    name = 'Hermes_Gateway' + ('_' + suffix if suffix else '')
    xml = _build_scheduled_task_xml(name, script, 'S-1-5-18' if wrong else sid)
    peer = tmp_path / 'supervisor.py'
    row = f'"\\{name}","N/A","Ready"'
    peer.write_text('import sys\n' + f'xml={xml!r}\n' +
        'if "/XML" in sys.argv: sys.stdout.buffer.write(xml.encode("utf-16"))\n'
        + f'elif "/Query" in sys.argv: print({row!r})\n', encoding='utf-8')
    actual = subprocess.run
    calls = []
    def boundary(argv, **kw):
        if argv[0] == 'whoami.exe':
            return actual(argv, **kw)
        assert argv[0] == 'schtasks.exe'
        calls.append(argv)
        return actual([sys.executable, str(peer), *argv[1:]], **kw)
    monkeypatch.setattr(subprocess, 'run', boundary)
    if wrong:
        with pytest.raises(service.RuntimeStartError, match='service_account_mismatch'):
            service.discover_existing_gateway_service(home, deadline=time.monotonic()+5)
        assert not any('/Run' in a for a in calls)
    else:
        found = service.discover_existing_gateway_service(home, deadline=time.monotonic()+5)
        assert found is not None
        service.start_existing_gateway_service(found, deadline=time.monotonic()+5)
        assert sum('/Run' in a for a in calls) == 1
    assert script.read_bytes() == original


@pytest.mark.platforms("linux")
def test_service_definition_reads_reject_fifo_without_waiting(tmp_path):
    from hermes_cli.gateway_runtime_service_identity import read_definition
    regular = tmp_path / 'regular'
    regular.write_bytes(b'installed definition')
    assert read_definition(regular) == regular.read_bytes()
    fifo = tmp_path / 'fifo'
    os.mkfifo(fifo, 0o600)
    with pytest.raises(ValueError, match='service_identity_unverified'):
        read_definition(fifo)


@pytest.mark.parametrize("outer", ["python", "echo"])
def test_gateway_wrapper_requires_actual_python_entrypoint(outer, tmp_path):
    from hermes_cli.gateway_runtime_service_identity import verify_gateway_argv
    argv = [sys.executable if outer == 'python' else '/bin/echo', '-m', 'hermes_cli.stderr_timestamp', '--error-log', str(tmp_path / 'error.log'), '--', sys.executable, '-m', 'hermes_cli.main', 'gateway', 'run']
    if outer == 'echo':
        with pytest.raises(ValueError, match='service_identity_unverified'):
            verify_gateway_argv(argv, tmp_path)
    else:
        verify_gateway_argv(argv, tmp_path)


def test_android_python_reaches_the_unmanaged_start(tmp_path, monkeypatch):
    """Termux is supported and has no service manager; Python 3.13+ there reports
    sys.platform == "android". Ensure must start the gateway unmanaged, not refuse the platform."""
    from hermes_cli import gateway_runtime as runtime, gateway_runtime_service as service, gateway_runtime_start as start

    home = tmp_path / "profile"
    home.mkdir(mode=0o700)
    spawned = []
    monkeypatch.setattr(service.sys, "platform", "android")
    monkeypatch.setattr(service, "_run", lambda *a, **k: pytest.fail("no supervisor to query on Android"))
    monkeypatch.setattr(runtime, "discover_gateway_endpoint", lambda *a, **k: runtime.GatewayDiscovery("absent"))
    monkeypatch.setattr(start, "spawn_unmanaged_gateway", lambda target, **k: spawned.append(Path(target)))
    result = runtime.ensure_gateway_runtime(home, timeout=0.3)
    assert result.reason_code != "unsupported_supervisor_platform"
    assert spawned == [home.resolve()]


def test_s6_runtime_ensure_starts_the_registered_gateway_slot(tmp_path, monkeypatch):
    """A Docker TUI client starts its s6 gateway instead of probing systemd."""
    from hermes_cli import gateway_runtime_service as service
    from hermes_cli import service_manager

    home = tmp_path / "data"
    home.mkdir()
    scandir = tmp_path / "run-service"
    (scandir / "gateway-default").mkdir(parents=True)
    started = []

    class FakeS6ServiceManager:
        def __init__(self):
            self.scandir = scandir

        def is_running(self, name):
            assert name == "gateway-default"
            return False

        def start(self, name):
            started.append(name)

    monkeypatch.setattr(service_manager, "_s6_running", lambda: True)
    monkeypatch.setattr(service_manager, "S6ServiceManager", FakeS6ServiceManager)
    read_text = Path.read_text
    monkeypatch.setattr(Path, "read_text", lambda path, *args, **kwargs: (
        str(home) if str(path).replace("\\", "/") == "/run/s6/container_environment/HERMES_HOME"
        else read_text(path, *args, **kwargs)))
    monkeypatch.setattr(service, "_systemd", lambda *_args: pytest.fail("s6 must not probe systemd"))

    found = service.discover_existing_gateway_service(home, deadline=time.monotonic() + 5)

    assert found is not None and found.backend == "s6"
    service.start_existing_gateway_service(found, deadline=time.monotonic() + 5)
    assert started == ["gateway-default"]


def test_s6_command_timeout_ends_at_the_runtime_deadline(tmp_path, monkeypatch):
    """A hung s6-svstat or s6-svc reaches ensure as its deadline, not an uncaught TimeoutExpired."""
    from hermes_cli import gateway_runtime_service as service
    from hermes_cli import service_manager

    home = tmp_path / "data"
    home.mkdir()
    scandir = tmp_path / "run-service"
    (scandir / "gateway-default").mkdir(parents=True)

    def hang(cmd, *_args, timeout=5, **_kwargs):
        raise subprocess.TimeoutExpired(cmd, timeout)

    monkeypatch.setattr(service_manager, "_s6_running", lambda: True)
    monkeypatch.setattr(service_manager, "_s6_run", hang)
    monkeypatch.setattr(service_manager.S6ServiceManager, "__init__",
                        lambda self: setattr(self, "scandir", scandir))
    read_text = Path.read_text
    monkeypatch.setattr(Path, "read_text", lambda path, *args, **kwargs: (
        str(home) if str(path).replace("\\", "/") == "/run/s6/container_environment/HERMES_HOME"
        else read_text(path, *args, **kwargs)))

    with pytest.raises(TimeoutError):
        service.discover_existing_gateway_service(home, deadline=time.monotonic() + 5)
    stopped = service.ExistingService("s6", ("gateway-default",), False)
    with pytest.raises(TimeoutError):
        service.start_existing_gateway_service(stopped, deadline=time.monotonic() + 5)


def test_windows_startup_folder_entry_still_restarts_a_stopped_owner(tmp_path, monkeypatch):
    """The Startup-folder fallback is login persistence only (no start verb, no identity): with it
    present and no Scheduled Task registered, ensure takes the same detached direct spawn manual
    `hermes gateway start` uses instead of refusing until the next login."""
    from types import SimpleNamespace
    from hermes_cli import gateway_runtime as runtime, gateway_runtime_service as service, gateway_runtime_start as start
    from hermes_cli import gateway_windows

    home = tmp_path / "profile"
    home.mkdir(mode=0o700)
    startup = tmp_path / "Startup"
    startup.mkdir()
    (startup / f"Hermes_Gateway_{service.service_suffix(home)}.vbs").write_text("' fallback\r\n")
    monkeypatch.setattr(gateway_windows, "_startup_dir", lambda: startup)
    queries, spawned = [], []
    monkeypatch.setattr(service, "_run", lambda argv, *a, **k: queries.append(argv) or SimpleNamespace(returncode=0, stdout=b""))
    monkeypatch.setattr(service.sys, "platform", "win32")
    monkeypatch.setattr(runtime, "discover_gateway_endpoint", lambda *a, **k: runtime.GatewayDiscovery("absent"))
    monkeypatch.setattr(start, "spawn_unmanaged_gateway", lambda target, **k: spawned.append(Path(target)))
    result = runtime.ensure_gateway_runtime(home, timeout=0.3)
    assert result.reason_code != "startup_service_requires_login"
    assert spawned == [home.resolve()]
    assert all(argv[:2] == ["schtasks.exe", "/Query"] for argv in queries)  # nothing installed or run


@pytest.mark.platforms("windows")
def test_task_discovery_reads_localized_schtasks_output(tmp_path, monkeypatch):
    """A localized host (zh-CN, console cp936) with Python in UTF-8 mode: schtasks and whoami write
    cp936 bytes, and /XML declares UTF-16 over them with the defaults elided (#139083, #139013).
    The installed task must still be verified, not refused as service_identity_unverified."""
    from types import SimpleNamespace
    from hermes_cli import gateway_runtime_service as service, gateway_windows
    from hermes_cli.gateway_windows import _build_scheduled_task_xml

    home = tmp_path / "profile"
    home.mkdir()
    script = _installed_vbs(home, str(home))
    name = "Hermes_Gateway_" + service.service_suffix(home)
    account, sid = "DESKTOP-1\\\u65b9\u821f", "S-1-5-21-123"
    # whoami first: it is queried with /FO CSV too.
    out = {"whoami.exe": f'"{account}","{sid}"\r\n', "/XML": _as_schtasks_exports(
        _build_scheduled_task_xml(name, script, account)), "/FO": f'"\\{name}","N/A","\u5c31\u7eea"\r\n'}
    # The console code page this host would report; the bytes below are what such a console writes.
    monkeypatch.setattr(gateway_windows, "_windows_console_encodings", lambda: ["cp936"])

    def run(argv, *_args, encoding=None, **_kwargs):
        # Windows subprocess decodes text in a reader thread: a decode error there leaves stdout None
        # (main then read an empty task list and fell through to an unmanaged spawn), it does not raise.
        data = next(text for key, text in out.items() if key in argv).encode("cp936")
        if encoding:
            try:
                data = data.decode(encoding)
            except UnicodeDecodeError:
                data = None
        return SimpleNamespace(returncode=0, stdout=data)

    monkeypatch.setattr(service, "_run", run)
    found = service.discover_existing_gateway_service(home, deadline=time.monotonic() + 5)
    assert found is not None and found.start_argv == ("schtasks.exe", "/Run", "/TN", name)