From 87a2e9bf2a67dcdcae718aaff0d7bb33bcf8ed25 Mon Sep 17 00:00:00 2001 From: unsupportedpastels Date: Sun, 25 Oct 2026 17:08:43 +0000 Subject: [PATCH] fix(gateway): skip schtasks listing lines that are not task rows on Windows An orphaned TaskCache entry (common on OEM/long-lived installs) makes `schtasks /Query /FO CSV /NH` print a localized one-field diagnostic among the task rows. Discovery refused the whole listing with service_state_unknown before looking for Hermes_Gateway, and it runs on every cold ensure, task installed or not, so Desktop could not start the gateway after a reboot. Only a well-formed three-column row can now name our task; every other line is ignored. A malformed line can therefore only fall through to the documented unmanaged start, never produce a false "task exists". A failing schtasks call is still refused as service_manager_unavailable. Refs #143547 Refs #142140 Co-authored-by: qd678 <12959191+qd678@users.noreply.github.com> Co-authored-by: lim000 <12020793+kouyichi@users.noreply.github.com> --- hermes_cli/gateway_runtime_service.py | 8 ++-- .../test_gateway_runtime_service.py | 37 +++++++++++++++++++ 2 files changed, 41 insertions(+), 4 deletions(-) diff --git a/hermes_cli/gateway_runtime_service.py b/hermes_cli/gateway_runtime_service.py index 2066940aff..794b054271 100644 --- a/hermes_cli/gateway_runtime_service.py +++ b/hermes_cli/gateway_runtime_service.py @@ -231,10 +231,10 @@ def _windows(home: Path, deadline: float) -> ExistingService | None: raise RuntimeStartError(refusal) return _decode_schtasks_output(result.stdout or b"") - rows = list(csv.reader(io.StringIO(query("service_manager_unavailable", "schtasks.exe", "/Query", "/FO", "CSV", "/NH")))) - if any(len(row) != 3 for row in rows if row): - raise RuntimeStartError("service_state_unknown") - if any(row and row[0].lstrip("\\") == name for row in rows): + rows = csv.reader(io.StringIO(query("service_manager_unavailable", "schtasks.exe", "/Query", "/FO", "CSV", "/NH"))) + # Orphaned TaskCache entries print localized one-field diagnostics among the rows (#142140). Only a + # well-formed row can name our task, so skipping the rest can only fall through to the unmanaged path. + if any(len(row) == 3 and row[0].lstrip("\\") == name for row in rows): from hermes_cli.gateway_runtime_service_identity import verify_windows_task unverified = "service_identity_unverified" definition = query(unverified, "schtasks.exe", "/Query", "/TN", name, "/XML") diff --git a/tests/hermes_cli/test_gateway_runtime_service.py b/tests/hermes_cli/test_gateway_runtime_service.py index b814accb7f..25f4ac0d15 100644 --- a/tests/hermes_cli/test_gateway_runtime_service.py +++ b/tests/hermes_cli/test_gateway_runtime_service.py @@ -295,6 +295,43 @@ def test_task_xml_binds_actual_action_and_principal(case, reason, tmp_path): verify_windows_task(xml, home, user, 'S-1-5-21-123') +@pytest.mark.parametrize("installed", [True, False]) +def test_task_discovery_skips_listing_lines_that_are_not_task_rows(tmp_path, monkeypatch, installed): + """An orphaned TaskCache entry (OEM leftovers) makes ``schtasks /Query /FO CSV`` print a localized + one-field diagnostic among the rows (#142140). That refused every cold start with + service_state_unknown, task or not. Such lines are skipped; one naming our task still never counts.""" + from types import SimpleNamespace + from hermes_cli import gateway_runtime_service as service + 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\\owner", "S-1-5-21-123" + listing = ('"\\Microsoft\\Windows\\Defrag\\ScheduledDefrag","N/A","Ready"\r\n' + '错误: 任务映像已损坏或已篡改。\r\n' + f'"\\{name}","N/A"\r\n') + if installed: + listing += f'"\\{name}","N/A","Ready"\r\n' + # whoami first: it is queried with /FO CSV too. + out = {"whoami.exe": f'"{account}","{sid}"\r\n', "/XML": _build_scheduled_task_xml(name, script, account), + "/FO": listing} + queries = [] + + def run(argv, *_args, **_kwargs): + queries.append(argv) + return SimpleNamespace(returncode=0, stdout=next(text for key, text in out.items() if key in argv).encode()) + + monkeypatch.setattr(service, "_run", run) + monkeypatch.setattr(service.sys, "platform", "win32") + found = service.discover_existing_gateway_service(home, deadline=time.monotonic() + 5) + if installed: + assert found is not None and found.start_argv == ("schtasks.exe", "/Run", "/TN", name) + else: + assert found is None and len(queries) == 1 + + @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):