hermes: fail closed on kanban created-event producer drift
The sticky-block gate added in the previous commit classifies a task from the `created` event payload that upstream `create_task` writes. That producer is code we do not own, so trusting it silently was the gap: if upstream renamed the key, dropped it, or stopped deriving it from `initial_status`, the image would still build and ship a consumer that mis-classifies every task it reads. Anchor the producer contract at build time, before the regression suite runs, with three assert-only preconditions: the `initial_status="blocked"` park resolves `task_status` to `"blocked"`, every non-park creation resolves it to something else, and the `created` event carries that same variable under `"status"`. None of them rewrite the producer. Textual anchors cannot see dataflow, so add the runtime net the reviewer asked for. The suite now drives the real API: create + claim an ordinary task, trip the circuit breaker once at failure_limit=1 so it parks with a `gave_up` event (leaving its own `created` event as the most recent create/block/unblock row), then recompute at failure_limit=2 and require promotion to ready. That case is red under an unconditional-true created predicate and red under producer drift that labels every created event blocked, while the explicit block/unblock, dependency-promotion and circuit-breaker-at-current-limit cases stay green. Non-blocked and malformed created payloads are pinned as controls, and the gate now rejects non-dict payloads rather than trusting `.get`. Also make the live placement correction durable: titan-04 is cordoned after repeated kernel undervoltage and kubelet failure and titan-19 was probe/Longhorn unstable under worker load, so both join the hard NotIn list; titan-05 is healthy but sits at 3592m/3600m requested CPU, so the main hermes container gives back 50m (350m -> 300m) to schedule there. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
parent
4f8dcfbbf7
commit
750dfa241f
@ -639,6 +639,54 @@ from pathlib import Path
|
|||||||
|
|
||||||
db_path = Path("/opt/hermes/hermes_cli/kanban_db.py")
|
db_path = Path("/opt/hermes/hermes_cli/kanban_db.py")
|
||||||
db_source = db_path.read_text()
|
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
|
sticky_doc_before = ''' The cheapest signal that distinguishes the two is the most recent
|
||||||
``"blocked"`` / ``"unblocked"`` event for the task. If the most
|
``"blocked"`` / ``"unblocked"`` event for the task. If the most
|
||||||
recent one is ``"blocked"`` (or there is a ``"blocked"`` event and
|
recent one is ``"blocked"`` (or there is a ``"blocked"`` event and
|
||||||
@ -684,6 +732,8 @@ sticky_after = ''' row = conn.execute(
|
|||||||
payload = json.loads(row["payload"] or "{}")
|
payload = json.loads(row["payload"] or "{}")
|
||||||
except (TypeError, ValueError):
|
except (TypeError, ValueError):
|
||||||
return False
|
return False
|
||||||
|
if not isinstance(payload, dict):
|
||||||
|
return False
|
||||||
return payload.get("status") == "blocked"
|
return payload.get("status") == "blocked"
|
||||||
'''
|
'''
|
||||||
if db_source.count(sticky_before) != 1:
|
if db_source.count(sticky_before) != 1:
|
||||||
|
|||||||
@ -2,6 +2,7 @@
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import json
|
||||||
import os
|
import os
|
||||||
import tempfile
|
import tempfile
|
||||||
import unittest
|
import unittest
|
||||||
@ -37,6 +38,47 @@ class BlockedTaskSchedulingTests(unittest.TestCase):
|
|||||||
self.assertIsNotNone(task)
|
self.assertIsNotNone(task)
|
||||||
return task.status
|
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:
|
def test_initial_block_without_parents_is_sticky(self) -> None:
|
||||||
task_id = self.create_task("operator parked", initial_status="blocked")
|
task_id = self.create_task("operator parked", initial_status="blocked")
|
||||||
|
|
||||||
@ -55,6 +97,18 @@ class BlockedTaskSchedulingTests(unittest.TestCase):
|
|||||||
self.assertEqual(kanban_db.recompute_ready(self.connection), 0)
|
self.assertEqual(kanban_db.recompute_ready(self.connection), 0)
|
||||||
self.assertEqual(self.status(task_id), "blocked")
|
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:
|
def test_initial_block_is_released_by_an_explicit_unblock(self) -> None:
|
||||||
task_id = self.create_task("operator parked", initial_status="blocked")
|
task_id = self.create_task("operator parked", initial_status="blocked")
|
||||||
|
|
||||||
@ -105,6 +159,91 @@ class BlockedTaskSchedulingTests(unittest.TestCase):
|
|||||||
)
|
)
|
||||||
self.assertEqual(self.status(task_id), "blocked")
|
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__":
|
if __name__ == "__main__":
|
||||||
unittest.main(verbosity=2)
|
unittest.main(verbosity=2)
|
||||||
|
|||||||
@ -111,7 +111,7 @@ spec:
|
|||||||
values: ["true"]
|
values: ["true"]
|
||||||
- key: kubernetes.io/hostname
|
- key: kubernetes.io/hostname
|
||||||
operator: NotIn
|
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:
|
preferredDuringSchedulingIgnoredDuringExecution:
|
||||||
- weight: 100
|
- weight: 100
|
||||||
preference:
|
preference:
|
||||||
@ -690,7 +690,7 @@ spec:
|
|||||||
seccompProfile:
|
seccompProfile:
|
||||||
type: RuntimeDefault
|
type: RuntimeDefault
|
||||||
resources:
|
resources:
|
||||||
requests: {cpu: 350m, memory: 768Mi}
|
requests: {cpu: 300m, memory: 768Mi}
|
||||||
limits: {cpu: "3", memory: 6Gi}
|
limits: {cpu: "3", memory: 6Gi}
|
||||||
- name: oauth2-proxy
|
- name: oauth2-proxy
|
||||||
image: quay.io/oauth2-proxy/oauth2-proxy:v7.15.3@sha256:10a1165743a192e1940b4708fb9647027185ce11a681a1c5519b442ff7f1f561
|
image: quay.io/oauth2-proxy/oauth2-proxy:v7.15.3@sha256:10a1165743a192e1940b4708fb9647027185ce11a681a1c5519b442ff7f1f561
|
||||||
|
|||||||
@ -1100,6 +1100,33 @@ def test_cli_lane_reserves_cpu_headroom_for_ui_and_auth():
|
|||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
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():
|
def test_agent_root_is_stock_dashboard_and_terminal_is_a_separate_path():
|
||||||
deployment = _agent_deployment()
|
deployment = _agent_deployment()
|
||||||
pod = deployment["spec"]["template"]["spec"]
|
pod = deployment["spec"]["template"]["spec"]
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user