hermes: reconcile pool tests and staging hygiene with the train API

Post-merge fixes after rebasing the distributed worker pool onto the
review-goal-semantics train tip:

- Pool SCM submission tests target the train's relocated receive-pack
  scanner: FEATURE_REF_RE now lives in receive_pack_scan, bodies are
  built via the shared _receive_command helper (valid pack), and the
  update-rejection assertion matches the train's message.
- Take the train's canonical test_hermes_scm_broker,
  test_hermes_cli_dispatch_runtime and test_hermes_cli_execution_edges,
  which exercise the train's broker/dispatch/execution behavior.
- Runtime staging tests patch os.fchown alongside os.chown so the
  UID-10000 _write_secret path passes under a non-root gate runner
  (production ownership behavior unchanged).
- Split the jenkins build-evidence contracts out of
  test_hermes_runtime_access into test_hermes_runtime_evidence to keep
  both files under the 500-LOC hygiene ceiling after the merge.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

testing/tests/test_hermes_cli_dispatch_runtime.py

@@ -14,8 +14,6 @@ from testing.tests.test_hermes_cli_support import SCRIPTS, lanes
 
 
 def test_board_slug_and_connection_failure_paths(monkeypatch):
-    assert lanes._owns_local_workspace(SimpleNamespace(workspace_path=" /workspace "))
-    assert not lanes._owns_local_workspace(SimpleNamespace(workspace_path=""))
     assert lanes._board_slug(SimpleNamespace(slug="cassandra")) == "cassandra"
     assert lanes._board_slug(SimpleNamespace(id="fallback")) == "fallback"
     failures = []
@@ -103,41 +101,6 @@ def test_claim_ready_handles_zero_limit_empty_boards_and_claim_errors(monkeypatc
     assert lanes.claim_ready(set(), 1) == []
 
 
-def test_dispatch_skips_unhealthy_ineligible_and_unclaimed_entries(monkeypatch):
-    class Connection:
-        def __init__(self, board):
-            self.board = board
-
-        def close(self):
-            return None
-
-    tasks = [
-        SimpleNamespace(id="t_ineligible", status="ready", assignee="cli-auto"),
-        SimpleNamespace(id="t_unclaimed", status="ready", assignee="cli-auto"),
-    ]
-    db = SimpleNamespace(
-        list_boards=lambda **_kwargs: [
-            {"slug": "offline"},
-            {"slug": "cassandra"},
-        ],
-        scoped_current_board=lambda _board: nullcontext(),
-        connect=lambda board: (_ for _ in ()).throw(OSError("offline"))
-        if board == "offline"
-        else Connection(board),
-        recompute_ready=lambda _conn: None,
-        list_tasks=lambda _conn: tasks,
-        claim_task=lambda *_args, **_kwargs: None,
-    )
-    monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=db))
-    assert lanes.claim_ready(
-        set(),
-        2,
-        eligible=lambda _board, task: task.id == "t_unclaimed",
-    ) == []
-    monkeypatch.setattr(lanes, "recover_pending_finalizations", lambda: 0)
-    lanes.recover_orphans()
-
-
 class _Future:
     def __init__(self):
         self.polls = 0

testing/tests/test_hermes_cli_execution_edges.py

@@ -16,13 +16,6 @@ class _Connection:
         return None
 
 
