hermes: harden worker isolation and blocked-task semantics #13

Merged
bstein merged 2 commits from wt/t_cca008de into main 2026-08-17 08:33:42 +00:00
6 changed files with 420 additions and 9 deletions

View File

@ -629,16 +629,125 @@ source = source[:runs_start] + runs_source
path.write_text(source)
PY
# A manual evidence-based close must not expose a parked task to the dispatcher
# between separate `unblock` and `complete` commands. Upstream already permits
# direct completion from `blocked`; extend the same atomic path to `scheduled`
# and make that contract visible in CLI help so agents do not create a ready
# window that can launch redundant work.
# ``create_task(initial_status="blocked")`` is an explicit operator park, but
# upstream only treats later block_task() events as sticky. Recognize the
# existing created(status=blocked) event too. This also protects tasks created
# before this image patch without making dependency or circuit-breaker blocks
# depend on a new persisted field.
RUN python - <<'PY'
from pathlib import Path
db_path = Path("/opt/hermes/hermes_cli/kanban_db.py")
db_source = db_path.read_text()
# The rewritten sticky-block gate below classifies a task from the ``created``
# event payload that ``create_task`` writes. That producer is upstream code we
# do not own, so anchor the exact contract the consumer depends on instead of
# trusting it: an ``initial_status="blocked"`` park must resolve ``task_status``
# to ``"blocked"``, every other creation must resolve it to something else, and
# the ``created`` event must carry that same variable under the ``"status"``
# key. If upstream renames the key, drops it, hardcodes it, or stops deriving it
# from ``initial_status``, fail the build here rather than ship a consumer that
# silently mis-classifies every task it reads. These assert only -- they
# deliberately do not rewrite the producer. The regression suite copied in below
# is the second net: it proves the same contract against the real create_task,
# for the drift these textual anchors cannot see.
producer_park_anchor = ''' if initial_status == "blocked":
task_status = "blocked"
'''
if db_source.count(producer_park_anchor) != 1:
raise SystemExit(
"Hermes Kanban create_task park semantics changed: expected 1, "
f"found {db_source.count(producer_park_anchor)}"
)
# The mirror of the park branch: every non-park creation must resolve to a
# status the gate does *not* read as a park. Without this, drift that widened
# the ladder to park ordinary work would leave both other anchors intact.
producer_unparked_anchor = ''' elif triage:
task_status = "triage"
else:
task_status = "ready"
'''
if db_source.count(producer_unparked_anchor) != 1:
raise SystemExit(
"Hermes Kanban create_task non-park status ladder changed: expected 1, "
f"found {db_source.count(producer_unparked_anchor)}"
)
producer_event_anchor = ''' _append_event(
conn,
task_id,
"created",
{
"assignee": assignee,
"status": task_status,
'''
if db_source.count(producer_event_anchor) != 1:
raise SystemExit(
"Hermes Kanban created-event status payload changed: expected 1, "
f"found {db_source.count(producer_event_anchor)}"
)
sticky_doc_before = ''' The cheapest signal that distinguishes the two is the most recent
``"blocked"`` / ``"unblocked"`` event for the task. If the most
recent one is ``"blocked"`` (or there is a ``"blocked"`` event and
no ``"unblocked"`` event has fired since), the task is sticky and
``recompute_ready`` must *not* auto-promote it.
Returns ``False`` when there is no such event at all (e.g. the task
'''
sticky_doc_after = ''' The cheapest signal that distinguishes the two is the most recent
``"created"`` / ``"blocked"`` / ``"unblocked"`` event for the task.
A ``"created"`` event whose payload has ``status="blocked"`` is the
explicit park requested by ``create_task(initial_status="blocked")``.
A later ``"unblocked"`` event clears either kind of sticky block.
Returns ``False`` when there is no such event at all (e.g. the task
'''
if db_source.count(sticky_doc_before) != 1:
raise SystemExit(
"Hermes Kanban sticky-block documentation changed: expected 1, "
f"found {db_source.count(sticky_doc_before)}"
)
db_source = db_source.replace(sticky_doc_before, sticky_doc_after, 1)
sticky_before = ''' row = conn.execute(
"SELECT kind FROM task_events "
"WHERE task_id = ? AND kind IN ('blocked', 'unblocked') "
"ORDER BY id DESC LIMIT 1",
(task_id,),
).fetchone()
return bool(row) and row["kind"] == "blocked"
'''
sticky_after = ''' row = conn.execute(
"SELECT kind, payload FROM task_events "
"WHERE task_id = ? AND kind IN ('created', 'blocked', 'unblocked') "
"ORDER BY id DESC LIMIT 1",
(task_id,),
).fetchone()
if not row or row["kind"] == "unblocked":
return False
if row["kind"] == "blocked":
return True
try:
payload = json.loads(row["payload"] or "{}")
except (TypeError, ValueError):
return False
if not isinstance(payload, dict):
return False
return payload.get("status") == "blocked"
'''
if db_source.count(sticky_before) != 1:
raise SystemExit(
"Hermes Kanban sticky-block gate changed: expected 1, "
f"found {db_source.count(sticky_before)}"
)
db_source = db_source.replace(sticky_before, sticky_after, 1)
# A manual evidence-based close must not expose a parked task to the dispatcher
# between separate `unblock` and `complete` commands. Upstream already permits
# direct completion from `blocked`; extend the same atomic path to `scheduled`
# and make that contract visible in CLI help so agents do not create a ready
# window that can launch redundant work.
state_before = ''' WHERE id = ?
AND status IN ('running', 'ready', 'blocked')
'''
@ -671,6 +780,10 @@ if cli_source.count(help_before) != 1:
cli_path.write_text(cli_source.replace(help_before, help_after, 1))
PY
COPY dockerfiles/hermes-kanban-blocked-regression.py /tmp/hermes-kanban-blocked-regression.py
RUN /opt/hermes/.venv/bin/python /tmp/hermes-kanban-blocked-regression.py \
&& rm /tmp/hermes-kanban-blocked-regression.py
COPY dockerfiles/hermes-python-sandbox-tool.py /opt/hermes/tools/python_sandbox_tool.py
COPY dockerfiles/hermes-public-extract/__init__.py /opt/hermes/plugins/web/public_extract/__init__.py
COPY dockerfiles/hermes-public-extract/plugin.yaml /opt/hermes/plugins/web/public_extract/plugin.yaml

