atlas-iac/scripts/ops/hermes_handoff_checks_workers.py

#!/usr/bin/env python3
"""Acceptance checks for the distributed worker pool and the visible surfaces.

A pool is only distributed if its workers occupy distinct nodes and distinct
volumes and leave state ownership with the coordinator; a replica count alone
satisfies none of that. A surface is only working if the work it did is visible
in it, which is why the session and Kanban checks look at the same data the UI
renders rather than at whether a pod is Running.

The ephemeral entries are placeholders in a default run. They report NOT_RUN
until the mutation mode is separately armed, and the armed mode replaces them
with real, mandatory results.
"""

from __future__ import annotations

from hermes_handoff_catalog import (
    CHAT,
    INIT_NAME_PROJECTION,
    OPERATOR,
    POOL_PROJECTION,
    SELF,
    Targets,
    check,
)
from hermes_handoff_catalog import step as step
from hermes_handoff_model import EPHEMERAL, CheckSpec
from hermes_handoff_policy import projection

KANBAN_STATUSES = [
    "archived",
    "blocked",
    "done",
    "ready",
    "review",
    "running",
    "scheduled",
    "todo",
    "triage",
]


def _pool(targets: Targets) -> list[CheckSpec]:
    selector = targets.pool_selector
    return [
        check(
            "pool.three-workers-have-distinct-nodes-and-volumes",
            "Exactly three Ready ordinal workers occupy distinct nodes and claims",
            "pool",
            "pool_topology",
            [
                step(
                    "statefulset",
                    OPERATOR,
                    "kubectl",
                    "--namespace",
                    targets.namespace,
                    "get",
                    f"statefulset/{targets.pool_statefulset}",
                    "-o",
                    projection(
                        "jsonpath={.spec.replicas}{'\\t'}{.status.readyReplicas}{'\\t'}{.status.currentReplicas}{'\\t'}{.spec.template.spec.automountServiceAccountToken}".replace(
                            "'", '"'
                        )
                    ),
                ),
                step(
                    "workers",
                    OPERATOR,
                    "kubectl",
                    "--namespace",
                    targets.namespace,
                    "get",
                    "pods",
                    "--selector",
                    selector,
                    "-o",
                    POOL_PROJECTION,
                ),
            ],
            {
                "state_step": "statefulset",
                "worker_step": "workers",
                "replicas": targets.pool_replicas,
                "name": targets.pool_statefulset,
            },
            rationale="Replica count, Ready state, ordinal coverage, node placement, PVC isolation, and tokenlessness are one fail-closed assertion.",
        ),
        check(
            "pool.coordinator-retains-sole-state-ownership",
            "No pool worker mounts the coordinator's durable state",
            "pool",
            "names_absent",
            [
                step(
                    "claims",
                    OPERATOR,
                    "kubectl",
                    "--namespace",
                    targets.namespace,
                    "get",
                    f"statefulset/{targets.pool_statefulset}",
                    "-o",
                    projection(
                        "jsonpath={range .spec.template.spec.volumes[*]}{.name}{'='}"
                        "{.persistentVolumeClaim.claimName}{'\\n'}{end}".replace(
                            "'", '"'
                        )
                    ),
                )
            ],
            {"step": "claims", "contains": targets.coordinator_claims},
            rationale="Two writers on one SQLite state file is the duplicate-work and lost-result failure this pool exists to avoid. One line per template volume keeps a drifted or empty projection NOT_RUN instead of a silent pass.",
        ),
        *_pool_env(targets),
    ]


def _pool_env(targets: Targets) -> list[CheckSpec]:
    """Return the mandatory assignment-safety configuration check."""
    return [
        check(
            "pool.assignment-safety-knobs-are-configured",
            "Pool workers carry the lease and de-duplication configuration",
            "pool",
            "names_present",
            [
                step(
                    "env",
                    OPERATOR,
                    "kubectl",
                    "--namespace",
                    targets.namespace,
                    "get",
                    f"statefulset/{targets.pool_statefulset}",
                    "-o",
                    projection(
                        'jsonpath={range .spec.template.spec.containers[*]}{range .env[*]}{.name}{"\\n"}{end}{end}'
                    ),
                )
            ],
            {"step": "env", "names": targets.pool_worker_env},
            rationale="Deployed configuration only; the stale-claim and duplicate-dispatch behaviour itself is covered by the pool's own regression suite.",
        ),
        check(
            "pool.assignment-fieldrefs-are-bound",
            "Worker node and ordinal identities come from fixed downward-API fields",
            "pool",
            "names_present",
            [
                step(
                    "source",
                    OPERATOR,
                    "git",
                    "show",
                    f"{targets.remote}/main:services/hermes/execution-worker-statefulset.yaml",
                    record=False,
                )
            ],
            {
                "step": "source",
                "names": (
                    "name: HERMES_WORKER_ORDINAL",
                    "fieldPath: metadata.labels['apps.kubernetes.io/pod-index']",
                    "name: HERMES_WORKER_NODE",
                    "fieldPath: spec.nodeName",
                ),
            },
            rationale="This prevents caller-authored ordinal/node values from impersonating another worker assignment.",
        ),
    ]