-def test_goal_rejection_history_is_list_bounded():
-    execution = __import__("cli_lane_execution")
-    history = ["first"]
-    assert execution._goal_rejections({"goal_rejections": history}) is history
-    assert execution._goal_rejections({"goal_rejections": "invalid"}) == []
-
-
 def test_missing_claim_is_a_noop(tmp_path: Path, monkeypatch):
     db = SimpleNamespace(
         scoped_current_board=lambda _board: nullcontext(),
@@ -71,18 +64,12 @@ def test_transient_callback_storage_errors_do_not_kill_provider(
 
     def provider(*args, **_kwargs):
         assert args[6]("still working") is True
-        return lanes.ProcessResult(
-            1,
-            "failed",
-            {"status": "incomplete", "blockers": ["synthetic provider failure"]},
-            False,
-        )
+        return lanes.ProcessResult(1, "failed", None, False)
 
     monkeypatch.setattr(lanes, "run_provider", provider)
     lanes.execute_claim("cassandra", "t_callbacks")
 
     assert blocks and blocks[0]["kind"] == "capability"
-    assert blocks[0]["reason"] == "synthetic provider failure"
 
 
 def test_restart_handoff_survives_missing_prior_log(tmp_path: Path, monkeypatch):

testing/tests/test_hermes_execution_pool_submission.py

@@ -23,7 +23,9 @@ sys.path[:0] = [str(SCRIPTS), str(SCM_SCRIPTS)]
 import execution_pool_client as client  # noqa: E402
 import execution_pool_protocol as protocol  # noqa: E402
 import execution_pool_scm as scm  # noqa: E402
+import receive_pack_scan  # noqa: E402
 import scm_broker  # noqa: E402
+from testing.tests.test_hermes_scm_broker_support import _receive_command  # noqa: E402
 from testing.tests.test_hermes_execution_pool_mediator import (  # noqa: E402
     KEY,
     RESULT,
@@ -50,7 +52,7 @@ def test_submission_refs_are_creation_only_and_attempt_scoped():
     # Every candidate stays inside a reviewed namespace the broker accepts, and
     # none of them is a protected or base ref.
     for candidate in first + retry:
-        assert scm_broker.FEATURE_REF_RE.fullmatch(f"refs/heads/{candidate}")
+        assert receive_pack_scan.FEATURE_REF_RE.fullmatch(f"refs/heads/{candidate}")
         assert candidate not in {"main", "master"}
 
 
@@ -65,15 +67,10 @@ def test_submission_refs_drop_oversized_names_and_fail_closed_when_empty():
 def test_the_broker_accepts_the_attempt_ref_as_a_creation_and_still_refuses_updates():
     zero, head = b"0" * 40, b"1" * 40
     ref = b"refs/heads/feature/pool-attempt-2"
-    create = b" ".join((zero, head, ref))
-    scm_broker._validate_receive_pack(
-        b"%04x%s\n0000" % (len(create) + 5, create), "token"
-    )
-    update = b" ".join((b"2" * 40, head, ref))
-    with pytest.raises(scm_broker.PolicyError, match="new namespaced branch"):
-        scm_broker._validate_receive_pack(
-            b"%04x%s\n0000" % (len(update) + 5, update), "token"
-        )
+    scm_broker._validate_receive_pack(_receive_command(zero, head, ref), "token")
+    update = _receive_command(b"2" * 40, head, ref)
+    with pytest.raises(scm_broker.PolicyError, match="feature-branch creation"):
+        scm_broker._validate_receive_pack(update, "token")
 
 
 def submit_harness(tmp_path, monkeypatch, remote_refs, *, attempt=1):

testing/tests/test_hermes_runtime_access.py

@@ -82,82 +82,6 @@ def test_gitea_api_rejects_unavailable_methods(method: str):
         )
 
 
-def test_jenkins_evidence_ignores_nested_execution_result():
-    evidence_reader = _load("jenkins_build_evidence")
-    evidence = evidence_reader.parse_build_xml(
-        """
-        <flow-build>
-          <actions><revision><hash>abc123</hash></revision></actions>
-          <timestamp>1000</timestamp><duration>0</duration>
-          <execution><result>SUCCESS</result></execution>
-        </flow-build>
-        """,
-        job="demo",
-        branch="master",
-        number=15,
-    )
-
-    assert evidence.revision == "abc123"
-    assert evidence.result is None
-    assert evidence.building is True
-
-
-def test_jenkins_evidence_reads_only_top_level_terminal_result():
-    evidence_reader = _load("jenkins_build_evidence")
-    evidence = evidence_reader.parse_build_xml(
-        """
-        <flow-build>
-          <actions><revision><hash>abc123</hash></revision></actions>
-          <timestamp>1000</timestamp><duration>250</duration>
-          <result>FAILURE</result>
-          <execution><result>SUCCESS</result></execution>
-        </flow-build>
-        """,
-        job="demo",
-        branch=None,
-        number=4,
-    )
-
-    assert evidence.result == "FAILURE"
-    assert evidence.building is False
-    assert evidence.duration_ms == 250
-
-
-def test_jenkins_evidence_removes_hidden_links_and_ansi_control_sequences():
-    evidence_reader = _load("jenkins_build_evidence")
-
-    value = (
-        "before\n\x1b[8mha:////private-jenkins-payload\x1b[0m[Pipeline] sh\n"
-        "\x1b[31m5 passed\x1b[0m\n"
-    )
-
-    assert evidence_reader.sanitize_log(value) == "before\n[Pipeline] sh\n5 passed\n"
-
-
-def test_jenkins_evidence_falls_back_to_multibranch_suffix(monkeypatch):
-    evidence_reader = _load("jenkins_build_evidence")
-    calls = []
-
-    def fake_find(job, branch, commit, log_lines):
-        calls.append((job, branch, commit, log_lines))
-        if job == "demo":
-            raise ValueError("not a multibranch job")
-        return "evidence"
-
-    monkeypatch.setattr(evidence_reader, "_find_build_for_job", fake_find)
-
-    assert evidence_reader.find_build("demo", "master", "abc", 20) == "evidence"
-    assert [call[0] for call in calls] == ["demo", "demo-branches"]
-
-
-@pytest.mark.parametrize("branch", ["../master", "/master", "feature//unsafe"])
-def test_jenkins_evidence_rejects_unsafe_branch_paths(branch: str):
-    evidence_reader = _load("jenkins_build_evidence")
-
-    with pytest.raises(ValueError):
-        evidence_reader._validate_branch(branch)
-
-
 def test_agent_runtime_stage_keeps_credentials_in_memory(tmp_path: Path, monkeypatch):
     stage = _load("stage_runtime_access")
     vault = tmp_path / "vault"
@@ -196,6 +120,7 @@ def test_agent_runtime_stage_keeps_credentials_in_memory(tmp_path: Path, monkeyp
     monkeypatch.setattr(stage, "RUNTIME_ROOT", runtime)
     monkeypatch.setattr(stage, "PERSISTENT_HOME", home)
     monkeypatch.setattr(stage.os, "chown", lambda *_args: None)
+    monkeypatch.setattr(stage.os, "fchown", lambda *_args: None)
 
     stage.stage_agent()
 
@@ -306,6 +231,7 @@ def test_invalid_runtime_json_is_removed(tmp_path: Path, monkeypatch):
     (vault / "credential").write_text("not-json\n", encoding="utf-8")
     monkeypatch.setattr(stage, "VAULT_ROOT", vault)
     monkeypatch.setattr(stage.os, "chown", lambda *_args: None)
+    monkeypatch.setattr(stage.os, "fchown", lambda *_args: None)
     destination = runtime / "credential.json"
 
     try:

testing/tests/test_hermes_runtime_evidence.py

@@ -0,0 +1,104 @@
+"""Contracts for Hermes' Jenkins build-evidence reader boundary."""
+
+from __future__ import annotations
+
+import importlib.util
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).parents[2]
+HERMES = ROOT / "services" / "hermes"
+SCRIPTS = HERMES / "scripts"
+SCM_SCRIPTS = HERMES / "scm-common" / "scripts"
+if str(SCRIPTS) not in sys.path:
+    sys.path.insert(0, str(SCRIPTS))
+if str(SCM_SCRIPTS) not in sys.path:
+    sys.path.insert(0, str(SCM_SCRIPTS))
+
+
+def _load(name: str):
+    root = SCM_SCRIPTS if name == "gitea_api" else SCRIPTS
+    spec = importlib.util.spec_from_file_location(name, root / f"{name}.py")
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_jenkins_evidence_ignores_nested_execution_result():
+    evidence_reader = _load("jenkins_build_evidence")
+    evidence = evidence_reader.parse_build_xml(
+        """
+        <flow-build>
+          <actions><revision><hash>abc123</hash></revision></actions>
+          <timestamp>1000</timestamp><duration>0</duration>
+          <execution><result>SUCCESS</result></execution>
+        </flow-build>
+        """,
+        job="demo",
+        branch="master",
+        number=15,
+    )
+
+    assert evidence.revision == "abc123"
+    assert evidence.result is None
+    assert evidence.building is True
+
+
+def test_jenkins_evidence_reads_only_top_level_terminal_result():
+    evidence_reader = _load("jenkins_build_evidence")
+    evidence = evidence_reader.parse_build_xml(
+        """
+        <flow-build>
+          <actions><revision><hash>abc123</hash></revision></actions>
+          <timestamp>1000</timestamp><duration>250</duration>
+          <result>FAILURE</result>
+          <execution><result>SUCCESS</result></execution>
+        </flow-build>
+        """,
+        job="demo",
+        branch=None,
+        number=4,
+    )
+
+    assert evidence.result == "FAILURE"
+    assert evidence.building is False
+    assert evidence.duration_ms == 250
+
+
+def test_jenkins_evidence_removes_hidden_links_and_ansi_control_sequences():
+    evidence_reader = _load("jenkins_build_evidence")
+
+    value = (
+        "before\n\x1b[8mha:////private-jenkins-payload\x1b[0m[Pipeline] sh\n"
+        "\x1b[31m5 passed\x1b[0m\n"
+    )
+
+    assert evidence_reader.sanitize_log(value) == "before\n[Pipeline] sh\n5 passed\n"
+
+
+def test_jenkins_evidence_falls_back_to_multibranch_suffix(monkeypatch):
+    evidence_reader = _load("jenkins_build_evidence")
+    calls = []
+
+    def fake_find(job, branch, commit, log_lines):
+        calls.append((job, branch, commit, log_lines))
+        if job == "demo":
+            raise ValueError("not a multibranch job")
+        return "evidence"
+
+    monkeypatch.setattr(evidence_reader, "_find_build_for_job", fake_find)
+
+    assert evidence_reader.find_build("demo", "master", "abc", 20) == "evidence"
+    assert [call[0] for call in calls] == ["demo", "demo-branches"]
+
+
+@pytest.mark.parametrize("branch", ["../master", "/master", "feature//unsafe"])
+def test_jenkins_evidence_rejects_unsafe_branch_paths(branch: str):
+    evidence_reader = _load("jenkins_build_evidence")
+
+    with pytest.raises(ValueError):
+        evidence_reader._validate_branch(branch)

testing/tests/test_hermes_runtime_stage_coverage.py

@@ -22,6 +22,7 @@ def _stage(tmp_path: Path, monkeypatch):
     monkeypatch.setattr(module, "RUNTIME_ROOT", runtime)
     monkeypatch.setattr(module, "PERSISTENT_HOME", home)
     monkeypatch.setattr(module.os, "chown", lambda *_args: None)
+    monkeypatch.setattr(module.os, "fchown", lambda *_args: None)
     return module, vault, runtime, home
 
 

testing/tests/test_hermes_scm_broker.py

@@ -48,7 +48,7 @@ def test_broker_metadata_boundary_reuses_explicit_read_allowlist():
             api.authorize_request("GET", path, None)
 
 
-def test_receive_pack_allows_only_new_branches_in_reviewed_namespaces():
+def test_receive_pack_allows_only_new_namespaced_feature_branch():
     broker = _load("scm_broker")
     zero = b"0" * 40
     commit = b"1" * 40
@@ -57,18 +57,9 @@ def test_receive_pack_allows_only_new_branches_in_reviewed_namespaces():
         _receive_command(zero, commit, b"refs/heads/hermes/focused-fix"),
         "runtime-sentinel",
     )
-    broker._validate_receive_pack(
-        _receive_command(zero, commit, b"refs/heads/wt/t_deadbeef"),
-        "runtime-sentinel",
-    )
-    broker._validate_receive_pack(
-        _receive_command(zero, commit, b"refs/heads/review/t_deadbeef"),
-        "runtime-sentinel",
-    )
     for old, new, ref in (
         (commit, b"2" * 40, b"refs/heads/hermes/focused-fix"),
         (commit, zero, b"refs/heads/hermes/focused-fix"),
-        (commit, commit, b"refs/heads/hermes/focused-fix"),
         (zero, commit, b"refs/heads/main"),
         (zero, commit, b"refs/heads/master"),
         (zero, commit, b"refs/tags/release"),