View File

@ -1,4 +1,5 @@
**
!dockerfiles/hermes-kanban-blocked-regression.py
!dockerfiles/hermes-python-sandbox-tool.py
!dockerfiles/hermes-public-extract/
!dockerfiles/hermes-public-extract/**

View File

@ -0,0 +1,249 @@
"""Regression tests for Hermes Kanban blocked-task scheduling semantics."""
from __future__ import annotations
import json
import os
import tempfile
import unittest
from pathlib import Path
_HERMES_HOME = tempfile.TemporaryDirectory(prefix="hermes-kanban-test-home-")
os.environ["HERMES_HOME"] = _HERMES_HOME.name
from hermes_cli import kanban_db # noqa: E402
class BlockedTaskSchedulingTests(unittest.TestCase):
"""Exercise the real upstream database API against isolated databases."""
def setUp(self) -> None:
self._database_dir = tempfile.TemporaryDirectory(
prefix="hermes-kanban-test-db-"
)
self.connection = kanban_db.connect(
Path(self._database_dir.name) / "kanban.db"
)
def tearDown(self) -> None:
self.connection.close()
self._database_dir.cleanup()
def create_task(self, title: str, **kwargs: object) -> str:
return kanban_db.create_task(self.connection, title=title, **kwargs)
def status(self, task_id: str) -> str:
task = kanban_db.get_task(self.connection, task_id)
self.assertIsNotNone(task)
return task.status
def created_payload(self, task_id: str) -> dict:
"""Return the decoded payload of the task's ``created`` event."""
row = self.connection.execute(
"SELECT payload FROM task_events "
"WHERE task_id = ? AND kind = 'created' ORDER BY id LIMIT 1",
(task_id,),
).fetchone()
self.assertIsNotNone(row, "create_task emitted no 'created' event")
return json.loads(row["payload"] or "{}")
def set_created_payload(self, task_id: str, payload: object) -> None:
"""Overwrite a ``created`` event payload to simulate producer drift."""
with self.connection:
self.connection.execute(
"UPDATE task_events SET payload = ? "
"WHERE task_id = ? AND kind = 'created'",
(payload, task_id),
)
def block_via_circuit_breaker(self, title: str) -> str:
"""Create + claim a normal task, then trip the breaker once on it.
The resulting task is ``blocked`` with exactly one recorded failure and
-- critically -- no ``blocked`` event, because the breaker emits
``gave_up``. Its most recent created/blocked/unblocked event is
therefore its own ``created`` event, which is what makes it the right
probe for the created-event branch of the sticky gate.
"""
task_id = self.create_task(title)
self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
self.assertTrue(
kanban_db._record_spawn_failure(
self.connection,
task_id,
"worker failed",
failure_limit=1,
)
)
self.assertEqual(self.status(task_id), "blocked")
return task_id
def test_initial_block_without_parents_is_sticky(self) -> None:
task_id = self.create_task("operator parked", initial_status="blocked")
self.assertEqual(kanban_db.recompute_ready(self.connection), 0)
self.assertEqual(self.status(task_id), "blocked")
def test_initial_block_with_complete_parents_is_sticky(self) -> None:
parent_id = self.create_task("completed prerequisite")
self.assertTrue(kanban_db.complete_task(self.connection, parent_id))
task_id = self.create_task(
"operator parked after prerequisite",
initial_status="blocked",
parents=[parent_id],
)
self.assertEqual(kanban_db.recompute_ready(self.connection), 0)
self.assertEqual(self.status(task_id), "blocked")
def test_initial_block_remains_sticky_when_parent_later_completes(self) -> None:
parent_id = self.create_task("incomplete prerequisite")
task_id = self.create_task(
"operator parked before prerequisite completes",
initial_status="blocked",
parents=[parent_id],
)
self.assertEqual(self.status(task_id), "blocked")
self.assertTrue(kanban_db.complete_task(self.connection, parent_id))
self.assertEqual(self.status(task_id), "blocked")
def test_initial_block_is_released_by_an_explicit_unblock(self) -> None:
task_id = self.create_task("operator parked", initial_status="blocked")
self.assertTrue(kanban_db.unblock_task(self.connection, task_id))
self.assertEqual(self.status(task_id), "ready")
def test_dependency_block_promotes_after_parent_completes(self) -> None:
parent_id = self.create_task("incomplete prerequisite")
task_id = self.create_task("dependency gated", parents=[parent_id])
self.assertEqual(self.status(task_id), "todo")
self.assertTrue(kanban_db.complete_task(self.connection, parent_id))
self.assertEqual(self.status(task_id), "ready")
def test_explicit_block_and_unblock_remain_sticky_and_reversible(self) -> None:
task_id = self.create_task("worker handoff")
self.assertTrue(
kanban_db.block_task(
self.connection,
task_id,
reason="human review required",
)
)
self.assertEqual(kanban_db.recompute_ready(self.connection), 0)
self.assertEqual(self.status(task_id), "blocked")
self.assertTrue(kanban_db.unblock_task(self.connection, task_id))
self.assertEqual(self.status(task_id), "ready")
def test_circuit_breaker_block_is_not_auto_promoted(self) -> None:
task_id = self.create_task("repeated worker failure")
self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))
# Upstream has no public circuit-breaker entry point; exercise the
# dispatcher helper that owns its persisted failure-limit semantics.
self.assertTrue(
kanban_db._record_spawn_failure(
self.connection,
task_id,
"worker failed",
failure_limit=1,
)
)
self.assertEqual(self.status(task_id), "blocked")
self.assertEqual(
kanban_db.recompute_ready(self.connection, failure_limit=1),
0,
)
self.assertEqual(self.status(task_id), "blocked")
def test_circuit_breaker_block_is_promoted_below_the_failure_limit(self) -> None:
"""Non-sticky oracle for the created-event branch of the sticky gate.
A circuit-breaker block must still auto-recover once the effective
failure limit rises above the recorded failure count -- that is the
pre-existing upstream contract the sticky-block patch must not widen.
This task's most recent created/blocked/unblocked event is its own
``created`` event carrying ``status="ready"``, so it is exactly the
case that a mis-classifying gate strands. It fails if the gate treats
any ``created`` event as a park (over-broad consumer) and it fails if
``create_task`` drifts to labelling every ``created`` event
``"blocked"`` (over-broad producer), while the
circuit-breaker-at-current-limit case above still holds.
"""
task_id = self.block_via_circuit_breaker("recoverable worker failure")
self.assertFalse(
kanban_db._has_sticky_block(self.connection, task_id),
"a circuit-breaker block must never be classified as a park",
)
self.assertEqual(
kanban_db.recompute_ready(self.connection, failure_limit=2),
1,
)
self.assertEqual(self.status(task_id), "ready")
def test_created_event_carries_the_status_the_sticky_gate_reads(self) -> None:
"""Producer contract asserted directly against the real create_task.
The sticky gate reads ``created`` payload ``status``. Pin both poles of
that contract so producer drift is a test failure here as well as a
build failure at the Dockerfile source anchor.
"""
parked_id = self.create_task("operator parked", initial_status="blocked")
normal_id = self.create_task("ordinary work")
self.assertEqual(self.created_payload(parked_id).get("status"), "blocked")
self.assertEqual(self.created_payload(normal_id).get("status"), "ready")
def test_non_blocked_created_payloads_are_not_treated_as_a_park(self) -> None:
"""Only ``status="blocked"`` parks; every other value must promote."""
for payload_status in ("ready", "todo", "running", "triage"):
with self.subTest(payload_status=payload_status):
task_id = self.block_via_circuit_breaker(
f"breaker trip with {payload_status} created payload"
)
self.set_created_payload(
task_id, json.dumps({"status": payload_status})
)
self.assertFalse(
kanban_db._has_sticky_block(self.connection, task_id)
)
self.assertEqual(
kanban_db.recompute_ready(self.connection, failure_limit=2),
1,
)
self.assertEqual(self.status(task_id), "ready")
def test_malformed_created_payloads_fail_closed_to_not_parked(self) -> None:
"""Undecodable or non-object payloads must not park -- or crash.
``recompute_ready`` walks every blocked task in one pass, so an
exception escaping the gate would stall promotion board-wide. Falling
back to "not parked" preserves upstream's pre-patch auto-recover
default for rows the gate cannot interpret.
"""
malformed = (None, "", " ", "{not json", "[]", "null", '"blocked"', "17")
for payload in malformed:
with self.subTest(payload=payload):
task_id = self.block_via_circuit_breaker(
f"breaker trip with payload {payload!r}"
)
self.set_created_payload(task_id, payload)
self.assertFalse(
kanban_db._has_sticky_block(self.connection, task_id)
)
self.assertEqual(
kanban_db.recompute_ready(self.connection, failure_limit=2),
1,
)
self.assertEqual(self.status(task_id), "ready")
if __name__ == "__main__":
unittest.main(verbosity=2)

View File

@ -91,6 +91,7 @@ spec:
vault.hashicorp.com/agent-limits-mem: 128Mi
spec:
serviceAccountName: hermes-agent
enableServiceLinks: false
automountServiceAccountToken: true
securityContext:
fsGroup: 10000
@ -110,7 +111,7 @@ spec:
values: ["true"]
- key: kubernetes.io/hostname
operator: NotIn
values: [titan-08, titan-13, titan-14, titan-17, titan-18]
values: [titan-04, titan-08, titan-13, titan-14, titan-17, titan-18, titan-19]
preferredDuringSchedulingIgnoredDuringExecution:
- weight: 100
preference:
@ -689,7 +690,7 @@ spec:
seccompProfile:
type: RuntimeDefault
resources:
requests: {cpu: 350m, memory: 768Mi}
requests: {cpu: 300m, memory: 768Mi}
limits: {cpu: "3", memory: 6Gi}
- name: oauth2-proxy
image: quay.io/oauth2-proxy/oauth2-proxy:v7.15.3@sha256:10a1165743a192e1940b4708fb9647027185ce11a681a1c5519b442ff7f1f561
@ -873,7 +874,7 @@ spec:
- {name: CLAUDE_CONFIG_DIR, value: /runtime-access/claude}
- {name: KUBECONFIG, value: /opt/data/home/.kube/config}
- {name: PYTHONPATH, value: /opt/hermes}
- {name: HERMES_CLI_LANE_CONCURRENCY, value: "4"}
- {name: HERMES_CLI_LANE_CONCURRENCY, value: "2"}
- {name: HERMES_AUTO_ROUTER_PROFILE, value: agent}
- {name: PATH, value: /opt/coordinator:/opt/data/tools/bin:/opt/data/home/.local/bin:/opt/hermes/.venv/bin:/usr/local/bin:/usr/bin:/bin}
securityContext:
@ -896,7 +897,7 @@ spec:
- {name: tmp, mountPath: /tmp}
resources:
requests: {cpu: 100m, memory: 256Mi}
limits: {cpu: "3", memory: 6Gi}
limits: {cpu: "2", memory: 6Gi}
- name: model-steward
image: registry.bstein.dev/bstein/hermes-agent@sha256:81970563e542f0720773e72297810b3a844b83e381e278f25c0916c78d930107
imagePullPolicy: IfNotPresent

View File

@ -120,6 +120,9 @@ def test_agent_image_completes_parked_kanban_tasks_atomically():
"Atomically mark running, ready, blocked, or scheduled tasks done"
in dockerfile
)
assert "kind IN ('created', 'blocked', 'unblocked')" in dockerfile
assert "payload.get(\"status\") == \"blocked\"" in dockerfile
assert "hermes-kanban-blocked-regression.py" in dockerfile
def test_sandbox_shares_only_the_tenant_workspace_without_credentials():

View File

@ -1074,6 +1074,7 @@ def test_agent_uses_one_native_kanban_control_plane():
deployment = _agent_deployment()
pod = deployment["spec"]["template"]["spec"]
assert pod["enableServiceLinks"] is False
names = {item["name"] for item in pod["containers"]}
assert "cli-lane-runner" in names
assert "terminal" in names
@ -1083,6 +1084,49 @@ def test_agent_uses_one_native_kanban_control_plane():
assert "herdr-dispatch" not in rendered
def test_cli_lane_reserves_cpu_headroom_for_ui_and_auth():
deployment = _agent_deployment()
containers = {
item["name"]: item
for item in deployment["spec"]["template"]["spec"]["containers"]
}
lane = containers["cli-lane-runner"]
environment = {item["name"]: item["value"] for item in lane["env"]}
assert environment["HERMES_CLI_LANE_CONCURRENCY"] == "2"
assert lane["resources"] == {
"requests": {"cpu": "100m", "memory": "256Mi"},
"limits": {"cpu": "2", "memory": "6Gi"},
}
def test_agent_avoids_unhealthy_nodes_and_fits_its_remaining_capacity():
"""Placement correction: keep the agent off nodes that cannot hold it.
titan-04 is cordoned after repeated kernel undervoltage and kubelet
failure, and titan-19 was probe/Longhorn unstable under worker load, so
both must join the existing hard exclusions. That leaves titan-05 as the
healthy candidate, which is tight enough on requested CPU that the main
container has to give back 50m to schedule there.
"""
pod = _agent_deployment()["spec"]["template"]["spec"]
hostnames = next(
item
for item in pod["affinity"]["nodeAffinity"][
"requiredDuringSchedulingIgnoredDuringExecution"
]["nodeSelectorTerms"][0]["matchExpressions"]
if item["key"] == "kubernetes.io/hostname"
)
assert hostnames["operator"] == "NotIn"
assert set(hostnames["values"]) >= {"titan-04", "titan-19"}
hermes = next(
item for item in pod["containers"] if item["name"] == "hermes"
)
assert hermes["resources"]["requests"]["cpu"] == "300m"
def test_agent_root_is_stock_dashboard_and_terminal_is_a_separate_path():
deployment = _agent_deployment()
pod = deployment["spec"]["template"]["spec"]