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

@@ -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")

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):