diff --git a/clusters/atlas/flux-system/applications/hermes/kustomization.yaml b/clusters/atlas/flux-system/applications/hermes/kustomization.yaml index 084e6074..09246661 100644 --- a/clusters/atlas/flux-system/applications/hermes/kustomization.yaml +++ b/clusters/atlas/flux-system/applications/hermes/kustomization.yaml @@ -30,10 +30,12 @@ spec: kind: Deployment name: hermes-agent namespace: hermes - - apiVersion: apps/v1 - kind: StatefulSet - name: hermes-execution-worker - namespace: hermes + # hermes-execution-worker and its mediators are deliberately absent. They run + # at scavenger priority on a best-effort pool whose first start installs the + # provider CLIs, so gating this Kustomization's 10m health window on them + # would stall hermes-chat and hermes-observer-bindings, which dependsOn + # hermes. The pool reports its own health through the worker readiness probe, + # the per-ordinal mediator /ready endpoint, and the coordinator on :9007. - apiVersion: apps/v1 kind: DaemonSet name: hermes-node-ssh-access diff --git a/services/hermes/execution-mediator.yaml b/services/hermes/execution-mediator.yaml index d41ee9f9..4970aa4f 100644 --- a/services/hermes/execution-mediator.yaml +++ b/services/hermes/execution-mediator.yaml @@ -83,17 +83,34 @@ spec: serviceAccountName: hermes-execution-worker automountServiceAccountToken: false enableServiceLinks: false + # Priority parity with the worker: a pressured node evicts the pair + # symmetrically instead of moving only the lower-priority half, and the + # pool never preempts anything else in the cluster. + priorityClassName: scavenger securityContext: fsGroup: 10000 fsGroupChangePolicy: OnRootMismatch seccompProfile: {type: RuntimeDefault} affinity: - podAffinity: + # Colocation with the worker is a throughput preference, not a + # requirement. The shared workspace claim is ReadWriteMany, so this + # mediator still schedules and serves when its worker is Pending or + # placed elsewhere -- which is what makes a drain self-healing. + nodeAffinity: requiredDuringSchedulingIgnoredDuringExecution: - - labelSelector: - matchLabels: - statefulset.kubernetes.io/pod-name: hermes-execution-worker-0 - topologyKey: kubernetes.io/hostname + nodeSelectorTerms: + - matchExpressions: + - {key: kubernetes.io/arch, operator: In, values: [arm64]} + - {key: node-role.kubernetes.io/worker, operator: In, values: ["true"]} + - {key: kubernetes.io/hostname, operator: NotIn, values: [titan-04, titan-13, titan-14, titan-17, titan-18, titan-19, titan-22, titan-24]} + podAffinity: + preferredDuringSchedulingIgnoredDuringExecution: + - weight: 100 + podAffinityTerm: + labelSelector: + matchLabels: + statefulset.kubernetes.io/pod-name: hermes-execution-worker-0 + topologyKey: kubernetes.io/hostname initContainers: - name: stage-mediator-access image: registry.bstein.dev/bstein/hermes-agent @@ -209,17 +226,34 @@ spec: serviceAccountName: hermes-execution-worker automountServiceAccountToken: false enableServiceLinks: false + # Priority parity with the worker: a pressured node evicts the pair + # symmetrically instead of moving only the lower-priority half, and the + # pool never preempts anything else in the cluster. + priorityClassName: scavenger securityContext: fsGroup: 10000 fsGroupChangePolicy: OnRootMismatch seccompProfile: {type: RuntimeDefault} affinity: - podAffinity: + # Colocation with the worker is a throughput preference, not a + # requirement. The shared workspace claim is ReadWriteMany, so this + # mediator still schedules and serves when its worker is Pending or + # placed elsewhere -- which is what makes a drain self-healing. + nodeAffinity: requiredDuringSchedulingIgnoredDuringExecution: - - labelSelector: - matchLabels: - statefulset.kubernetes.io/pod-name: hermes-execution-worker-1 - topologyKey: kubernetes.io/hostname + nodeSelectorTerms: + - matchExpressions: + - {key: kubernetes.io/arch, operator: In, values: [arm64]} + - {key: node-role.kubernetes.io/worker, operator: In, values: ["true"]} + - {key: kubernetes.io/hostname, operator: NotIn, values: [titan-04, titan-13, titan-14, titan-17, titan-18, titan-19, titan-22, titan-24]} + podAffinity: + preferredDuringSchedulingIgnoredDuringExecution: + - weight: 100 + podAffinityTerm: + labelSelector: + matchLabels: + statefulset.kubernetes.io/pod-name: hermes-execution-worker-1 + topologyKey: kubernetes.io/hostname initContainers: - name: stage-mediator-access image: registry.bstein.dev/bstein/hermes-agent @@ -335,17 +369,34 @@ spec: serviceAccountName: hermes-execution-worker automountServiceAccountToken: false enableServiceLinks: false + # Priority parity with the worker: a pressured node evicts the pair + # symmetrically instead of moving only the lower-priority half, and the + # pool never preempts anything else in the cluster. + priorityClassName: scavenger securityContext: fsGroup: 10000 fsGroupChangePolicy: OnRootMismatch seccompProfile: {type: RuntimeDefault} affinity: - podAffinity: + # Colocation with the worker is a throughput preference, not a + # requirement. The shared workspace claim is ReadWriteMany, so this + # mediator still schedules and serves when its worker is Pending or + # placed elsewhere -- which is what makes a drain self-healing. + nodeAffinity: requiredDuringSchedulingIgnoredDuringExecution: - - labelSelector: - matchLabels: - statefulset.kubernetes.io/pod-name: hermes-execution-worker-2 - topologyKey: kubernetes.io/hostname + nodeSelectorTerms: + - matchExpressions: + - {key: kubernetes.io/arch, operator: In, values: [arm64]} + - {key: node-role.kubernetes.io/worker, operator: In, values: ["true"]} + - {key: kubernetes.io/hostname, operator: NotIn, values: [titan-04, titan-13, titan-14, titan-17, titan-18, titan-19, titan-22, titan-24]} + podAffinity: + preferredDuringSchedulingIgnoredDuringExecution: + - weight: 100 + podAffinityTerm: + labelSelector: + matchLabels: + statefulset.kubernetes.io/pod-name: hermes-execution-worker-2 + topologyKey: kubernetes.io/hostname initContainers: - name: stage-mediator-access image: registry.bstein.dev/bstein/hermes-agent diff --git a/services/hermes/execution-worker-statefulset.yaml b/services/hermes/execution-worker-statefulset.yaml index 6794f4da..33f13f23 100644 --- a/services/hermes/execution-worker-statefulset.yaml +++ b/services/hermes/execution-worker-statefulset.yaml @@ -144,13 +144,24 @@ spec: imagePullPolicy: IfNotPresent command: [/bin/sh, -ec] args: + # The tools volume is durable, so this install runs once per pinned + # version rather than on every Pod start. The version marker is only + # trusted when both binaries are actually present and executable, so + # a partial or pruned cache reinstalls instead of failing closed + # forever. The install itself is time-bounded so a stalled registry + # surfaces as a fast init failure with backoff instead of an + # open-ended wait that a Flux health timeout would have to absorb. - | tools=/worker-data/tools + marker="${tools}/.cli-versions-0.147.0-2.1.226" mkdir -p "${tools}/bin" - if [ ! -f "${tools}/.cli-versions-0.147.0-2.1.226" ]; then - npm install --global --omit=dev --no-audit --no-fund --prefix "${tools}" \ + if [ ! -f "${marker}" ] || [ ! -x "${tools}/bin/codex" ] \ + || [ ! -x "${tools}/bin/claude" ]; then + rm -f "${marker}" + timeout 900 npm install --global --omit=dev --no-audit --no-fund \ + --fetch-timeout=120000 --fetch-retries=2 --prefix "${tools}" \ @openai/codex@0.147.0 @anthropic-ai/claude-code@2.1.226 - touch "${tools}/.cli-versions-0.147.0-2.1.226" + touch "${marker}" fi test -x "${tools}/bin/codex" test -x "${tools}/bin/claude" @@ -218,8 +229,6 @@ spec: volumes: - name: worker-data emptyDir: {sizeLimit: 128Mi} - - name: tools - emptyDir: {sizeLimit: 1Gi} - name: coordinator configMap: name: hermes-execution-pool @@ -242,16 +251,34 @@ spec: items: - {path: namespace, fieldRef: {fieldPath: metadata.namespace}} volumeClaimTemplates: + # The workspace is shared with this ordinal's mediator, which performs every + # SCM operation on it. ReadWriteMany removes the cross-node Multi-Attach + # deadlock that a drain or preemption used to create between the two Pods, + # so each can be evicted and rescheduled independently. Longhorn already + # serves the hermes-chat tenant workspaces this way on the same class. - metadata: name: workspace labels: app: hermes-execution-worker + spec: + accessModes: [ReadWriteMany] + storageClassName: astreae + resources: + requests: + storage: 30Gi + # Provider CLIs are installed once per pinned version onto a durable volume + # instead of being re-downloaded into an emptyDir on every Pod start. + - metadata: + name: tools + labels: + app: hermes-execution-worker spec: accessModes: [ReadWriteOnce] storageClassName: astreae resources: requests: - storage: 30Gi + storage: 2Gi + # OAuth refresh ownership is single-mounter by design: keep it ReadWriteOnce. - metadata: name: provider-access labels: diff --git a/services/hermes/kustomization.yaml b/services/hermes/kustomization.yaml index 6becbaf6..e0dfc35c 100644 --- a/services/hermes/kustomization.yaml +++ b/services/hermes/kustomization.yaml @@ -83,8 +83,10 @@ configMapGenerator: - cli_lane_routing.py=scripts/cli_lane_routing.py - cli_lane_runner.py=scripts/cli_lane_runner.py - execution_pool_protocol.py=scripts/execution_pool_protocol.py + - execution_pool_store.py=scripts/execution_pool_store.py - execution_pool_project.py=scripts/execution_pool_project.py - execution_pool_coordinator.py=scripts/execution_pool_coordinator.py + - execution_pool_maintenance.py=scripts/execution_pool_maintenance.py - execution_pool_server.py=scripts/execution_pool_server.py - execution_pool_client.py=scripts/execution_pool_client.py - execution_pool_worker.py=scripts/execution_pool_worker.py diff --git a/services/hermes/scripts/execution_pool_client.py b/services/hermes/scripts/execution_pool_client.py index a1dc6622..2d376c48 100644 --- a/services/hermes/scripts/execution_pool_client.py +++ b/services/hermes/scripts/execution_pool_client.py @@ -137,16 +137,42 @@ class ClientBoundary: raise ProtocolError("coordinator acknowledgement binding changed") return {"ack": response["payload"]} + def _submit( + self, assignment: dict[str, Any], request: dict[str, Any], + structured: dict[str, Any], + ) -> None: + """Publish the run's work, downgrading the result rather than losing it. + + A refused push or draft used to unwind the whole worker, so the run's + actual output was discarded and the task was blocked with only the SCM + error. The commits live on the durable workspace volume, so the result is + still reported here -- as blocked, carrying the reason -- which keeps the + prior work recoverable and reviewable by a human. + """ + try: + submission = self.scm.submit(assignment, request) + except (OSError, ProtocolError, RuntimeError, ValueError) as error: + structured["status"] = "blocked" + reason = ( + "Distributed SCM submission failed; the commits remain on this " + f"ordinal's workspace: {type(error).__name__}: {error}" + ) + if reason not in structured["blockers"]: + structured["blockers"].append(reason) + return + pull = str(submission.get("pull_request") or "") + branch = str(submission.get("branch") or "") + for artifact in (pull, f"branch:{branch}" if branch else ""): + if artifact and artifact not in structured["artifacts"]: + structured["artifacts"].append(artifact) + def finish(self, request: dict[str, Any]) -> dict[str, Any]: payload = _validate_result(request.get("payload")) with self.lock: assignment, binding = self._current_for(request.get("binding")) structured = payload["structured"] if structured["status"] == "completed" and int(payload.get("returncode", 1)) == 0: - submission = self.scm.submit(assignment, request) - pull = str(submission.get("pull_request") or "") - if pull and pull not in structured["artifacts"]: - structured["artifacts"].append(pull) + self._submit(assignment, request, structured) response = self._post( "/v1/result", sign_envelope(self.key, "result", binding, payload) ) diff --git a/services/hermes/scripts/execution_pool_coordinator.py b/services/hermes/scripts/execution_pool_coordinator.py index 6fc0f5a4..1f4b0e85 100644 --- a/services/hermes/scripts/execution_pool_coordinator.py +++ b/services/hermes/scripts/execution_pool_coordinator.py @@ -6,8 +6,6 @@ from __future__ import annotations import json import os import re -import sqlite3 -import sys import threading import time from http.server import BaseHTTPRequestHandler @@ -15,22 +13,17 @@ from pathlib import Path from typing import Any import cli_lane_goal -import cli_lane_dispatch from cli_lane_config import canonical_run_id -from execution_pool_project import ( - ProjectPolicyError, - distributed_workspace_eligible, - resolve_assignment, -) +from execution_pool_project import resolve_assignment from execution_pool_protocol import ( MAX_ACTIVITY_BYTES, - PoolStore, ProtocolError, derive_ordinal_key, envelope_ordinal, sign_envelope, verify_envelope, ) +from execution_pool_store import PoolStore REDACTIONS = ( re.compile(r"(?i)bearer\s+[A-Za-z0-9._~+/-]{12,}"), @@ -274,186 +267,36 @@ class Coordinator: finally: connection.close() + @staticmethod + def assignment_payload( + kanban_db: Any, connection: Any, task: Any, board: str + ) -> dict[str, Any]: + """Expose payload construction to the maintenance passes.""" + return assignment_payload(kanban_db, connection, task, board) + def recover_results(self) -> None: - for record in self.store.pending_results(): - try: - self.finalize(record) - except (OSError, sqlite3.Error) as error: - print(f"result recovery deferred: {error}", file=sys.stderr, flush=True) + """Re-apply every durable result that Kanban has not confirmed yet.""" + from execution_pool_maintenance import recover_results + + recover_results(self) def expire_leases(self) -> None: """Fence dead attempts; after bounded retries, surface and release the run.""" - from hermes_cli import kanban_db + from execution_pool_maintenance import expire_leases - for record in self.store.expire_leases(): - if record.get("state") != "lease_failed": - continue - binding = self._binding(record) - run_id = canonical_run_id(binding["run_id"]) - if run_id is None: - self.store.finalize(binding, "stale") - continue - with self._kanban_lock, kanban_db.scoped_current_board(binding["board"]): - connection = kanban_db.connect(board=binding["board"]) - try: - changed = kanban_db.block_task( - connection, - binding["task_id"], - reason=( - "Distributed worker lease expired after " - f"{binding['attempt']} fenced attempts" - ), - kind="transient", - expected_run_id=run_id, - ) - self.store.finalize( - binding, "finalized" if changed else "stale" - ) - finally: - connection.close() + expire_leases(self) def reconcile(self) -> None: """Recover the narrow claim/assignment crash gaps without double execution.""" - from hermes_cli import kanban_db + from execution_pool_maintenance import reconcile - active = self.store.active_assignments() - active_runs = { - (record["board"], record["task_id"], record["run_id"]) - for record in active - } - for record in active: - try: - with kanban_db.scoped_current_board(record["board"]): - connection = kanban_db.connect(board=record["board"]) - try: - task = kanban_db.get_task(connection, record["task_id"]) - current_run_id = ( - canonical_run_id(_task_value(task, "current_run_id", None)) - if task - else None - ) - current = str(current_run_id or "") - status = str(_task_value(task, "status", "") or "") if task else "" - finally: - connection.close() - if current != record["run_id"] or status != "running": - self.store.finalize(record, "stale") - except (OSError, sqlite3.Error): - continue - ordinals = self.store.available_ordinals() - if not ordinals: - return - for raw_board in kanban_db.list_boards(include_archived=False): - board = cli_lane_dispatch._board_slug(raw_board) - if not board: - continue - with kanban_db.scoped_current_board(board): - connection = kanban_db.connect(board=board) - try: - tasks = kanban_db.list_tasks(connection) - for task in tasks: - task_id = str(_task_value(task, "id", "") or "") - raw_run_id = canonical_run_id( - _task_value(task, "current_run_id", None) - ) - run_id = str(raw_run_id or "") - assignee = str(_task_value(task, "assignee", "") or "") - if ( - not ordinals - or not task_id - or not run_id - or str(_task_value(task, "status", "")) != "running" - or not assignee.startswith("cli-") - or not distributed_workspace_eligible(task) - or (board, task_id, run_id) in active_runs - ): - continue - try: - payload = assignment_payload(kanban_db, connection, task, board) - except Exception as error: - kanban_db.block_task( - connection, task_id, - reason=( - "Distributed assignment recovery failed: " - f"{type(error).__name__}: {error}" - ), - kind="capability", expected_run_id=raw_run_id, - ) - continue - ordinal = ordinals.pop(0) - self.store.add( - { - "board": board, "task_id": task_id, - "run_id": run_id, "worker_ordinal": ordinal, - "attempt": 1, - }, - payload, - ) - finally: - connection.close() + reconcile(self) def dispatch(self) -> None: """Claim up to the free ordinal count and materialize assignments.""" - from hermes_cli import kanban_db + from execution_pool_maintenance import dispatch - ordinals = self.store.available_ordinals() - if not ordinals: - return - claimed = cli_lane_dispatch.claim_ready( - set(), len(ordinals), - lambda _board, task: distributed_workspace_eligible(task), - ) - for ordinal, (board, task_id) in zip(ordinals, claimed, strict=False): - run_id = "" - database_run_id: int | None = None - try: - with kanban_db.scoped_current_board(board): - connection = kanban_db.connect(board=board) - try: - task = kanban_db.get_task(connection, task_id) - if task is None: - continue - if not distributed_workspace_eligible(task): - raw_run = canonical_run_id( - _task_value(task, "current_run_id", None) - ) - if raw_run is not None: - kanban_db.block_task( - connection, task_id, - reason=( - "Distributed claim fenced because an existing " - "workspace is owned by the local lane" - ), - kind="capability", - expected_run_id=raw_run, - ) - continue - raw_run = canonical_run_id( - _task_value(task, "current_run_id", None) - ) - if raw_run is None: - raise ProjectPolicyError("claimed task has no canonical run ID") - database_run_id = raw_run - run_id = str(raw_run) - payload = assignment_payload(kanban_db, connection, task, board) - finally: - connection.close() - binding = { - "board": board, "task_id": task_id, "run_id": run_id, - "worker_ordinal": ordinal, "attempt": 1, - } - self.store.add(binding, payload) - except Exception as error: - reason = f"Distributed assignment preparation failed: {type(error).__name__}: {error}" - with kanban_db.scoped_current_board(board): - connection = kanban_db.connect(board=board) - try: - kanban_db.block_task( - connection, task_id, reason=reason, kind="capability", - expected_run_id=database_run_id, - ) - finally: - connection.close() + dispatch(self) def handler_factory(coordinator: Coordinator) -> type[BaseHTTPRequestHandler]: diff --git a/services/hermes/scripts/execution_pool_maintenance.py b/services/hermes/scripts/execution_pool_maintenance.py new file mode 100644 index 00000000..c3d3d1d5 --- /dev/null +++ b/services/hermes/scripts/execution_pool_maintenance.py @@ -0,0 +1,371 @@ +#!/usr/bin/env python3 +"""Recoverable maintenance passes for the Hermes execution-pool coordinator. + +Every pass here runs against a shared Kanban SQLite database that other writers +(the agent UI, the local CLI lane, crons) hold concurrently, so *any* board call +can fail transiently. The rules this module enforces: + +* No pass ever leaves a claimed run without a durable row that some later pass + will revisit. A Kanban write that fails leaves the row in its retryable state. +* No single poisoned row, task, or board aborts the work queued behind it. +* A durable row only reaches a terminal state on authoritative evidence about + its Kanban run, because terminal rows are the only ones ever collected. +* A coordinator-side fault is never converted into a Kanban mutation. Only a + genuine capability failure of the run itself parks that exact run. +""" + +from __future__ import annotations + +import sqlite3 +import sys +from typing import Any + +import cli_lane_dispatch +from cli_lane_config import canonical_run_id +from execution_pool_project import ProjectPolicyError, distributed_workspace_eligible +from execution_pool_protocol import ProtocolError + + +RECOVERABLE_BOARD_ERRORS = (OSError, sqlite3.Error) +DEFERRALS: dict[str, str] = {} +MAX_DEFERRALS = 64 + + +def _task_value(task: Any, name: str, default: Any = None) -> Any: + return getattr(task, name, default) + + +def _defer(context: str, error: BaseException) -> None: + """Report one recoverable pool-side failure once per distinct detail. + + The maintenance loop re-runs every few seconds, so an unchanged failure is + reported once rather than on every pass. The table is bounded because its + keys include run identities. + """ + detail = f"{type(error).__name__}: {error}" + if DEFERRALS.get(context) == detail: + return + if len(DEFERRALS) >= MAX_DEFERRALS: + DEFERRALS.clear() + DEFERRALS[context] = detail + print(f"pool work deferred for {context}: {detail}", file=sys.stderr, flush=True) + + +def _settled(context: str) -> None: + DEFERRALS.pop(context, None) + + +def _context(record: dict[str, Any]) -> str: + return f"{record['board']}/{record['task_id']}#{record['run_id']}" + + +def _block_exact( + kanban_db: Any, + board: str, + task_id: str, + run_id: int | None, + reason: str, + *, + kind: str = "capability", +) -> None: + """Park exactly one run, never letting the write itself escape.""" + try: + with kanban_db.scoped_current_board(board): + connection = kanban_db.connect(board=board) + try: + kanban_db.block_task( + connection, task_id, reason=reason, kind=kind, + expected_run_id=run_id, + ) + finally: + connection.close() + except Exception as error: # noqa: BLE001 - a failed park must not wedge the pass + _defer(f"{board}/{task_id} block", error) + + +def recover_results(pool: Any) -> None: + """Re-apply each durable result in isolation from its neighbours.""" + for record in pool.store.pending_results(): + try: + pool.finalize(record) + _settled(_context(record)) + except Exception as error: # noqa: BLE001 - one bad result must not block the rest + print( + f"result recovery deferred: {type(error).__name__}: {error}", + file=sys.stderr, + flush=True, + ) + + +def _lease_failure_state( + kanban_db: Any, connection: Any, binding: dict[str, Any], run_id: int +) -> str: + """Classify one exhausted lease from authoritative Kanban evidence only.""" + task = kanban_db.get_task(connection, binding["task_id"]) + current = ( + canonical_run_id(_task_value(task, "current_run_id", None)) + if task is not None + else None + ) + if current != run_id: + return "stale" + if str(_task_value(task, "status", "") or "") != "running": + return "finalized" + if kanban_db.block_task( + connection, + binding["task_id"], + reason=( + "Distributed worker lease expired after " + f"{binding['attempt']} fenced attempts" + ), + kind="transient", + expected_run_id=run_id, + ): + return "finalized" + # Kanban refused the exact-run park yet still reports the run as running, so + # there is no terminal evidence. Keep the row for the next pass to retry. + raise ProtocolError("Kanban refused the exact-run lease park") + + +def record_lease_failure(pool: Any, record: dict[str, Any]) -> None: + """Record one exhausted lease against its exact Kanban run, idempotently. + + The row stays ``lease_failed`` unless this pass obtains authoritative + evidence: the run moved on or vanished (``stale``), Kanban already parked it + (``finalized``), or this call parked it (``finalized``). Anything else -- + including a refused write or an unreadable board -- leaves the row for the + next pass, so no run is ever silently dropped. + """ + from hermes_cli import kanban_db + + binding = pool._binding(record) + run_id = canonical_run_id(binding["run_id"]) + if run_id is None: + pool.store.finalize(binding, "stale") + return + with pool._kanban_lock, kanban_db.scoped_current_board(binding["board"]): + connection = kanban_db.connect(board=binding["board"]) + try: + state = _lease_failure_state(kanban_db, connection, binding, run_id) + finally: + connection.close() + pool.store.finalize(binding, state) + _settled(_context(record)) + + +def recover_lease_failures(pool: Any) -> None: + """Retry every released-but-unrecorded lease failure, each in isolation.""" + for record in pool.store.failed_leases(): + try: + record_lease_failure(pool, record) + except Exception as error: # noqa: BLE001 - retried on the next pass + _defer(f"{_context(record)} lease park", error) + + +def expire_leases(pool: Any) -> None: + """Fence dead attempts, then drain every unconfirmed terminal record.""" + pool.store.expire_leases() + recover_lease_failures(pool) + + +def _release_moved_run(pool: Any, kanban_db: Any, record: dict[str, Any]) -> None: + """Release one durable assignment whose Kanban run is no longer ours.""" + with kanban_db.scoped_current_board(record["board"]): + connection = kanban_db.connect(board=record["board"]) + try: + task = kanban_db.get_task(connection, record["task_id"]) + current_run_id = ( + canonical_run_id(_task_value(task, "current_run_id", None)) + if task + else None + ) + current = str(current_run_id or "") + status = str(_task_value(task, "status", "") or "") if task else "" + finally: + connection.close() + if current != record["run_id"] or status != "running": + pool.store.finalize(record, "stale") + + +def _adopt_task( + pool: Any, + kanban_db: Any, + connection: Any, + board: str, + task: Any, + ordinals: list[int], + known: set[tuple[str, str, str]], +) -> None: + """Adopt one orphaned pool-eligible run onto a free ordinal.""" + task_id = str(_task_value(task, "id", "") or "") + raw_run_id = canonical_run_id(_task_value(task, "current_run_id", None)) + run_id = str(raw_run_id or "") + assignee = str(_task_value(task, "assignee", "") or "") + if ( + not task_id + or not run_id + or str(_task_value(task, "status", "")) != "running" + or not assignee.startswith("cli-") + or not distributed_workspace_eligible(task) + or (board, task_id, run_id) in known + ): + return + try: + payload = pool.assignment_payload(kanban_db, connection, task, board) + except RECOVERABLE_BOARD_ERRORS as error: + _defer(f"{board}/{task_id} recovery payload", error) + return + except Exception as error: # noqa: BLE001 - a real capability failure of this run + _block_exact( + kanban_db, board, task_id, raw_run_id, + "Distributed assignment recovery failed: " + f"{type(error).__name__}: {error}", + ) + return + binding = { + "board": board, "task_id": task_id, "run_id": run_id, + "worker_ordinal": ordinals[0], "attempt": 1, + } + try: + pool.store.add(binding, payload) + except ProtocolError as error: + # A durable row already owns this identity or ordinal. Leave the ordinal + # free and leave Kanban untouched; a later pass revisits it. + _defer(f"{board}/{task_id} adoption", error) + return + known.add((board, task_id, run_id)) + ordinals.pop(0) + + +def _adopt_board( + pool: Any, + kanban_db: Any, + board: str, + ordinals: list[int], + known: set[tuple[str, str, str]], +) -> None: + """Adopt this board's orphaned runs without touching any other board.""" + with kanban_db.scoped_current_board(board): + connection = kanban_db.connect(board=board) + try: + for task in kanban_db.list_tasks(connection): + if not ordinals: + return + _adopt_task(pool, kanban_db, connection, board, task, ordinals, known) + finally: + connection.close() + + +def reconcile(pool: Any) -> None: + """Recover the narrow claim/assignment crash gaps without double execution.""" + from hermes_cli import kanban_db + + for record in pool.store.active_assignments(): + try: + _release_moved_run(pool, kanban_db, record) + except Exception as error: # noqa: BLE001 - other assignments still reconcile + _defer(f"{_context(record)} release", error) + ordinals = pool.store.available_ordinals() + if not ordinals: + return + known = pool.store.known_runs() + try: + boards = kanban_db.list_boards(include_archived=False) + except RECOVERABLE_BOARD_ERRORS as error: + _defer("board-registry", error) + return + _settled("board-registry") + for raw_board in boards: + board = cli_lane_dispatch._board_slug(raw_board) + if not board or not ordinals: + continue + try: + _adopt_board(pool, kanban_db, board, ordinals, known) + _settled(f"{board} adoption") + except Exception as error: # noqa: BLE001 - later boards must still run + _defer(f"{board} adoption", error) + + +def _materialize( + pool: Any, + kanban_db: Any, + board: str, + task_id: str, + ordinal: int, + known: set[tuple[str, str, str]], +) -> None: + """Turn one freshly claimed run into exactly one durable assignment.""" + database_run_id: int | None = None + try: + with kanban_db.scoped_current_board(board): + connection = kanban_db.connect(board=board) + try: + task = kanban_db.get_task(connection, task_id) + if task is None: + return + raw_run = canonical_run_id( + _task_value(task, "current_run_id", None) + ) + if not distributed_workspace_eligible(task): + if raw_run is not None: + kanban_db.block_task( + connection, task_id, + reason=( + "Distributed claim fenced because an existing " + "workspace is owned by the local lane" + ), + kind="capability", expected_run_id=raw_run, + ) + return + if raw_run is None: + raise ProjectPolicyError("claimed task has no canonical run ID") + database_run_id = raw_run + payload = pool.assignment_payload(kanban_db, connection, task, board) + finally: + connection.close() + run_id = str(database_run_id) + if (board, task_id, run_id) in known: + return + pool.store.add( + { + "board": board, "task_id": task_id, "run_id": run_id, + "worker_ordinal": ordinal, "attempt": 1, + }, + payload, + ) + known.add((board, task_id, run_id)) + _settled(f"{board}/{task_id} dispatch") + except (ProtocolError, *RECOVERABLE_BOARD_ERRORS) as error: + # Coordinator-side storage/identity conditions. Never convert these into + # a Kanban capability block; reconcile() adopts the claim on a later pass. + _defer(f"{board}/{task_id} dispatch", error) + except Exception as error: # noqa: BLE001 - a real capability failure of this run + _block_exact( + kanban_db, board, task_id, database_run_id, + "Distributed assignment preparation failed: " + f"{type(error).__name__}: {error}", + ) + + +def dispatch(pool: Any) -> None: + """Claim up to the free ordinal count and materialize assignments.""" + from hermes_cli import kanban_db + + ordinals = pool.store.available_ordinals() + if not ordinals: + return + try: + # Only storage faults are absorbed here. An incompatible claim API must + # still surface, because silently skipping the eligibility predicate is + # how the pool would start claiming the local lane's owned workspaces. + claimed = cli_lane_dispatch.claim_ready( + set(), len(ordinals), + lambda _board, task: distributed_workspace_eligible(task), + ) + except RECOVERABLE_BOARD_ERRORS as error: + _defer("claim-ready", error) + return + _settled("claim-ready") + known = pool.store.known_runs() + for ordinal, (board, task_id) in zip(ordinals, claimed, strict=False): + _materialize(pool, kanban_db, board, task_id, ordinal, known) diff --git a/services/hermes/scripts/execution_pool_protocol.py b/services/hermes/scripts/execution_pool_protocol.py index 599279be..007b8b0f 100644 --- a/services/hermes/scripts/execution_pool_protocol.py +++ b/services/hermes/scripts/execution_pool_protocol.py @@ -8,7 +8,6 @@ import hmac import json import os import re -import sqlite3 import stat import threading import time @@ -237,260 +236,15 @@ def parse_wire(body: bytes) -> dict[str, Any]: return value -class PoolStore: - """Coordinator-owned durable assignments; never stores provider secrets.""" +def __getattr__(name: str) -> Any: + """Expose the durable store on the stable protocol surface. - def __init__(self, path: Path, lease_seconds: int = 90): - self.path = path - self.lease_seconds = max(60, min(int(lease_seconds), 600)) - self._lock = threading.RLock() - path.parent.mkdir(parents=True, exist_ok=True) - self._initialize() + ``PoolStore`` moved to :mod:`execution_pool_store` to keep both modules under + the managed line ceiling. It is resolved lazily here so the store can keep + importing this module's primitives without an import cycle. + """ + if name == "PoolStore": + from execution_pool_store import PoolStore - def _connect(self) -> sqlite3.Connection: - connection = sqlite3.connect(self.path, timeout=10, isolation_level=None) - connection.row_factory = sqlite3.Row - connection.execute("PRAGMA journal_mode=WAL") - connection.execute("PRAGMA synchronous=FULL") - connection.execute("PRAGMA busy_timeout=10000") - return connection - - def _initialize(self) -> None: - with self._connect() as connection: - connection.executescript( - """ - CREATE TABLE IF NOT EXISTS assignments ( - board TEXT NOT NULL, task_id TEXT NOT NULL, run_id TEXT NOT NULL, - worker_ordinal INTEGER NOT NULL CHECK(worker_ordinal BETWEEN 0 AND 2), - attempt INTEGER NOT NULL, assignment_digest TEXT NOT NULL, - payload_json TEXT NOT NULL, state TEXT NOT NULL, - lease_until REAL NOT NULL DEFAULT 0, last_heartbeat REAL NOT NULL DEFAULT 0, - result_digest TEXT, result_json TEXT, created_at REAL NOT NULL, - updated_at REAL NOT NULL, PRIMARY KEY(board, task_id, run_id) - ); - CREATE UNIQUE INDEX IF NOT EXISTS one_live_assignment_per_worker - ON assignments(worker_ordinal) WHERE state IN ('assigned','running','result'); - CREATE TABLE IF NOT EXISTS deliveries ( - delivery_id TEXT PRIMARY KEY, kind TEXT NOT NULL, digest TEXT NOT NULL, - received_at REAL NOT NULL - ); - """ - ) - - @staticmethod - def _record(row: sqlite3.Row | None) -> dict[str, Any] | None: - if row is None: - return None - value = dict(row) - value["payload"] = json.loads(value.pop("payload_json")) - if value.get("result_json"): - value["result"] = json.loads(value["result_json"]) - return value - - def add(self, binding: dict[str, Any], payload: dict[str, Any]) -> bool: - """Create exactly one assignment for a claimed run and free ordinal.""" - now = time.time() - digest = payload_digest(payload) - values = ( - binding["board"], binding["task_id"], binding["run_id"], - binding["worker_ordinal"], binding["attempt"], digest, - canonical_json(payload).decode(), "assigned", now, now, - ) - with self._lock, self._connect() as connection: - try: - connection.execute("BEGIN IMMEDIATE") - connection.execute( - """INSERT INTO assignments - (board,task_id,run_id,worker_ordinal,attempt,assignment_digest, - payload_json,state,created_at,updated_at) - VALUES (?,?,?,?,?,?,?,?,?,?)""", - values, - ) - connection.commit() - return True - except sqlite3.IntegrityError as error: - connection.rollback() - existing = connection.execute( - """SELECT assignment_digest,worker_ordinal,attempt FROM assignments - WHERE board=? AND task_id=? AND run_id=?""", - values[:3], - ).fetchone() - if existing and tuple(existing) == ( - digest, binding["worker_ordinal"], binding["attempt"] - ): - return False - if existing: - raise ProtocolError("conflicting duplicate assignment") from error - raise ProtocolError("worker ordinal already has a live assignment") from error - - def available_ordinals(self) -> list[int]: - with self._connect() as connection: - rows = connection.execute( - "SELECT worker_ordinal FROM assignments WHERE state IN ('assigned','running','result')" - ).fetchall() - occupied = {int(row[0]) for row in rows} - return [ordinal for ordinal in range(3) if ordinal not in occupied] - - def active_assignments(self) -> list[dict[str, Any]]: - with self._connect() as connection: - rows = connection.execute( - "SELECT * FROM assignments WHERE state IN ('assigned','running','result')" - ).fetchall() - return [self._record(row) or {} for row in rows] - - def offer(self, ordinal: int) -> dict[str, Any] | None: - """Return the ordinal's durable assignment, preserving restart identity.""" - now = time.time() - with self._lock, self._connect() as connection: - connection.execute("BEGIN IMMEDIATE") - row = connection.execute( - """SELECT * FROM assignments WHERE worker_ordinal=? - AND state IN ('assigned','running') ORDER BY created_at LIMIT 1""", - (ordinal,), - ).fetchone() - if row is not None: - connection.execute( - """UPDATE assignments SET state='running',lease_until=?,last_heartbeat=?,updated_at=? - WHERE board=? AND task_id=? AND run_id=?""", - (now + self.lease_seconds, now, now, row["board"], row["task_id"], row["run_id"]), - ) - connection.commit() - return self._record(row) - - def _matching(self, connection: sqlite3.Connection, envelope: dict[str, Any]) -> sqlite3.Row: - row = connection.execute( - "SELECT * FROM assignments WHERE board=? AND task_id=? AND run_id=?", - (envelope["board"], envelope["task_id"], envelope["run_id"]), - ).fetchone() - if row is None: - raise ProtocolError("assignment is unknown or stale") - if int(row["worker_ordinal"]) != int(envelope["worker_ordinal"]): - raise ProtocolError("worker ordinal does not own this assignment") - if int(row["attempt"]) != int(envelope["attempt"]): - raise ProtocolError("assignment attempt is stale") - return row - - def heartbeat(self, envelope: dict[str, Any]) -> tuple[bool, bool]: - now = time.time() - delivery_digest = payload_digest( - { - name: envelope[name] - for name in ( - "kind", "board", "task_id", "run_id", "worker_ordinal", - "attempt", "payload_digest", - ) - } - ) - with self._lock, self._connect() as connection: - connection.execute("BEGIN IMMEDIATE") - duplicate = connection.execute( - "SELECT digest FROM deliveries WHERE delivery_id=?", - (envelope["delivery_id"],), - ).fetchone() - row = self._matching(connection, envelope) - if row["state"] not in {"assigned", "running"}: - raise ProtocolError("assignment is no longer running") - if row["state"] == "running" and float(row["lease_until"]) < now: - raise ProtocolError("assignment lease expired") - if duplicate and duplicate[0] != delivery_digest: - raise ProtocolError("delivery identifier was reused") - if not duplicate: - connection.execute( - "INSERT INTO deliveries VALUES (?,?,?,?)", - (envelope["delivery_id"], "heartbeat", delivery_digest, now), - ) - connection.execute( - "UPDATE assignments SET state='running',lease_until=?,last_heartbeat=?,updated_at=? WHERE board=? AND task_id=? AND run_id=?", - (now + self.lease_seconds, now, now, envelope["board"], envelope["task_id"], envelope["run_id"]), - ) - connection.commit() - return True, bool(duplicate) - - def accept_result(self, envelope: dict[str, Any]) -> tuple[dict[str, Any], bool]: - now = time.time() - with self._lock, self._connect() as connection: - connection.execute("BEGIN IMMEDIATE") - row = self._matching(connection, envelope) - digest = envelope["payload_digest"] - if row["result_digest"]: - if row["result_digest"] != digest: - raise ProtocolError("conflicting result for completed delivery") - connection.rollback() - return self._record(row) or {}, True - if row["state"] not in {"assigned", "running"}: - raise ProtocolError("assignment cannot accept a result") - if row["state"] == "running" and float(row["lease_until"]) < now: - raise ProtocolError("assignment lease expired") - connection.execute( - """UPDATE assignments SET state='result',result_digest=?,result_json=?,updated_at=? - WHERE board=? AND task_id=? AND run_id=?""", - (digest, canonical_json(envelope["payload"]).decode(), now, - envelope["board"], envelope["task_id"], envelope["run_id"]), - ) - connection.commit() - row = connection.execute( - "SELECT * FROM assignments WHERE board=? AND task_id=? AND run_id=?", - (envelope["board"], envelope["task_id"], envelope["run_id"]), - ).fetchone() - return self._record(row) or {}, False - - def pending_results(self) -> list[dict[str, Any]]: - with self._connect() as connection: - rows = connection.execute("SELECT * FROM assignments WHERE state='result' ORDER BY updated_at").fetchall() - return [self._record(row) or {} for row in rows] - - def expire_leases( - self, *, now: float | None = None, max_attempts: int = 3 - ) -> list[dict[str, Any]]: - """Fence expired attempts and re-offer or terminally release their ordinals.""" - current = time.time() if now is None else float(now) - maximum = max(1, min(int(max_attempts), 10)) - changed: list[dict[str, Any]] = [] - with self._lock, self._connect() as connection: - connection.execute("BEGIN IMMEDIATE") - rows = connection.execute( - """SELECT * FROM assignments WHERE state='running' - AND lease_until > 0 AND lease_until < ? ORDER BY updated_at""", - (current,), - ).fetchall() - for row in rows: - if int(row["attempt"]) >= maximum: - state, attempt = "lease_failed", int(row["attempt"]) - else: - state, attempt = "assigned", int(row["attempt"]) + 1 - connection.execute( - """UPDATE assignments SET state=?,attempt=?,lease_until=0, - last_heartbeat=0,updated_at=? WHERE board=? AND task_id=? AND run_id=? - AND attempt=? AND state='running'""", - ( - state, attempt, current, row["board"], row["task_id"], - row["run_id"], row["attempt"], - ), - ) - updated = connection.execute( - "SELECT * FROM assignments WHERE board=? AND task_id=? AND run_id=?", - (row["board"], row["task_id"], row["run_id"]), - ).fetchone() - if updated is not None: - changed.append(self._record(updated) or {}) - connection.commit() - return changed - - def finalize(self, binding: dict[str, Any], state: str) -> None: - if state not in {"finalized", "stale"}: - raise ProtocolError("invalid terminal assignment state") - with self._lock, self._connect() as connection: - connection.execute( - "UPDATE assignments SET state=?,updated_at=? WHERE board=? AND task_id=? AND run_id=? AND worker_ordinal=? AND attempt=?", - (state, time.time(), *(binding[name] for name in ("board", "task_id", "run_id", "worker_ordinal", "attempt"))), - ) - - def garbage_collect(self, retention_seconds: int) -> int: - cutoff = time.time() - max(3600, retention_seconds) - with self._lock, self._connect() as connection: - cursor = connection.execute( - "DELETE FROM assignments WHERE state IN ('finalized','stale') AND updated_at < ?", - (cutoff,), - ) - connection.execute("DELETE FROM deliveries WHERE received_at < ?", (cutoff,)) - return int(cursor.rowcount) + return PoolStore + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/services/hermes/scripts/execution_pool_scm.py b/services/hermes/scripts/execution_pool_scm.py index 6badfbb9..393031db 100644 --- a/services/hermes/scripts/execution_pool_scm.py +++ b/services/hermes/scripts/execution_pool_scm.py @@ -144,6 +144,46 @@ def _state_path(envelope: dict[str, Any]) -> Path: return path +def submission_refs(branch: str, attempt: int, head: str) -> tuple[str, ...]: + """Candidate push refs for one attempt, in order of preference. + + The broker accepts branch *creation* only, so a retry that adds commits can + never update the ref a previous attempt already published, and no attempt may + ever move a protected ref. Each fallback is therefore a fresh name in the same + reviewed namespace -- bound first to the exact attempt, then to the exact + content -- so every push stays a creation and stays idempotent under replay. + """ + candidates = ( + branch, + f"{branch}-attempt-{max(1, int(attempt))}", + f"{branch}-{head[:12]}", + ) + allowed: list[str] = [] + for candidate in dict.fromkeys(candidates): + try: + allowed.append(validate_branch(candidate, feature=True)) + except ValueError: + continue + if not allowed: + raise ProtocolError("no reviewed branch name is available for this attempt") + return tuple(allowed) + + +def _remote_heads(destination: Path, refs: tuple[str, ...]) -> dict[str, str]: + """Read the broker's current head for each candidate ref, read-only.""" + output = _run( + "ls-remote", "--heads", "hermes-broker", + *(f"refs/heads/{ref}" for ref in refs), + cwd=destination, timeout=300, + ) + heads: dict[str, str] = {} + for line in output.splitlines(): + fields = line.split() + if len(fields) == 2 and fields[1].startswith("refs/heads/"): + heads[fields[1].removeprefix("refs/heads/")] = fields[0] + return heads + + def _workspace_identity(destination: Path, repo: str, branch: str) -> str: """Validate the checkout using bounded Git plumbing with all hooks disabled.""" if destination.is_symlink() or not (destination / ".git").is_dir(): @@ -270,12 +310,26 @@ class Boundary: ) if status: raise ProtocolError("workspace has uncommitted or untracked files") - ahead = int(_run("rev-list", "--count", f"{baseline}..{head}", cwd=destination)) - if ahead <= 0: - return {"workspace": str(destination), "branch": branch, "pull_request": ""} - _run( - "push", "hermes-broker", f"HEAD:refs/heads/{branch}", - cwd=destination, timeout=900, + candidates = submission_refs(branch, int(envelope["attempt"]), head) + heads = _remote_heads(destination, candidates) + # A candidate already at this exact head was published by an earlier + # attempt or an interrupted one: adopt it instead of pushing again, so + # replay is a no-op and prior work is never re-derived or dropped. + published = next( + (ref for ref in candidates if heads.get(ref) == head), None ) - pull = self._draft(repo, branch, base, head, title, body) - return {"workspace": str(destination), "branch": branch, "pull_request": pull} + ahead = int(_run("rev-list", "--count", f"{baseline}..{head}", cwd=destination)) + if ahead <= 0 and published is None: + return {"workspace": str(destination), "branch": branch, "pull_request": ""} + target = published or next( + (ref for ref in candidates if ref not in heads), None + ) + if target is None: + raise ProtocolError("every reviewed branch name for this attempt is taken") + if published is None: + _run( + "push", "hermes-broker", f"HEAD:refs/heads/{target}", + cwd=destination, timeout=900, + ) + pull = self._draft(repo, target, base, head, title, body) + return {"workspace": str(destination), "branch": target, "pull_request": pull} diff --git a/services/hermes/scripts/execution_pool_server.py b/services/hermes/scripts/execution_pool_server.py index b2a01713..75e08b64 100644 --- a/services/hermes/scripts/execution_pool_server.py +++ b/services/hermes/scripts/execution_pool_server.py @@ -9,6 +9,7 @@ import sqlite3 import sys import threading import time +from collections.abc import Iterable from http.server import BaseHTTPRequestHandler from pathlib import Path from typing import Any @@ -17,12 +18,12 @@ from execution_pool_protocol import ( MAX_WIRE_BYTES, PROTOCOL_VERSION, BoundedHTTPServer, - PoolStore, ProtocolError, canonical_json, parse_wire, read_key, ) +from execution_pool_store import PoolStore DATA_ROOT = Path(os.environ.get("HERMES_HOME", "/opt/data")) @@ -92,19 +93,41 @@ def handler_factory(coordinator: Any) -> type[BaseHTTPRequestHandler]: return Handler +def maintenance_cycle(operations: Iterable[Any]) -> int: + """Run one maintenance pass; a failing operation never aborts the others. + + Startup uses this too. A durable row that one pass cannot process must not + stop the coordinator from binding its port, because the store lives on a PVC + and would otherwise turn a single bad row into a permanent crash loop. + """ + deferred = 0 + for operation in operations: + try: + operation() + except Exception as error: # noqa: BLE001 - the loop is the recovery mechanism + deferred += 1 + print( + f"pool maintenance deferred: {type(error).__name__}: {error}", + file=sys.stderr, + flush=True, + ) + return deferred + + def run(coordinator_type: type[Any]) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--once", action="store_true") args = parser.parse_args() store = PoolStore(STATE_ROOT / "assignments.db") coordinator = coordinator_type(read_key(KEY_PATH), store) - for operation in ( + operations = ( coordinator.expire_leases, coordinator.recover_results, coordinator.reconcile, coordinator.dispatch, - ): - operation() + lambda: store.garbage_collect(RETENTION_SECONDS), + ) + maintenance_cycle(operations) if args.once: return 0 server = BoundedHTTPServer( @@ -112,19 +135,5 @@ def run(coordinator_type: type[Any]) -> int: ) threading.Thread(target=server.serve_forever, daemon=True).start() while True: - for operation in ( - coordinator.expire_leases, - coordinator.recover_results, - coordinator.reconcile, - coordinator.dispatch, - lambda: store.garbage_collect(RETENTION_SECONDS), - ): - try: - operation() - except Exception as error: - print( - f"pool maintenance deferred: {type(error).__name__}: {error}", - file=sys.stderr, - flush=True, - ) + maintenance_cycle(operations) time.sleep(5) diff --git a/services/hermes/scripts/execution_pool_store.py b/services/hermes/scripts/execution_pool_store.py new file mode 100644 index 00000000..1022f78a --- /dev/null +++ b/services/hermes/scripts/execution_pool_store.py @@ -0,0 +1,321 @@ +#!/usr/bin/env python3 +"""Durable, restart-safe assignment state for the Hermes execution pool. + +The store owns every transition of a claimed Kanban run through the pool. It +deliberately keeps two invariants that the coordinator depends on for recovery: + +* ``lease_failed`` is a *retryable* state, not a terminal one. A row reaches it + when the pool has released the ordinal but has not yet confirmed the outcome + with Kanban, so the coordinator must keep retrying it. +* Only ``finalized`` and ``stale`` are terminal, and only terminal rows are ever + garbage-collected. A row is therefore never removed before the pool holds + authoritative evidence about what happened to its run. +""" + +from __future__ import annotations + +import json +import sqlite3 +import threading +import time +from pathlib import Path +from typing import Any + +from execution_pool_protocol import ProtocolError, canonical_json, payload_digest + + +LIVE_STATES = ("assigned", "running", "result") +TERMINAL_STATES = ("finalized", "stale") +LEASE_FAILED = "lease_failed" +SCHEMA = """ +CREATE TABLE IF NOT EXISTS assignments ( + board TEXT NOT NULL, task_id TEXT NOT NULL, run_id TEXT NOT NULL, + worker_ordinal INTEGER NOT NULL CHECK(worker_ordinal BETWEEN 0 AND 2), + attempt INTEGER NOT NULL, assignment_digest TEXT NOT NULL, + payload_json TEXT NOT NULL, state TEXT NOT NULL, + lease_until REAL NOT NULL DEFAULT 0, last_heartbeat REAL NOT NULL DEFAULT 0, + result_digest TEXT, result_json TEXT, created_at REAL NOT NULL, + updated_at REAL NOT NULL, PRIMARY KEY(board, task_id, run_id) +); +CREATE UNIQUE INDEX IF NOT EXISTS one_live_assignment_per_worker + ON assignments(worker_ordinal) WHERE state IN ('assigned','running','result'); +CREATE TABLE IF NOT EXISTS deliveries ( + delivery_id TEXT PRIMARY KEY, kind TEXT NOT NULL, digest TEXT NOT NULL, + received_at REAL NOT NULL +); +""" + + +class PoolStore: + """Coordinator-owned durable assignments; never stores provider secrets.""" + + def __init__(self, path: Path, lease_seconds: int = 90): + self.path = path + self.lease_seconds = max(60, min(int(lease_seconds), 600)) + self._lock = threading.RLock() + path.parent.mkdir(parents=True, exist_ok=True) + self._initialize() + + def _connect(self) -> sqlite3.Connection: + connection = sqlite3.connect(self.path, timeout=10, isolation_level=None) + connection.row_factory = sqlite3.Row + connection.execute("PRAGMA journal_mode=WAL") + connection.execute("PRAGMA synchronous=FULL") + connection.execute("PRAGMA busy_timeout=10000") + return connection + + def _initialize(self) -> None: + with self._connect() as connection: + connection.executescript(SCHEMA) + + @staticmethod + def _record(row: sqlite3.Row | None) -> dict[str, Any] | None: + if row is None: + return None + value = dict(row) + value["payload"] = json.loads(value.pop("payload_json")) + if value.get("result_json"): + value["result"] = json.loads(value["result_json"]) + return value + + def _select(self, sql: str) -> list[dict[str, Any]]: + with self._connect() as connection: + rows = connection.execute(sql).fetchall() + return [self._record(row) or {} for row in rows] + + def add(self, binding: dict[str, Any], payload: dict[str, Any]) -> bool: + """Create exactly one assignment for a claimed run and free ordinal.""" + now = time.time() + digest = payload_digest(payload) + values = ( + binding["board"], binding["task_id"], binding["run_id"], + binding["worker_ordinal"], binding["attempt"], digest, + canonical_json(payload).decode(), "assigned", now, now, + ) + with self._lock, self._connect() as connection: + try: + connection.execute("BEGIN IMMEDIATE") + connection.execute( + """INSERT INTO assignments + (board,task_id,run_id,worker_ordinal,attempt,assignment_digest, + payload_json,state,created_at,updated_at) + VALUES (?,?,?,?,?,?,?,?,?,?)""", + values, + ) + connection.commit() + return True + except sqlite3.IntegrityError as error: + connection.rollback() + existing = connection.execute( + """SELECT assignment_digest,worker_ordinal,attempt FROM assignments + WHERE board=? AND task_id=? AND run_id=?""", + values[:3], + ).fetchone() + if existing and tuple(existing) == ( + digest, binding["worker_ordinal"], binding["attempt"] + ): + return False + if existing: + raise ProtocolError("conflicting duplicate assignment") from error + raise ProtocolError("worker ordinal already has a live assignment") from error + + def available_ordinals(self) -> list[int]: + with self._connect() as connection: + rows = connection.execute( + "SELECT worker_ordinal FROM assignments WHERE state IN ('assigned','running','result')" + ).fetchall() + occupied = {int(row[0]) for row in rows} + return [ordinal for ordinal in range(3) if ordinal not in occupied] + + def active_assignments(self) -> list[dict[str, Any]]: + return self._select( + "SELECT * FROM assignments WHERE state IN ('assigned','running','result')" + ) + + def known_runs(self) -> set[tuple[str, str, str]]: + """Every run identity this store already owns a row for, in any state. + + Adoption must consult this rather than only the live states: re-adding a + run that still has a retrying or not-yet-collected row would collide with + ``PRIMARY KEY(board, task_id, run_id)`` under a different payload digest. + """ + with self._connect() as connection: + rows = connection.execute( + "SELECT board,task_id,run_id FROM assignments" + ).fetchall() + return {(str(row[0]), str(row[1]), str(row[2])) for row in rows} + + def offer(self, ordinal: int) -> dict[str, Any] | None: + """Return the ordinal's durable assignment, preserving restart identity.""" + now = time.time() + with self._lock, self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + row = connection.execute( + """SELECT * FROM assignments WHERE worker_ordinal=? + AND state IN ('assigned','running') ORDER BY created_at LIMIT 1""", + (ordinal,), + ).fetchone() + if row is not None: + connection.execute( + """UPDATE assignments SET state='running',lease_until=?,last_heartbeat=?,updated_at=? + WHERE board=? AND task_id=? AND run_id=?""", + (now + self.lease_seconds, now, now, row["board"], row["task_id"], row["run_id"]), + ) + connection.commit() + return self._record(row) + + def _matching(self, connection: sqlite3.Connection, envelope: dict[str, Any]) -> sqlite3.Row: + row = connection.execute( + "SELECT * FROM assignments WHERE board=? AND task_id=? AND run_id=?", + (envelope["board"], envelope["task_id"], envelope["run_id"]), + ).fetchone() + if row is None: + raise ProtocolError("assignment is unknown or stale") + if int(row["worker_ordinal"]) != int(envelope["worker_ordinal"]): + raise ProtocolError("worker ordinal does not own this assignment") + if int(row["attempt"]) != int(envelope["attempt"]): + raise ProtocolError("assignment attempt is stale") + return row + + def heartbeat(self, envelope: dict[str, Any]) -> tuple[bool, bool]: + now = time.time() + delivery_digest = payload_digest( + { + name: envelope[name] + for name in ( + "kind", "board", "task_id", "run_id", "worker_ordinal", + "attempt", "payload_digest", + ) + } + ) + with self._lock, self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + duplicate = connection.execute( + "SELECT digest FROM deliveries WHERE delivery_id=?", + (envelope["delivery_id"],), + ).fetchone() + row = self._matching(connection, envelope) + if row["state"] not in {"assigned", "running"}: + raise ProtocolError("assignment is no longer running") + if row["state"] == "running" and float(row["lease_until"]) < now: + raise ProtocolError("assignment lease expired") + if duplicate and duplicate[0] != delivery_digest: + raise ProtocolError("delivery identifier was reused") + if not duplicate: + connection.execute( + "INSERT INTO deliveries VALUES (?,?,?,?)", + (envelope["delivery_id"], "heartbeat", delivery_digest, now), + ) + connection.execute( + "UPDATE assignments SET state='running',lease_until=?,last_heartbeat=?,updated_at=? WHERE board=? AND task_id=? AND run_id=?", + (now + self.lease_seconds, now, now, envelope["board"], envelope["task_id"], envelope["run_id"]), + ) + connection.commit() + return True, bool(duplicate) + + def accept_result(self, envelope: dict[str, Any]) -> tuple[dict[str, Any], bool]: + now = time.time() + with self._lock, self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + row = self._matching(connection, envelope) + digest = envelope["payload_digest"] + if row["result_digest"]: + if row["result_digest"] != digest: + raise ProtocolError("conflicting result for completed delivery") + connection.rollback() + return self._record(row) or {}, True + if row["state"] not in {"assigned", "running"}: + raise ProtocolError("assignment cannot accept a result") + if row["state"] == "running" and float(row["lease_until"]) < now: + raise ProtocolError("assignment lease expired") + connection.execute( + """UPDATE assignments SET state='result',result_digest=?,result_json=?,updated_at=? + WHERE board=? AND task_id=? AND run_id=?""", + (digest, canonical_json(envelope["payload"]).decode(), now, + envelope["board"], envelope["task_id"], envelope["run_id"]), + ) + connection.commit() + row = connection.execute( + "SELECT * FROM assignments WHERE board=? AND task_id=? AND run_id=?", + (envelope["board"], envelope["task_id"], envelope["run_id"]), + ).fetchone() + return self._record(row) or {}, False + + def pending_results(self) -> list[dict[str, Any]]: + return self._select( + "SELECT * FROM assignments WHERE state='result' ORDER BY updated_at" + ) + + def failed_leases(self) -> list[dict[str, Any]]: + """Rows whose ordinal is released but whose Kanban outcome is unconfirmed. + + The coordinator drains this on every maintenance pass, so a Kanban write + that failed at expiry time is retried instead of stranding the run. + """ + return self._select( + "SELECT * FROM assignments WHERE state='lease_failed' ORDER BY updated_at" + ) + + def expire_leases( + self, *, now: float | None = None, max_attempts: int = 3 + ) -> list[dict[str, Any]]: + """Fence expired attempts and re-offer or terminally release their ordinals.""" + current = time.time() if now is None else float(now) + maximum = max(1, min(int(max_attempts), 10)) + changed: list[dict[str, Any]] = [] + with self._lock, self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + rows = connection.execute( + """SELECT * FROM assignments WHERE state='running' + AND lease_until > 0 AND lease_until < ? ORDER BY updated_at""", + (current,), + ).fetchall() + for row in rows: + if int(row["attempt"]) >= maximum: + state, attempt = LEASE_FAILED, int(row["attempt"]) + else: + state, attempt = "assigned", int(row["attempt"]) + 1 + connection.execute( + """UPDATE assignments SET state=?,attempt=?,lease_until=0, + last_heartbeat=0,updated_at=? WHERE board=? AND task_id=? AND run_id=? + AND attempt=? AND state='running'""", + ( + state, attempt, current, row["board"], row["task_id"], + row["run_id"], row["attempt"], + ), + ) + updated = connection.execute( + "SELECT * FROM assignments WHERE board=? AND task_id=? AND run_id=?", + (row["board"], row["task_id"], row["run_id"]), + ).fetchone() + if updated is not None: + changed.append(self._record(updated) or {}) + connection.commit() + return changed + + def finalize(self, binding: dict[str, Any], state: str) -> bool: + """Apply one terminal state to the exact run, ordinal, and attempt.""" + if state not in TERMINAL_STATES: + raise ProtocolError("invalid terminal assignment state") + with self._lock, self._connect() as connection: + cursor = connection.execute( + "UPDATE assignments SET state=?,updated_at=? WHERE board=? AND task_id=? AND run_id=? AND worker_ordinal=? AND attempt=?", + (state, time.time(), *(binding[name] for name in ("board", "task_id", "run_id", "worker_ordinal", "attempt"))), + ) + return bool(cursor.rowcount) + + def garbage_collect(self, retention_seconds: int) -> int: + """Remove aged rows, and only ones with authoritative terminal evidence. + + ``lease_failed`` is excluded on purpose: dropping a row whose Kanban + outcome was never confirmed would lose the pool's only record that the + run still needs to be surfaced. + """ + cutoff = time.time() - max(3600, retention_seconds) + with self._lock, self._connect() as connection: + cursor = connection.execute( + "DELETE FROM assignments WHERE state IN ('finalized','stale') AND updated_at < ?", + (cutoff,), + ) + connection.execute("DELETE FROM deliveries WHERE received_at < ?", (cutoff,)) + return int(cursor.rowcount) diff --git a/testing/quality_contract.json b/testing/quality_contract.json index c5598914..1a3e10e2 100644 --- a/testing/quality_contract.json +++ b/testing/quality_contract.json @@ -29,10 +29,12 @@ "services/hermes/scm-common/scripts/scm_broker_server.py", "services/hermes/scripts/execution_pool_client.py", "services/hermes/scripts/execution_pool_coordinator.py", + "services/hermes/scripts/execution_pool_maintenance.py", "services/hermes/scripts/execution_pool_project.py", "services/hermes/scripts/execution_pool_protocol.py", "services/hermes/scripts/execution_pool_scm.py", "services/hermes/scripts/execution_pool_server.py", + "services/hermes/scripts/execution_pool_store.py", "services/hermes/scripts/execution_pool_worker.py", "services/hermes/scripts/hermes_coordinator.py", "services/hermes/scripts/node_account_audit.py", @@ -83,10 +85,12 @@ "services/hermes/scm-common/scripts", "services/hermes/scripts/execution_pool_client.py", "services/hermes/scripts/execution_pool_coordinator.py", + "services/hermes/scripts/execution_pool_maintenance.py", "services/hermes/scripts/execution_pool_project.py", "services/hermes/scripts/execution_pool_protocol.py", "services/hermes/scripts/execution_pool_scm.py", "services/hermes/scripts/execution_pool_server.py", + "services/hermes/scripts/execution_pool_store.py", "services/hermes/scripts/execution_pool_worker.py", "services/hermes/scripts/hermes_coordinator.py", "services/hermes/scripts/node_account_audit.py", @@ -262,10 +266,12 @@ "services/hermes/scripts/stage_runtime_access.py", "services/hermes/scripts/execution_pool_client.py", "services/hermes/scripts/execution_pool_coordinator.py", + "services/hermes/scripts/execution_pool_maintenance.py", "services/hermes/scripts/execution_pool_project.py", "services/hermes/scripts/execution_pool_protocol.py", "services/hermes/scripts/execution_pool_scm.py", "services/hermes/scripts/execution_pool_server.py", + "services/hermes/scripts/execution_pool_store.py", "services/hermes/scripts/execution_pool_worker.py", "services/hermes/scripts/cli_lane_board.py", "services/hermes/scripts/cli_lane_capabilities.py", @@ -302,10 +308,12 @@ "services/hermes/scm-common/scripts/scm_broker_server.py", "services/hermes/scripts/execution_pool_client.py", "services/hermes/scripts/execution_pool_coordinator.py", + "services/hermes/scripts/execution_pool_maintenance.py", "services/hermes/scripts/execution_pool_project.py", "services/hermes/scripts/execution_pool_protocol.py", "services/hermes/scripts/execution_pool_scm.py", "services/hermes/scripts/execution_pool_server.py", + "services/hermes/scripts/execution_pool_store.py", "services/hermes/scripts/execution_pool_worker.py", "services/hermes/scripts/hermes_coordinator.py", "services/hermes/scripts/node_account_audit.py", diff --git a/testing/tests/test_hermes_execution_pool.py b/testing/tests/test_hermes_execution_pool.py index 8b9966d5..c82df4d7 100644 --- a/testing/tests/test_hermes_execution_pool.py +++ b/testing/tests/test_hermes_execution_pool.py @@ -348,8 +348,17 @@ def test_three_node_statefulset_contract_and_cross_worker_isolation(): item["metadata"]["name"]: item for item in stateful["spec"]["volumeClaimTemplates"] } - assert set(claims) == {"workspace", "provider-access"} - assert all(item["spec"]["accessModes"] == ["ReadWriteOnce"] for item in claims.values()) + assert set(claims) == {"workspace", "tools", "provider-access"} + # The workspace is co-mounted by this ordinal's mediator, so it must be + # ReadWriteMany: a ReadWriteOnce claim deadlocks on Multi-Attach when a drain + # or preemption separates the two Pods. Everything only the worker mounts + # stays single-mounter. + assert claims["workspace"]["spec"]["accessModes"] == ["ReadWriteMany"] + assert claims["tools"]["spec"]["accessModes"] == ["ReadWriteOnce"] + assert claims["provider-access"]["spec"]["accessModes"] == ["ReadWriteOnce"] + assert all( + item["spec"]["storageClassName"] == "astreae" for item in claims.values() + ) anti = pod["affinity"]["podAntiAffinity"]["requiredDuringSchedulingIgnoredDuringExecution"] assert anti[0]["topologyKey"] == "kubernetes.io/hostname" spread = pod["topologySpreadConstraints"][0] diff --git a/testing/tests/test_hermes_execution_pool_dispatch_v2.py b/testing/tests/test_hermes_execution_pool_dispatch_v2.py index 57dd0f74..c63e9825 100644 --- a/testing/tests/test_hermes_execution_pool_dispatch_v2.py +++ b/testing/tests/test_hermes_execution_pool_dispatch_v2.py @@ -16,6 +16,7 @@ SCM_SCRIPTS = ROOT / "services/hermes/scm-common/scripts" sys.path[:0] = [str(SCRIPTS), str(SCM_SCRIPTS)] import execution_pool_coordinator as coordinator # noqa: E402 +import execution_pool_maintenance as maintenance # noqa: E402 import execution_pool_protocol as protocol # noqa: E402 import execution_pool_server as server # noqa: E402 from testing.tests.test_hermes_execution_pool_coordinator_v2 import ( # noqa: E402 @@ -84,7 +85,7 @@ def test_dispatch_claims_only_pathless_task_with_safe_assignment_branch( observed.append((active, limit, eligible("metis", item))) return [("metis", item.id)] - monkeypatch.setattr(coordinator.cli_lane_dispatch, "claim_ready", claim) + monkeypatch.setattr(maintenance.cli_lane_dispatch, "claim_ready", claim) monkeypatch.setattr( coordinator, "assignment_payload", lambda *_a: assignment_payload(branch="wt/t_deadbeef"), @@ -109,7 +110,7 @@ def test_dispatch_incompatible_unversioned_claim_api_fails_closed( calls.append((active, limit)) return [("metis", owned.id)] - monkeypatch.setattr(coordinator.cli_lane_dispatch, "claim_ready", old_claim) + monkeypatch.setattr(maintenance.cli_lane_dispatch, "claim_ready", old_claim) with pytest.raises(TypeError): pool.dispatch() assert calls == [] @@ -125,7 +126,7 @@ def test_dispatch_workspace_ownership_race_is_exactly_fenced( store = protocol.PoolStore(tmp_path / "pool.db") pool = coordinator.Coordinator(MASTER, store) monkeypatch.setattr( - coordinator.cli_lane_dispatch, + maintenance.cli_lane_dispatch, "claim_ready", lambda *_a: [("metis", owned.id)], ) @@ -154,7 +155,7 @@ def test_dispatch_preparation_failure_is_surfaced_and_full_pool_does_not_claim( store = protocol.PoolStore(tmp_path / "pool.db") pool = coordinator.Coordinator(MASTER, store) monkeypatch.setattr( - coordinator.cli_lane_dispatch, + maintenance.cli_lane_dispatch, "claim_ready", lambda *_a: [("metis", broken.id)], ) @@ -170,7 +171,7 @@ def test_dispatch_preparation_failure_is_surfaced_and_full_pool_does_not_claim( ) pool = coordinator.Coordinator(MASTER, full) monkeypatch.setattr( - coordinator.cli_lane_dispatch, + maintenance.cli_lane_dispatch, "claim_ready", lambda *_a: (_ for _ in ()).throw(AssertionError("must not claim")), ) @@ -285,7 +286,7 @@ def test_reconcile_and_dispatch_cover_empty_error_and_missing_task_paths( empty = protocol.PoolStore(tmp_path / "missing.db") monkeypatch.setattr( - coordinator.cli_lane_dispatch, "claim_ready", lambda *_a: [("metis", "missing")] + maintenance.cli_lane_dispatch, "claim_ready", lambda *_a: [("metis", "missing")] ) coordinator.Coordinator(MASTER, empty).dispatch() diff --git a/testing/tests/test_hermes_execution_pool_isolation.py b/testing/tests/test_hermes_execution_pool_isolation.py new file mode 100644 index 00000000..7bac8be5 --- /dev/null +++ b/testing/tests/test_hermes_execution_pool_isolation.py @@ -0,0 +1,391 @@ +"""Fault isolation contracts for the execution-pool maintenance passes. + +No single poisoned row, task, or board may abort the work queued behind it, and a +coordinator-side fault must never be converted into a Kanban mutation. Startup +maintenance is held to exactly the same standard as the steady-state loop, +because the durable store lives on a PVC and would otherwise crash-loop. +""" + +from __future__ import annotations + +import sqlite3 +import sys +from pathlib import Path + +import pytest + + +ROOT = Path(__file__).parents[2] +SCRIPTS = ROOT / "services/hermes/scripts" +SCM_SCRIPTS = ROOT / "services/hermes/scm-common/scripts" +sys.path[:0] = [str(SCRIPTS), str(SCM_SCRIPTS)] + +import execution_pool_coordinator as coordinator # noqa: E402 +import execution_pool_maintenance as maintenance # noqa: E402 +import execution_pool_protocol as protocol # noqa: E402 +import execution_pool_server as server # noqa: E402 +import execution_pool_store as pool_store # noqa: E402 +from testing.tests.test_hermes_execution_pool_coordinator_v2 import ( # noqa: E402 + MASTER, + assignment_payload, + binding, + install_kanban, + task, +) +from testing.tests.test_hermes_execution_pool_support import ( # noqa: E402 + LOCKED, + clear_pool_deferrals, # noqa: F401 - autouse fixture, resolved by name + exhausted_store, + states, +) + + +def test_one_poisoned_board_and_task_never_abort_the_work_behind_them( + tmp_path, monkeypatch, capsys +): + first = task(id="t_first", current_run_id=41) + second = task(id="t_second", current_run_id=42) + kanban = install_kanban( + monkeypatch, [first, second], tmp_path, boards=["broken", "metis"] + ) + real_list_tasks = kanban.list_tasks + payloads = {"calls": 0} + + def flaky_list_tasks(connection): + if kanban.current == "broken": + raise LOCKED + return real_list_tasks(connection) + + def flaky_payload(_db, _connection, item, _board): + payloads["calls"] += 1 + if item.id == "t_first": + raise LOCKED + return assignment_payload() + + def scoped(board): + kanban.current = board + return kanban.scope(board) + + kanban.current = "" + kanban.scope = kanban.scoped_current_board + kanban.scoped_current_board = scoped + kanban.list_tasks = flaky_list_tasks + monkeypatch.setattr(coordinator, "assignment_payload", flaky_payload) + store = pool_store.PoolStore(tmp_path / "poison.db") + coordinator.Coordinator(MASTER, store).reconcile() + + assert payloads["calls"] == 2 + assert states(store) == [("42", 0, 1, "assigned")] + assert kanban.blocked == [] + captured = capsys.readouterr().err + assert "broken adoption" in captured + assert "t_first recovery payload" in captured + + +def test_adoption_conflict_leaves_the_ordinal_free_and_kanban_untouched( + tmp_path, monkeypatch, capsys +): + live = task() + kanban = install_kanban(monkeypatch, [live], tmp_path) + monkeypatch.setattr( + coordinator, "assignment_payload", + lambda _db, _connection, _task, _board: assignment_payload(), + ) + store = pool_store.PoolStore(tmp_path / "conflict.db") + pool = coordinator.Coordinator(MASTER, store) + + def conflict(_binding, _payload): + raise protocol.ProtocolError("conflicting duplicate assignment") + + monkeypatch.setattr(store, "add", conflict) + pool.reconcile() + assert store.available_ordinals() == [0, 1, 2] + assert kanban.blocked == [] + assert "t_deadbeef adoption" in capsys.readouterr().err + + +def test_dispatch_defers_storage_and_identity_faults_without_blocking_the_run( + tmp_path, monkeypatch, capsys +): + live = task() + kanban = install_kanban(monkeypatch, [live], tmp_path) + monkeypatch.setattr( + maintenance.cli_lane_dispatch, "claim_ready", + lambda *_args, **_kwargs: [("metis", live.id)], + ) + store = pool_store.PoolStore(tmp_path / "dispatch.db") + pool = coordinator.Coordinator(MASTER, store) + + monkeypatch.setattr( + coordinator, "assignment_payload", + lambda *_args: (_ for _ in ()).throw(LOCKED), + ) + pool.dispatch() + assert kanban.blocked == [] + assert store.active_assignments() == [] + assert "metis/t_deadbeef dispatch" in capsys.readouterr().err + + monkeypatch.setattr( + coordinator, "assignment_payload", + lambda _db, _connection, _task, _board: assignment_payload(), + ) + monkeypatch.setattr( + store, "add", + lambda *_args: (_ for _ in ()).throw( + protocol.ProtocolError("worker ordinal already has a live assignment") + ), + ) + pool.dispatch() + assert kanban.blocked == [] + + +def test_dispatch_still_parks_the_exact_run_on_a_real_capability_failure( + tmp_path, monkeypatch +): + live = task() + kanban = install_kanban(monkeypatch, [live], tmp_path) + monkeypatch.setattr( + maintenance.cli_lane_dispatch, "claim_ready", + lambda *_args, **_kwargs: [("metis", live.id)], + ) + monkeypatch.setattr( + coordinator, "assignment_payload", + lambda *_args: (_ for _ in ()).throw(RuntimeError("context exceeds 32KiB")), + ) + store = pool_store.PoolStore(tmp_path / "capability.db") + coordinator.Coordinator(MASTER, store).dispatch() + assert kanban.blocked[-1][0] == "t_deadbeef" + assert kanban.blocked[-1][1]["kind"] == "capability" + assert kanban.blocked[-1][1]["expected_run_id"] == 23 + assert "32KiB" in kanban.blocked[-1][1]["reason"] + + +def test_a_failing_park_write_is_absorbed_rather_than_wedging_the_pass( + tmp_path, monkeypatch, capsys +): + live = task() + kanban = install_kanban(monkeypatch, [live], tmp_path) + kanban.block_task = lambda *_args, **_kwargs: (_ for _ in ()).throw(LOCKED) + monkeypatch.setattr( + maintenance.cli_lane_dispatch, "claim_ready", + lambda *_args, **_kwargs: [("metis", live.id)], + ) + monkeypatch.setattr( + coordinator, "assignment_payload", + lambda *_args: (_ for _ in ()).throw(RuntimeError("bad")), + ) + store = pool_store.PoolStore(tmp_path / "park.db") + coordinator.Coordinator(MASTER, store).dispatch() + assert "t_deadbeef block" in capsys.readouterr().err + + +def test_one_unprocessable_result_never_blocks_the_results_behind_it( + tmp_path, monkeypatch, capsys +): + install_kanban(monkeypatch, [task(), task(id="t_second", current_run_id=24)], tmp_path) + store = pool_store.PoolStore(tmp_path / "results.db") + pool = coordinator.Coordinator(MASTER, store) + handled = [] + + def finalize(record): + if record["task_id"] == "t_deadbeef": + raise protocol.ProtocolError("result payload must be an object") + handled.append(record["task_id"]) + + monkeypatch.setattr( + store, "pending_results", + lambda: [{**binding(), "result": []}, {**binding(task_id="t_second"), "result": {}}], + ) + monkeypatch.setattr(pool, "finalize", finalize) + pool.recover_results() + assert handled == ["t_second"] + assert "result recovery deferred" in capsys.readouterr().err + + +def test_startup_maintenance_is_as_safe_as_the_steady_state_loop( + tmp_path, monkeypatch, capsys +): + """A poisoned store must not stop the coordinator from binding its port.""" + live = task() + kanban = install_kanban(monkeypatch, [live], tmp_path) + kanban.block_task = lambda *_args, **_kwargs: (_ for _ in ()).throw(LOCKED) + kanban.list_boards = lambda include_archived=False: (_ for _ in ()).throw( + RuntimeError("board registry unavailable") + ) + key = tmp_path / "key" + key.write_bytes(b"m" * 48) + key.chmod(0o600) + monkeypatch.setattr(server, "STATE_ROOT", tmp_path / "state") + monkeypatch.setattr(server, "KEY_PATH", key) + monkeypatch.setattr(sys, "argv", ["execution_pool_server", "--once"]) + exhausted_store(tmp_path / "state/assignments.db") + + assert server.run(coordinator.Coordinator) == 0 + captured = capsys.readouterr().err + assert "pool maintenance deferred" in captured + row = sqlite3.connect(tmp_path / "state/assignments.db").execute( + "SELECT state FROM assignments" + ).fetchone() + assert row[0] == "lease_failed" + + +def test_maintenance_cycle_counts_deferrals_and_keeps_running(capsys): + calls = [] + count = server.maintenance_cycle( + ( + lambda: calls.append("first"), + lambda: (_ for _ in ()).throw(RuntimeError("boom")), + lambda: calls.append("last"), + ) + ) + assert count == 1 + assert calls == ["first", "last"] + assert "pool maintenance deferred: RuntimeError: boom" in capsys.readouterr().err + + +def test_the_poison_row_survives_restart_on_the_same_pvc_and_then_resolves( + tmp_path, monkeypatch, capsys +): + """The store lives on hermes-agent-home, so recovery must survive restarts.""" + live = task() + kanban = install_kanban(monkeypatch, [live], tmp_path) + kanban.block_task = lambda *_args, **_kwargs: (_ for _ in ()).throw(LOCKED) + database = tmp_path / "pvc/assignments.db" + first = exhausted_store(database) + coordinator.Coordinator(MASTER, first).expire_leases() + assert states(first) == [("23", 0, 3, "lease_failed")] + del first + capsys.readouterr() + + # A fresh process attaches the same PVC path and finds the row waiting. + restarted = pool_store.PoolStore(database) + assert restarted.failed_leases()[0]["state"] == "lease_failed" + assert restarted.available_ordinals() == [0, 1, 2] + kanban.block_task = lambda _connection, task_id, **values: ( + kanban.blocked.append((task_id, values)) or True + ) + coordinator.Coordinator(MASTER, restarted).expire_leases() + assert states(restarted) == [("23", 0, 3, "finalized")] + assert kanban.blocked[-1][1]["expected_run_id"] == 23 + + +def test_a_store_write_fault_while_recording_leaves_the_row_retryable( + tmp_path, monkeypatch, capsys +): + live = task() + install_kanban(monkeypatch, [live], tmp_path) + store = exhausted_store(tmp_path / "store-fault.db") + pool = coordinator.Coordinator(MASTER, store) + real_finalize = store.finalize + monkeypatch.setattr( + store, "finalize", lambda *_args: (_ for _ in ()).throw(LOCKED) + ) + pool.expire_leases() + assert states(store) == [("23", 0, 3, "lease_failed")] + assert "lease park" in capsys.readouterr().err + + monkeypatch.setattr(store, "finalize", real_finalize) + pool.expire_leases() + assert states(store) == [("23", 0, 3, "finalized")] + + +def test_finalize_reports_whether_the_exact_row_changed(tmp_path): + store = pool_store.PoolStore(tmp_path / "finalize.db") + store.add(binding(), assignment_payload()) + assert store.finalize(binding(attempt=9), "stale") is False + assert store.finalize(binding(), "stale") is True + with pytest.raises(protocol.ProtocolError, match="terminal"): + store.finalize(binding(), "lease_failed") + + +def test_deferral_reporting_is_deduplicated_and_bounded(capsys): + maintenance._defer("ctx", LOCKED) + maintenance._defer("ctx", LOCKED) + assert capsys.readouterr().err.count("pool work deferred") == 1 + maintenance._defer("ctx", RuntimeError("changed")) + assert "changed" in capsys.readouterr().err + for index in range(maintenance.MAX_DEFERRALS + 1): + maintenance._defer(f"ctx-{index}", LOCKED) + assert len(maintenance.DEFERRALS) <= maintenance.MAX_DEFERRALS + capsys.readouterr() + + +def test_protocol_exposes_only_the_store_on_its_compatibility_surface(): + assert protocol.PoolStore is pool_store.PoolStore + missing = "SomethingElse" + with pytest.raises(AttributeError, match="no attribute"): + getattr(protocol, missing) + + +def test_adoption_stops_as_soon_as_the_last_ordinal_is_taken(tmp_path, monkeypatch): + items = [task(id=f"t_{index}", current_run_id=50 + index) for index in range(4)] + kanban = install_kanban(monkeypatch, items, tmp_path) + monkeypatch.setattr( + coordinator, "assignment_payload", + lambda _db, _connection, _task, _board: assignment_payload(), + ) + store = pool_store.PoolStore(tmp_path / "full.db") + coordinator.Coordinator(MASTER, store).reconcile() + assert len(store.active_assignments()) == 3 + assert store.available_ordinals() == [] + assert kanban.blocked == [] + + +def test_an_unreadable_board_registry_defers_instead_of_escaping( + tmp_path, monkeypatch, capsys +): + kanban = install_kanban(monkeypatch, [task()], tmp_path) + kanban.list_boards = lambda include_archived=False: (_ for _ in ()).throw(LOCKED) + store = pool_store.PoolStore(tmp_path / "registry.db") + coordinator.Coordinator(MASTER, store).reconcile() + assert store.active_assignments() == [] + assert "board-registry" in capsys.readouterr().err + + +def test_an_unreadable_claim_path_defers_instead_of_escaping( + tmp_path, monkeypatch, capsys +): + install_kanban(monkeypatch, [task()], tmp_path) + monkeypatch.setattr( + maintenance.cli_lane_dispatch, "claim_ready", + lambda *_args, **_kwargs: (_ for _ in ()).throw(LOCKED), + ) + store = pool_store.PoolStore(tmp_path / "claim.db") + coordinator.Coordinator(MASTER, store).dispatch() + assert store.active_assignments() == [] + assert "claim-ready" in capsys.readouterr().err + + +def test_an_owned_workspace_without_a_canonical_run_is_skipped_silently( + tmp_path, monkeypatch +): + owned = task(workspace_path="/owned", current_run_id=None) + kanban = install_kanban(monkeypatch, [owned], tmp_path) + monkeypatch.setattr( + maintenance.cli_lane_dispatch, "claim_ready", + lambda *_args, **_kwargs: [("metis", owned.id)], + ) + store = pool_store.PoolStore(tmp_path / "owned.db") + coordinator.Coordinator(MASTER, store).dispatch() + assert kanban.blocked == [] + assert store.active_assignments() == [] + + +def test_a_result_cannot_be_accepted_onto_an_already_terminal_row(tmp_path): + store = pool_store.PoolStore(tmp_path / "terminal.db") + store.add(binding(), assignment_payload()) + store.finalize(binding(), "finalized") + envelope = {**binding(), "payload_digest": "d" * 64, "payload": {}} + with pytest.raises(protocol.ProtocolError, match="cannot accept a result"): + store.accept_result(envelope) + + +def test_a_nonstring_message_kind_is_rejected_before_any_lookup(): + envelope = { + "version": protocol.PROTOCOL_VERSION, "kind": 7, "board": "metis", + "task_id": "t_deadbeef", "run_id": "23", "worker_ordinal": 0, "attempt": 1, + "delivery_id": "d1", "issued_at": 0, "expires_at": 1, + "payload_digest": "x", "payload": {}, "signature": "s", + } + with pytest.raises(protocol.ProtocolError, match="unexpected message kind"): + protocol.verify_envelope(b"k" * 32, envelope) diff --git a/testing/tests/test_hermes_execution_pool_manifests.py b/testing/tests/test_hermes_execution_pool_manifests.py new file mode 100644 index 00000000..8530e0ec --- /dev/null +++ b/testing/tests/test_hermes_execution_pool_manifests.py @@ -0,0 +1,83 @@ +"""Shared manifest and scheduling helpers for the execution-pool suites.""" + +from __future__ import annotations + +from pathlib import Path + +import yaml + + +ROOT = Path(__file__).parents[2] +HERMES = ROOT / "services/hermes" +FLUX = ROOT / "clusters/atlas/flux-system/applications" + + +def documents(path): + """Every non-empty YAML document in one manifest file.""" + return [item for item in yaml.safe_load_all(path.read_text()) if item] + + +def worker(): + """The rendered-source hermes-execution-worker StatefulSet.""" + return documents(HERMES / "execution-worker-statefulset.yaml")[0] + + +def mediators(): + """The three per-ordinal mediator Deployments, in ordinal order.""" + return [ + item + for item in documents(HERMES / "execution-mediator.yaml") + if item["kind"] == "Deployment" + ] + + +def node(name, *, ready=True, cordoned=False): + """One simulated pool-eligible worker node.""" + return { + "name": name, "ready": ready, "cordoned": cordoned, + "labels": { + "kubernetes.io/arch": "arm64", + "node-role.kubernetes.io/worker": "true", + "kubernetes.io/hostname": name, + }, + } + + +def matches(candidate, expressions): + """Whether one node satisfies a nodeAffinity match-expression list.""" + for expression in expressions: + actual = candidate["labels"].get(expression["key"]) + values = expression.get("values", []) + if expression["operator"] == "In" and actual not in values: + return False + if expression["operator"] == "NotIn" and actual in values: + return False + return True + + +def placeable(pod, nodes, *, occupied=()): + """Nodes this Pod spec could be scheduled onto, honouring hard constraints.""" + affinity = pod.get("affinity", {}) + required = affinity.get("nodeAffinity", {}).get( + "requiredDuringSchedulingIgnoredDuringExecution", {} + ) + terms = required.get("nodeSelectorTerms", [{"matchExpressions": []}]) + hard_pod = affinity.get("podAffinity", {}).get( + "requiredDuringSchedulingIgnoredDuringExecution", [] + ) + anti = affinity.get("podAntiAffinity", {}).get( + "requiredDuringSchedulingIgnoredDuringExecution", [] + ) + allowed = [] + for candidate in nodes: + if not candidate["ready"] or candidate["cordoned"]: + continue + if not any(matches(candidate, term["matchExpressions"]) for term in terms): + continue + if anti and candidate["name"] in occupied: + continue + if hard_pod: + # A hard podAffinity can only be satisfied where its target already is. + continue + allowed.append(candidate["name"]) + return allowed diff --git a/testing/tests/test_hermes_execution_pool_placement.py b/testing/tests/test_hermes_execution_pool_placement.py new file mode 100644 index 00000000..906c128e --- /dev/null +++ b/testing/tests/test_hermes_execution_pool_placement.py @@ -0,0 +1,197 @@ +"""Placement and rollout-cost contracts for the distributed worker pool. + +The independent review found that mediator-N pinned itself hard to worker-N while +sharing a ReadWriteOnce claim with it, so a drain or preemption that moved only +the worker deadlocked the ordinal on Multi-Attach until an operator deleted a +Pod. It also found that the provider-CLI install was redone on every Pod start +inside the 10m Flux health window for the whole ``hermes`` app. +""" + +from __future__ import annotations + +import json + +import yaml + +from testing.tests.test_hermes_execution_pool_manifests import ( # noqa: E402 + FLUX, + HERMES, + documents, + mediators, + node, + placeable, + worker, +) + + +SCRIPTS = HERMES / "scripts" + + +def test_the_shared_workspace_claim_cannot_deadlock_on_multi_attach(): + claims = { + item["metadata"]["name"]: item["spec"]["accessModes"] + for item in worker()["spec"]["volumeClaimTemplates"] + } + assert claims["workspace"] == ["ReadWriteMany"] + assert claims["tools"] == ["ReadWriteOnce"] + assert claims["provider-access"] == ["ReadWriteOnce"] + + # Every claim the mediator co-mounts with the worker must be multi-attachable. + for ordinal, deployment in enumerate(mediators()): + shared = { + volume["persistentVolumeClaim"]["claimName"] + for volume in deployment["spec"]["template"]["spec"]["volumes"] + if "persistentVolumeClaim" in volume + } + assert f"workspace-hermes-execution-worker-{ordinal}" in shared + for name in shared: + if name.startswith("workspace-hermes-execution-worker-"): + assert claims["workspace"] == ["ReadWriteMany"] + else: + assert name == f"hermes-execution-mediator-state-{ordinal}" + + +def test_mediator_placement_is_a_preference_and_shares_the_worker_priority(): + stateful = worker()["spec"]["template"]["spec"] + worker_terms = stateful["affinity"]["nodeAffinity"][ + "requiredDuringSchedulingIgnoredDuringExecution" + ]["nodeSelectorTerms"] + for ordinal, deployment in enumerate(mediators()): + pod = deployment["spec"]["template"]["spec"] + assert pod["priorityClassName"] == stateful["priorityClassName"] == "scavenger" + assert deployment["spec"]["strategy"]["type"] == "Recreate" + affinity = pod["affinity"] + # No hard podAffinity: the mediator must never be unschedulable purely + # because its worker is Pending or placed on a different node. + assert "requiredDuringSchedulingIgnoredDuringExecution" not in affinity["podAffinity"] + preferred = affinity["podAffinity"][ + "preferredDuringSchedulingIgnoredDuringExecution" + ] + assert preferred[0]["podAffinityTerm"]["labelSelector"]["matchLabels"] == { + "statefulset.kubernetes.io/pod-name": f"hermes-execution-worker-{ordinal}" + } + assert affinity["nodeAffinity"][ + "requiredDuringSchedulingIgnoredDuringExecution" + ]["nodeSelectorTerms"] == worker_terms + + +def test_a_preemption_that_moves_the_worker_leaves_both_pods_schedulable(): + """The exact drain/pressure sequence that used to need a manual Pod delete.""" + nodes = [node(f"titan-{index:02d}") for index in (5, 6, 7, 8)] + stateful = worker()["spec"]["template"]["spec"] + mediator = mediators()[1]["spec"]["template"]["spec"] + + # Steady state: worker-1 and mediator-1 are colocated on titan-05. + assert "titan-05" in placeable(stateful, nodes) + assert "titan-05" in placeable(mediator, nodes) + + # titan-05 comes under pressure and evicts the scavenger-priority worker, + # which reschedules elsewhere while the mediator has not moved yet. + survivors = placeable(stateful, nodes, occupied=("titan-05",)) + assert survivors and "titan-05" not in survivors + + # The mediator still serves from titan-05 -- the shared claim is RWX, so the + # rescheduled worker does not block on Multi-Attach. + assert "titan-05" in placeable(mediator, nodes) + + # And when titan-05 itself is drained the mediator follows independently, + # because nothing pins it to a Pod that may still be Pending. + drained = [ + node(item["name"], cordoned=item["name"] == "titan-05") for item in nodes + ] + rescheduled = placeable(mediator, drained) + assert rescheduled and "titan-05" not in rescheduled + + +def test_three_workers_still_spread_across_distinct_eligible_nodes(): + stateful = worker()["spec"]["template"]["spec"] + nodes = [node(f"titan-{index:02d}") for index in (5, 6, 7)] + [ + node("titan-13"), node("titan-22"), + ] + occupied: list[str] = [] + for _replica in range(3): + allowed = placeable(stateful, nodes, occupied=tuple(occupied)) + assert allowed, f"replica {len(occupied)} is unschedulable" + occupied.append(allowed[0]) + assert sorted(occupied) == ["titan-05", "titan-06", "titan-07"] + + +def test_provider_clis_install_once_onto_a_durable_verified_volume(): + stateful = worker() + pod = stateful["spec"]["template"]["spec"] + claims = {item["metadata"]["name"] for item in stateful["spec"]["volumeClaimTemplates"]} + assert "tools" in claims + ephemeral = { + volume["name"] for volume in pod["volumes"] if "emptyDir" in volume + } + assert "tools" not in ephemeral, "the CLI cache must survive a Pod restart" + + install = next( + item for item in pod["initContainers"] if item["name"] == "install-provider-clis" + ) + script = install["args"][0] + # The marker alone is not trusted: a pruned or partial cache reinstalls. + assert 'if [ ! -f "${marker}" ] || [ ! -x "${tools}/bin/codex" ]' in script + assert 'rm -f "${marker}"' in script + # And the install itself is bounded, so a stalled registry fails fast. + assert "timeout 900 npm install" in script + assert "--fetch-timeout=120000" in script + assert script.count("npm install") == 1 + + worker_mount = next( + item for item in pod["containers"][0]["volumeMounts"] + if item["name"] == "tools" + ) + assert worker_mount["readOnly"] is True + + +def test_flux_health_for_hermes_does_not_wait_on_the_best_effort_pool(): + kustomization = documents(FLUX / "hermes/kustomization.yaml")[0] + gated = { + (item["kind"], item["name"]) + for item in kustomization["spec"]["healthChecks"] + } + assert ("StatefulSet", "hermes-execution-worker") not in gated + assert not any(name.startswith("hermes-execution-") for _kind, name in gated) + # The rest of the app's health contract is untouched, so its dependents keep + # the same guarantees they had before the pool existed. + assert ("Deployment", "hermes-agent") in gated + assert ("Deployment", "hermes-switchyard") in gated + assert kustomization["spec"]["timeout"] == "10m" + + for name in ("hermes-chat", "hermes-observer-bindings"): + dependent = documents(FLUX / name / "kustomization.yaml")[0] + assert {item["name"] for item in dependent["spec"]["dependsOn"]} >= {"hermes"} + + +def test_the_pool_still_reports_its_own_readiness(): + pod = worker()["spec"]["template"]["spec"] + assert pod["containers"][0]["readinessProbe"]["exec"]["command"][-1].startswith( + "test -w /workspace" + ) + for deployment in mediators(): + container = deployment["spec"]["template"]["spec"]["containers"][0] + assert container["readinessProbe"]["httpGet"] == { + "path": "/ready", "port": "mediator" + } + assert container["startupProbe"]["failureThreshold"] == 60 + + +def test_the_rendered_pool_configmap_carries_every_mounted_module(): + kustomization = yaml.safe_load((HERMES / "kustomization.yaml").read_text()) + generator = next( + item for item in kustomization["configMapGenerator"] + if item["name"] == "hermes-execution-pool" + ) + keys = {entry.split("=", 1)[0] for entry in generator["files"]} + assert { + "execution_pool_store.py", + "execution_pool_maintenance.py", + "execution_pool_coordinator.py", + "execution_pool_server.py", + } <= keys + modules = sorted( + path.name for path in SCRIPTS.glob("execution_pool_*.py") + ) + assert set(modules) <= keys, "a pool module is missing from its own mount" + assert json.dumps(sorted(keys)) diff --git a/testing/tests/test_hermes_execution_pool_recovery.py b/testing/tests/test_hermes_execution_pool_recovery.py new file mode 100644 index 00000000..e78dd2da --- /dev/null +++ b/testing/tests/test_hermes_execution_pool_recovery.py @@ -0,0 +1,175 @@ +"""Lease-failure recovery contracts for the execution-pool coordinator. + +The independent review reproduced a wedge in which a Kanban write that failed +during lease expiry left a ``lease_failed`` row that was invisible, immortal, +and fatal. These tests pin the recovered behaviour: the row stays retryable, is +only collected on authoritative terminal evidence, and survives a restart on the +PVC the store lives on. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + + +ROOT = Path(__file__).parents[2] +SCRIPTS = ROOT / "services/hermes/scripts" +SCM_SCRIPTS = ROOT / "services/hermes/scm-common/scripts" +sys.path[:0] = [str(SCRIPTS), str(SCM_SCRIPTS)] + +import execution_pool_coordinator as coordinator # noqa: E402 +import execution_pool_maintenance as maintenance # noqa: E402 +import execution_pool_store as pool_store # noqa: E402 +from testing.tests.test_hermes_execution_pool_coordinator_v2 import ( # noqa: E402 + MASTER, + assignment_payload, + binding, + install_kanban, + task, +) +from testing.tests.test_hermes_execution_pool_support import ( # noqa: E402 + LOCKED, + age, + clear_pool_deferrals, # noqa: F401 - autouse fixture, resolved by name + exhausted_store, + states, +) + + +def test_locked_board_at_lease_expiry_stays_retryable_and_never_wedges( + tmp_path, monkeypatch, capsys +): + """The exact review repro: a locked board must not strand the run.""" + live = task() + kanban = install_kanban(monkeypatch, [live], tmp_path) + monkeypatch.setattr( + coordinator, "assignment_payload", + lambda _db, _connection, _task, _board: assignment_payload(), + ) + + def locked(*_args, **_kwargs): + raise LOCKED + + kanban.block_task = locked + store = exhausted_store(tmp_path / "wedge.db") + pool = coordinator.Coordinator(MASTER, store) + + pool.expire_leases() + assert states(store) == [("23", 0, 3, "lease_failed")] + assert store.available_ordinals() == [0, 1, 2] + assert "lease park" in capsys.readouterr().err + + # 1. The row is never collected while its outcome is unconfirmed. + age(store, 10 * 365 * 86400) + assert store.garbage_collect(0) == 0 + assert states(store) == [("23", 0, 3, "lease_failed")] + + # 2. reconcile() no longer hits the conflicting-duplicate primary key. + for _cycle in range(3): + pool.reconcile() + assert states(store) == [("23", 0, 3, "lease_failed")] + + # 3. dispatch() makes no spurious Kanban mutation. + monkeypatch.setattr( + maintenance.cli_lane_dispatch, "claim_ready", + lambda *_args, **_kwargs: [("metis", live.id)], + ) + pool.dispatch() + assert kanban.blocked == [] + assert states(store) == [("23", 0, 3, "lease_failed")] + + # 4. Once the board recovers, the retry records the exact run and settles. + kanban.block_task = lambda _connection, task_id, **values: ( + kanban.blocked.append((task_id, values)) or True + ) + pool.expire_leases() + assert kanban.blocked[-1][1] == { + "reason": "Distributed worker lease expired after 3 fenced attempts", + "kind": "transient", + "expected_run_id": 23, + } + assert states(store) == [("23", 0, 3, "finalized")] + + # 5. Only now, with authoritative evidence, is the row collectable. + age(store, 10 * 365 * 86400) + assert store.garbage_collect(0) == 1 + assert states(store) == [] + + +def test_refused_exact_run_park_keeps_the_row_for_the_next_pass(tmp_path, monkeypatch): + """No terminal state without evidence, even when the write is refused.""" + live = task() + kanban = install_kanban(monkeypatch, [live], tmp_path) + kanban.block_task = lambda _connection, task_id, **values: ( + kanban.blocked.append((task_id, values)) or False + ) + store = exhausted_store(tmp_path / "refused.db") + pool = coordinator.Coordinator(MASTER, store) + + pool.expire_leases() + assert states(store) == [("23", 0, 3, "lease_failed")] + assert len(kanban.blocked) == 1 + + # The next pass retries rather than giving up on the run. + pool.expire_leases() + assert len(kanban.blocked) == 2 + assert states(store) == [("23", 0, 3, "lease_failed")] + + +def test_lease_failure_settles_stale_when_the_run_moved_or_vanished( + tmp_path, monkeypatch +): + moved = task(current_run_id=99) + kanban = install_kanban(monkeypatch, [moved], tmp_path) + store = exhausted_store(tmp_path / "moved.db") + coordinator.Coordinator(MASTER, store).expire_leases() + assert states(store) == [("23", 0, 3, "stale")] + assert kanban.blocked == [] + + kanban.tasks.clear() + gone = exhausted_store(tmp_path / "gone.db", binding(task_id="t_gone", attempt=3)) + coordinator.Coordinator(MASTER, gone).expire_leases() + assert states(gone) == [("23", 0, 3, "stale")] + + +def test_lease_failure_settles_finalized_when_kanban_already_parked_the_run( + tmp_path, monkeypatch +): + parked = task(status="blocked") + kanban = install_kanban(monkeypatch, [parked], tmp_path) + store = exhausted_store(tmp_path / "parked.db") + coordinator.Coordinator(MASTER, store).expire_leases() + assert states(store) == [("23", 0, 3, "finalized")] + assert kanban.blocked == [] + + +def test_noncanonical_run_lease_failure_is_released_without_a_board_call( + tmp_path, monkeypatch +): + kanban = install_kanban(monkeypatch, [task()], tmp_path) + store = exhausted_store( + tmp_path / "bad-run.db", binding(run_id="run-bad", attempt=3) + ) + coordinator.Coordinator(MASTER, store).expire_leases() + assert states(store) == [("run-bad", 0, 3, "stale")] + assert kanban.blocked == [] + + +def test_a_durable_row_in_any_state_blocks_conflicting_re_adoption(tmp_path, monkeypatch): + """known_runs() covers terminal and retrying rows, not just live ones.""" + live = task() + install_kanban(monkeypatch, [live], tmp_path) + monkeypatch.setattr( + coordinator, "assignment_payload", + lambda _db, _connection, _task, _board: assignment_payload(context="different"), + ) + store = pool_store.PoolStore(tmp_path / "known.db") + store.add(binding(), assignment_payload()) + pool = coordinator.Coordinator(MASTER, store) + for state in ("lease_failed", "finalized", "stale"): + with store._connect() as connection: + connection.execute("UPDATE assignments SET state=?", (state,)) + assert store.known_runs() == {("metis", "t_deadbeef", "23")} + pool.reconcile() + assert states(store) == [("23", 0, 1, state)] diff --git a/testing/tests/test_hermes_execution_pool_submission.py b/testing/tests/test_hermes_execution_pool_submission.py new file mode 100644 index 00000000..2ea26692 --- /dev/null +++ b/testing/tests/test_hermes_execution_pool_submission.py @@ -0,0 +1,253 @@ +"""Retry-submission contracts against the creation-only Atlas SCM broker. + +The broker deliberately permits only new namespaced branch creation, so a pool +retry that adds commits to an already-published branch could never submit: the +push was refused, the worker unwound, and the run's actual work was discarded +along with it. Submission now targets a fresh attempt- or content-scoped ref, and +a refused submission downgrades the result instead of losing it. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +import pytest + + +ROOT = Path(__file__).parents[2] +SCRIPTS = ROOT / "services/hermes/scripts" +SCM_SCRIPTS = ROOT / "services/hermes/scm-common/scripts" +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 scm_broker # noqa: E402 +from testing.tests.test_hermes_execution_pool_mediator import ( # noqa: E402 + KEY, + RESULT, + assignment, + binding, + payload, +) + + +def attempt_assignment(attempt): + """One signed assignment envelope bound to an exact retry attempt.""" + return protocol.sign_envelope( + KEY, "assignment", binding(attempt=attempt), payload() + ) + + +def test_submission_refs_are_creation_only_and_attempt_scoped(): + first = scm.submission_refs("feature/pool", 1, "a" * 40) + assert first[0] == "feature/pool" + assert first[1] == "feature/pool-attempt-1" + assert first[2] == f"feature/pool-{'a' * 12}" + retry = scm.submission_refs("feature/pool", 3, "b" * 40) + assert retry[1] == "feature/pool-attempt-3" + # 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 candidate not in {"main", "master"} + + +def test_submission_refs_drop_oversized_names_and_fail_closed_when_empty(): + long_branch = "feature/" + "a" * 185 + candidates = scm.submission_refs(long_branch, 2, "c" * 40) + assert candidates == (long_branch,) + with pytest.raises(protocol.ProtocolError, match="reviewed branch name"): + scm.submission_refs("main", 1, "d" * 40) + + +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" + ) + + +def submit_harness(tmp_path, monkeypatch, remote_refs, *, attempt=1): + """A Boundary whose broker advertises exactly ``remote_refs``.""" + monkeypatch.setattr(scm, "WORKSPACE_ROOT", tmp_path / "workspace") + monkeypatch.setattr(scm, "SCM_ROOT", tmp_path / "state") + monkeypatch.setattr(scm, "ORDINAL", 0) + exact = attempt_assignment(attempt) + destination = scm.workspace_path(exact) + destination.mkdir(parents=True) + protocol.atomic_json(scm._state_path(exact), {"baseline_sha": "a" * 40}) + head = "b" * 40 + monkeypatch.setattr(scm, "_workspace_identity", lambda *_a: head) + calls: list[tuple[str, ...]] = [] + + def run(*arguments, **_kwargs): + calls.append(arguments) + if arguments[0] == "ls-remote": + return "\n".join( + f"{sha}\trefs/heads/{ref}" for ref, sha in remote_refs.items() + ) + if arguments[0] == "rev-list": + return "2" + return "" + + monkeypatch.setattr(scm, "_run", run) + boundary = scm.Boundary(KEY) + monkeypatch.setattr(boundary, "_draft", lambda *args: f"https://scm/pulls/{args[1]}") + return boundary, exact, calls, head + + +def test_a_retry_publishes_an_attempt_ref_instead_of_updating_the_existing_one( + tmp_path, monkeypatch +): + boundary, exact, calls, _head = submit_harness( + tmp_path, monkeypatch, {"wt/t_deadbeef": "c" * 40}, attempt=2 + ) + result = boundary.submit(exact, {"title": "retry", "body": "evidence"}) + pushes = [item for item in calls if item[0] == "push"] + assert pushes == [("push", "hermes-broker", "HEAD:refs/heads/wt/t_deadbeef-attempt-2")] + assert result["branch"] == "wt/t_deadbeef-attempt-2" + assert result["pull_request"] == "https://scm/pulls/wt/t_deadbeef-attempt-2" + + +def test_an_already_published_head_is_adopted_without_a_second_push( + tmp_path, monkeypatch +): + boundary, exact, calls, head = submit_harness( + tmp_path, monkeypatch, {"wt/t_deadbeef-attempt-2": "b" * 40}, attempt=2 + ) + assert head == "b" * 40 + result = boundary.submit(exact, {"title": "replay", "body": "evidence"}) + assert [item for item in calls if item[0] == "push"] == [] + assert result["branch"] == "wt/t_deadbeef-attempt-2" + assert result["pull_request"].endswith("wt/t_deadbeef-attempt-2") + + +def test_a_run_with_no_new_commits_still_reports_work_a_prior_attempt_published( + tmp_path, monkeypatch +): + boundary, exact, calls, _head = submit_harness( + tmp_path, monkeypatch, {"wt/t_deadbeef": "b" * 40} + ) + + real_run = scm._run + + def run(*arguments, **kwargs): + if arguments[0] == "rev-list": + return "0" + return real_run(*arguments, **kwargs) + + monkeypatch.setattr(scm, "_run", run) + result = boundary.submit(exact, {"title": "nothing new", "body": "evidence"}) + assert [item for item in calls if item[0] == "push"] == [] + assert result["pull_request"].endswith("wt/t_deadbeef") + + +def test_a_run_that_produced_nothing_at_all_submits_nothing(tmp_path, monkeypatch): + boundary, exact, calls, _head = submit_harness(tmp_path, monkeypatch, {}) + + real_run = scm._run + + def run(*arguments, **kwargs): + if arguments[0] == "rev-list": + return "0" + return real_run(*arguments, **kwargs) + + monkeypatch.setattr(scm, "_run", run) + result = boundary.submit(exact, {}) + assert [item for item in calls if item[0] == "push"] == [] + assert result["pull_request"] == "" and result["branch"] == "wt/t_deadbeef" + + +def test_submission_fails_closed_when_every_candidate_ref_is_taken( + tmp_path, monkeypatch +): + taken = { + "wt/t_deadbeef": "c" * 40, + "wt/t_deadbeef-attempt-1": "d" * 40, + f"wt/t_deadbeef-{'b' * 12}": "e" * 40, + } + boundary, exact, calls, _head = submit_harness(tmp_path, monkeypatch, taken) + with pytest.raises(protocol.ProtocolError, match="every reviewed branch name"): + boundary.submit(exact, {"title": "blocked", "body": "evidence"}) + assert [item for item in calls if item[0] == "push"] == [] + + +def test_remote_head_parsing_ignores_anything_that_is_not_a_branch( + tmp_path, monkeypatch +): + monkeypatch.setattr( + scm, "_run", + lambda *_a, **_k: "aaa\trefs/tags/v1\nbbb\trefs/heads/feature/x\ngarbage\n", + ) + assert scm._remote_heads(tmp_path, ("feature/x",)) == {"feature/x": "bbb"} + + +class FailingSCM: + def __init__(self, error): + self.error = error + self.calls = 0 + + def submit(self, _assignment, _request): + self.calls += 1 + raise self.error + + +def test_a_refused_submission_downgrades_the_result_instead_of_discarding_it(): + failing = FailingSCM(protocol.ProtocolError("every reviewed branch name is taken")) + boundary = client.ClientBoundary(KEY, failing) + boundary.current = assignment() + boundary._post = lambda _path, _envelope: protocol.sign_envelope( + KEY, "ack", binding(), {"accepted": True, "duplicate": False} + ) + finished = boundary.finish( + { + "binding": binding(), + "payload": {"structured": dict(RESULT), "returncode": 0}, + "title": "safe", "body": "evidence", + } + ) + assert failing.calls == 1 + structured = finished["structured"] + # The run is recorded rather than unwound, and it says exactly why. + assert structured["status"] == "blocked" + assert structured["changed_files"] == ["safe.py"] + assert len(structured["blockers"]) == 1 + assert "commits remain on this ordinal's workspace" in structured["blockers"][0] + assert "every reviewed branch name is taken" in structured["blockers"][0] + assert finished["ack"]["accepted"] is True + + +def test_a_successful_submission_records_both_the_draft_and_the_exact_branch(): + class Recording: + def submit(self, _assignment, _request): + return { + "workspace": "/workspace/runs/metis/t_deadbeef/42", + "branch": "wt/t_deadbeef-attempt-2", + "pull_request": "https://scm/pulls/9", + } + + boundary = client.ClientBoundary(KEY, Recording()) + boundary.current = assignment() + boundary._post = lambda _path, _envelope: protocol.sign_envelope( + KEY, "ack", binding(), {"accepted": True, "duplicate": False} + ) + finished = boundary.finish( + { + "binding": binding(), + "payload": {"structured": dict(RESULT), "returncode": 0}, + "title": "safe", "body": "evidence", + } + ) + assert finished["structured"]["status"] == "completed" + assert finished["structured"]["artifacts"] == [ + "https://scm/pulls/9", "branch:wt/t_deadbeef-attempt-2", + ] diff --git a/testing/tests/test_hermes_execution_pool_support.py b/testing/tests/test_hermes_execution_pool_support.py new file mode 100644 index 00000000..d98fd373 --- /dev/null +++ b/testing/tests/test_hermes_execution_pool_support.py @@ -0,0 +1,60 @@ +"""Shared fixtures for the execution-pool recovery and isolation suites.""" + +from __future__ import annotations + +import sqlite3 +import sys +import time +from pathlib import Path + +import pytest + + +ROOT = Path(__file__).parents[2] +SCRIPTS = ROOT / "services/hermes/scripts" +SCM_SCRIPTS = ROOT / "services/hermes/scm-common/scripts" +sys.path[:0] = [str(SCRIPTS), str(SCM_SCRIPTS)] + +import execution_pool_maintenance as maintenance # noqa: E402 +import execution_pool_store as pool_store # noqa: E402 +from testing.tests.test_hermes_execution_pool_coordinator_v2 import ( # noqa: E402 + assignment_payload, + binding, +) + + +LOCKED = sqlite3.OperationalError("database is locked") + + +@pytest.fixture(autouse=True) +def clear_pool_deferrals(): + """Keep the bounded deferral report from leaking between tests.""" + maintenance.DEFERRALS.clear() + yield + maintenance.DEFERRALS.clear() + + +def exhausted_store(path, exact=None): + """A store whose only assignment has burned every fenced attempt.""" + store = pool_store.PoolStore(path) + store.add(exact or binding(attempt=3), assignment_payload()) + store.offer(int((exact or binding(attempt=3))["worker_ordinal"])) + with store._connect() as connection: + connection.execute("UPDATE assignments SET lease_until=1") + return store + + +def states(store): + """Every durable row as (run_id, ordinal, attempt, state).""" + return [ + tuple(row) + for row in store._connect().execute( + "SELECT run_id,worker_ordinal,attempt,state FROM assignments ORDER BY run_id" + ) + ] + + +def age(store, seconds): + """Backdate every row so retention-based collection can be exercised.""" + with store._connect() as connection: + connection.execute("UPDATE assignments SET updated_at=?", (time.time() - seconds,))