def _surfaces(targets: Targets) -> list[CheckSpec]:
    checks = [
        check(
            "surfaces.kanban-activity-is-visible",
            "Durable Kanban activity is visible to the agent surface",
            "surfaces",
            "all_of_field",
            [
                step(
                    "tasks",
                    SELF,
                    "hermes",
                    "kanban",
                    "list",
                    "--json",
                    record=False,
                    max_bytes=targets.listing_bytes,
                )
            ],
            {"step": "tasks", "key_field": "id", "fields": {"status": KANBAN_STATUSES}},
            rationale="Non-empty and well-formed. An empty board is NOT_RUN, because it cannot distinguish a quiet week from a broken read.",
        ),
        check(
            "surfaces.durable-sessions-render-without-blank-rows",
            "Durable sessions are listed with a preview, a source, and an id",
            "surfaces",
            "lines_match",
            [
                step(
                    "sessions",
                    SELF,
                    "hermes",
                    "sessions",
                    "list",
                    "--limit",
                    "20",
                    record=False,
                )
            ],
            {"step": "sessions", "pattern": r"^\S.*\S$", "skip": 2, "minimum": 1},
            rationale="A row whose leading column is blank is the session-render regression seen from the data side.",
        ),
        check(
            "surfaces.agent-ui-session-patches-are-deployed",
            "The agent surface runs the session-activity rendering patches",
            "surfaces",
            "names_present",
            [
                step(
                    "init",
                    OPERATOR,
                    "kubectl",
                    "--namespace",
                    targets.namespace,
                    "get",
                    f"deploy/{targets.agent_deployment}",
                    "-o",
                    INIT_NAME_PROJECTION,
                )
            ],
            {
                "step": "init",
                "names": ("patch-web-session-activity", "patch-api-server-sessions"),
            },
            rationale="Blank session rows were a rendering regression; this asserts the patches that fixed it are in the running pod.",
        ),
        check(
            "surfaces.chat-session-patches-are-deployed",
            "The chat surface runs the API-session continuity patches",
            "surfaces",
            "names_present",
            [
                step(
                    "init",
                    OPERATOR,
                    "kubectl",
                    "--namespace",
                    targets.namespace,
                    "get",
                    f"statefulset/{targets.chat_statefulset}",
                    "-o",
                    INIT_NAME_PROJECTION,
                )
            ],
            {"step": "init", "names": targets.chat_init_containers},
        ),
        check(
            "surfaces.chat-telegram-topic-state-is-durable",
            "Telegram topic and tenant state is held on durable storage",
            "surfaces",
            "names_present",
            [
                step(
                    "claims",
                    OPERATOR,
                    "kubectl",
                    "--namespace",
                    targets.namespace,
                    "get",
                    f"deploy/{targets.chat_router_deployment}",
                    "-o",
                    projection(
                        'jsonpath={range .spec.template.spec.volumes[*]}{.persistentVolumeClaim.claimName}{"\\n"}{end}'
                    ),
                )
            ],
            {"step": "claims", "contains": ("state",)},
            rationale="Topic selection survives a router restart only if the tenant state file is on a claim rather than in the pod.",
        ),
    ]
    checks.append(
        check(
            "surfaces.chat-telegram-sessions-are-continuous",
            "The selected chat tenant still holds durable Telegram-sourced sessions",
            "surfaces",
            "lines_match",
            [
                step(
                    "telegram",
                    CHAT,
                    "hermes",
                    "sessions",
                    "list",
                    "--source",
                    "telegram",
                    "--limit",
                    "20",
                    record=False,
                )
            ],
            {"step": "telegram", "pattern": r"^\S.*\S$", "skip": 2, "minimum": 1},
            rationale="Session continuity is per tenant and depends on there having been Telegram traffic, so it is asserted only when the operator names the tenant that carries the identity.",
        )
    )
    checks.append(
        check(
            "surfaces.chat-runs-the-telegram-topic-revision",
            "The chat surface runs the durable Telegram topic configuration",
            "surfaces",
            "stdout_matches",
            [
                step(
                    "revision",
                    OPERATOR,
                    "kubectl",
                    "--namespace",
                    targets.namespace,
                    "get",
                    f"statefulset/{targets.chat_statefulset}",
                    "-o",
                    projection(
                        "jsonpath={.spec.template.metadata.annotations.ai\\.bstein\\.dev/config-rev}"
                    ),
                )
            ],
            {"step": "revision", "equals": targets.chat_config_revision},
            rationale="Telegram topic continuity is carried by the config revision the pod template pins.",
        )
    )
    return checks


def _ephemeral() -> list[CheckSpec]:
    return [
        check(
            f"ephemeral.{name}",
            title,
            "ephemeral",
            "not_armed",
            [],
            {"reason": reason},
            mandatory=False,
            scope=EPHEMERAL,
            rationale="Mutating checks are separately armed. A default run reports them NOT_RUN and never touches the forge.",
        )
        for name, title, reason in (
            (
                "feature-branch-push",
                "A unique ephemeral feature branch can be pushed and removed",
                "ephemeral mutation mode is not armed",
            ),
            (
                "draft-pull-request",
                "A draft pull request can be opened and closed against the ephemeral branch",
                "ephemeral mutation mode is not armed",
            ),
            (
                "protected-branch-refusal",
                "Arming against a protected branch is refused before any network call",
                "ephemeral mutation mode is not armed",
            ),
            (
                "cleanup-verified",
                "Every ephemeral ref and pull request is removed and the removal is verified",
                "ephemeral mutation mode is not armed",
            ),
        )
    ]


def worker_checks(targets: Targets) -> list[CheckSpec]:
    """Return the pool, surface, and ephemeral-placeholder checks."""
    return [*_pool(targets), *_surfaces(targets), *_ephemeral()]