atlas-iac/dockerfiles/hermes-kanban-blocked-regression.py

"""Regression tests for Hermes Kanban blocked-task scheduling semantics."""

from __future__ import annotations

import os
import tempfile
import unittest
from pathlib import Path


_HERMES_HOME = tempfile.TemporaryDirectory(prefix="hermes-kanban-test-home-")
os.environ["HERMES_HOME"] = _HERMES_HOME.name

from hermes_cli import kanban_db  # noqa: E402


class BlockedTaskSchedulingTests(unittest.TestCase):
    """Exercise the real upstream database API against isolated databases."""

    def setUp(self) -> None:
        self._database_dir = tempfile.TemporaryDirectory(
            prefix="hermes-kanban-test-db-"
        )
        self.connection = kanban_db.connect(
            Path(self._database_dir.name) / "kanban.db"
        )

    def tearDown(self) -> None:
        self.connection.close()
        self._database_dir.cleanup()

    def create_task(self, title: str, **kwargs: object) -> str:
        return kanban_db.create_task(self.connection, title=title, **kwargs)

    def status(self, task_id: str) -> str:
        task = kanban_db.get_task(self.connection, task_id)
        self.assertIsNotNone(task)
        return task.status

    def test_initial_block_without_parents_is_sticky(self) -> None:
        task_id = self.create_task("operator parked", initial_status="blocked")

        self.assertEqual(kanban_db.recompute_ready(self.connection), 0)
        self.assertEqual(self.status(task_id), "blocked")

    def test_initial_block_with_complete_parents_is_sticky(self) -> None:
        parent_id = self.create_task("completed prerequisite")
        self.assertTrue(kanban_db.complete_task(self.connection, parent_id))
        task_id = self.create_task(
            "operator parked after prerequisite",
            initial_status="blocked",
            parents=[parent_id],
        )

        self.assertEqual(kanban_db.recompute_ready(self.connection), 0)
        self.assertEqual(self.status(task_id), "blocked")

    def test_initial_block_is_released_by_an_explicit_unblock(self) -> None:
        task_id = self.create_task("operator parked", initial_status="blocked")

        self.assertTrue(kanban_db.unblock_task(self.connection, task_id))
        self.assertEqual(self.status(task_id), "ready")

    def test_dependency_block_promotes_after_parent_completes(self) -> None:
        parent_id = self.create_task("incomplete prerequisite")
        task_id = self.create_task("dependency gated", parents=[parent_id])
        self.assertEqual(self.status(task_id), "todo")

        self.assertTrue(kanban_db.complete_task(self.connection, parent_id))
        self.assertEqual(self.status(task_id), "ready")

    def test_explicit_block_and_unblock_remain_sticky_and_reversible(self) -> None:
        task_id = self.create_task("worker handoff")

        self.assertTrue(
            kanban_db.block_task(
                self.connection,
                task_id,
                reason="human review required",
            )
        )
        self.assertEqual(kanban_db.recompute_ready(self.connection), 0)
        self.assertEqual(self.status(task_id), "blocked")
        self.assertTrue(kanban_db.unblock_task(self.connection, task_id))
        self.assertEqual(self.status(task_id), "ready")

    def test_circuit_breaker_block_is_not_auto_promoted(self) -> None:
        task_id = self.create_task("repeated worker failure")
        self.assertIsNotNone(kanban_db.claim_task(self.connection, task_id))

        # Upstream has no public circuit-breaker entry point; exercise the
        # dispatcher helper that owns its persisted failure-limit semantics.
        self.assertTrue(
            kanban_db._record_spawn_failure(
                self.connection,
                task_id,
                "worker failed",
                failure_limit=1,
            )
        )
        self.assertEqual(self.status(task_id), "blocked")
        self.assertEqual(
            kanban_db.recompute_ready(self.connection, failure_limit=1),
            0,
        )
        self.assertEqual(self.status(task_id), "blocked")


if __name__ == "__main__":
    unittest.main(verbosity=2)