diff --git a/hermes_cli/kanban_db.py b/hermes_cli/kanban_db.py index 0db694ff5b1b8..4edbe5231f9aa 100644 --- a/hermes_cli/kanban_db.py +++ b/hermes_cli/kanban_db.py @@ -796,7 +796,15 @@ class Event: -- ``max_retries=1`` blocks on the first failure. NULL (the common -- case) falls through to the dispatcher-level ``kanban.failure_limit`` -- config and then ``DEFAULT_FAILURE_LIMIT``. - max_retries INTEGER + max_retries INTEGER, + -- Triage labels for routing. JSON-encoded array of strings (tags + -- like "bug", "infra", "needs-spec"). Populated by the Kanban + -- specifier (LLM) during triage and consumed by the routing engine + -- to pick an assignee/template. Stored separately from + -- ``metadata`` on ``task_runs`` so tags stay structurally distinct + -- from arbitrary per-run data. Defaults to ``'[]'`` so existing + -- reads never see NULL. + labels TEXT NOT NULL DEFAULT '[]' ); CREATE TABLE IF NOT EXISTS task_links ( @@ -1076,6 +1084,11 @@ def _migrate_add_optional_columns(conn: sqlite3.Connection) -> None: # they were getting before the column existed). _add_column_if_missing(conn, "tasks", "max_retries", "max_retries INTEGER") + # Triage labels (JSON array of strings) for the Kanban routing + # layer. Adds NOT NULL DEFAULT '[]' so legacy rows immediately + # satisfy ``json.loads(labels)`` invariants. + _migrate_add_labels_column(conn) + # task_events gained a run_id column; back-fill it as NULL for # historical events (they predate runs and can't be attributed). ev_cols = {row["name"] for row in conn.execute("PRAGMA table_info(task_events)")} @@ -1169,6 +1182,27 @@ def _migrate_add_optional_columns(conn: sqlite3.Connection) -> None: ) +def _migrate_add_labels_column(conn: sqlite3.Connection) -> bool: + """Add ``tasks.labels`` (JSON array of strings) to legacy DBs. + + Phase-1 of the Kanban triage layer: the column is written by the + LLM specifier and read by the routing engine. Kept as its own + migration so the schema change is auditable in isolation and can + be added/removed without touching ``_migrate_add_optional_columns``. + + Idempotent: a ``PRAGMA table_info`` check skips the ALTER on DBs + that already have the column (fresh installs via ``SCHEMA_SQL`` + and old DBs we've already migrated). Returns ``True`` when the + column was actually added by this call. + """ + cols = {row["name"] for row in conn.execute("PRAGMA table_info(tasks)")} + if "labels" in cols: + return False + return _add_column_if_missing( + conn, "tasks", "labels", "labels TEXT NOT NULL DEFAULT '[]'" + ) + + @contextlib.contextmanager def write_txn(conn: sqlite3.Connection): """Context manager for an IMMEDIATE write transaction. @@ -1442,6 +1476,74 @@ def get_task(conn: sqlite3.Connection, task_id: str) -> Optional[Task]: return Task.from_row(row) if row else None +def get_task_labels(conn: sqlite3.Connection, task_id: str) -> list[str]: + """Return the triage labels for ``task_id`` as a list of strings. + + Returns an empty list when the task has no labels (the column + default) or when the task does not exist. Malformed rows (column + holds non-array JSON, written by some external tool) also return + ``[]`` rather than raising; the routing engine should treat + "unlabeled" as the safe default. + """ + row = conn.execute( + "SELECT labels FROM tasks WHERE id = ?", (task_id,) + ).fetchone() + if row is None: + return [] + raw = row["labels"] + if not raw: + return [] + try: + parsed = json.loads(raw) + except Exception: + return [] + if not isinstance(parsed, list): + return [] + return [str(item) for item in parsed if isinstance(item, str)] + + +def set_task_labels( + conn: sqlite3.Connection, task_id: str, labels: list[str] +) -> bool: + """Replace the triage labels for ``task_id`` with ``labels``. + + ``labels`` must be a list of strings; tuples and other iterables + are rejected so a stray ``"foo"`` (a string is iterable!) doesn't + silently become ``["f", "o", "o"]``. Each entry is stripped and + empty entries are dropped; duplicates are de-duped (order + preserved) so the LLM specifier can re-emit a tag set without + growing the column. + + Returns ``True`` when a row was updated, ``False`` when no task + with ``task_id`` exists. Raises :class:`TypeError` for malformed + input. + """ + if not isinstance(labels, list): + raise TypeError( + f"labels must be a list of strings, got {type(labels).__name__}" + ) + cleaned: list[str] = [] + seen: set[str] = set() + for item in labels: + if not isinstance(item, str): + raise TypeError( + "labels must contain only strings, got " + f"{type(item).__name__}: {item!r}" + ) + stripped = item.strip() + if not stripped or stripped in seen: + continue + seen.add(stripped) + cleaned.append(stripped) + encoded = json.dumps(cleaned, ensure_ascii=False) + with write_txn(conn): + cur = conn.execute( + "UPDATE tasks SET labels = ? WHERE id = ?", + (encoded, task_id), + ) + return cur.rowcount > 0 + + def list_tasks( conn: sqlite3.Connection, *, @@ -2697,6 +2799,8 @@ def specify_triage_task( *, title: Optional[str] = None, body: Optional[str] = None, + assignee_suggestion: Optional[str] = None, + labels: Optional[list[str]] = None, author: Optional[str] = None, ) -> bool: """Flesh out a triage task and promote it to ``todo``. @@ -2711,15 +2815,54 @@ def specify_triage_task( dispatcher tick, which keeps the normal parent-gating behaviour intact for specified tasks that happen to have open parents. + ``assignee_suggestion`` is a *suggestion only*. It is written into the + ``assignee`` column ONLY when the task currently has no assignee — we + never overwrite an existing assignee, since a human (or a routing + engine) may already have made a deliberate choice. A downstream + routing engine may still override this on a later pass. + + ``labels`` are short routing/cost tags. If the ``tasks.labels`` column + exists (added by a parallel migration), labels are stored there as a + JSON array. If it doesn't exist, we fall back to a ``tasks.metadata`` + column (also JSON, with labels nested under the ``labels`` key) when + that column is present. If neither column exists, labels are recorded + in the ``specified`` event payload so they're recoverable from the + audit log. + ``author`` is recorded on an audit comment only when at least one of - ``title`` / ``body`` actually changed — avoids noisy comment spam for + the writable fields actually changed — avoids noisy comment spam for status-only promotions. """ if title is not None and not title.strip(): raise ValueError("title cannot be blank") + # Decide up-front whether we have a ``labels`` or ``metadata`` column to + # write into. PRAGMA is cheap (single round-trip, no locking) so doing + # it here keeps the SQL builder below tidy. Cached schema lookups would + # be a micro-optimisation we don't need yet — specify is rare. + task_cols = { + row["name"] for row in conn.execute("PRAGMA table_info(tasks)") + } + has_labels_col = "labels" in task_cols + has_metadata_col = "metadata" in task_cols + norm_labels: Optional[list[str]] = None + if labels is not None: + # Defensive normalisation at the DB boundary too: callers should + # have already cleaned the list, but a bogus entry here would + # corrupt the stored JSON. + norm_labels = [ + str(label).strip() + for label in labels + if isinstance(label, str) and str(label).strip() + ] + norm_assignee_suggestion: Optional[str] = None + if assignee_suggestion is not None: + cleaned = assignee_suggestion.strip() + if cleaned: + norm_assignee_suggestion = cleaned with write_txn(conn): existing = conn.execute( - "SELECT title, body FROM tasks WHERE id = ? AND status = 'triage'", + "SELECT title, body, assignee FROM tasks " + "WHERE id = ? AND status = 'triage'", (task_id,), ).fetchone() if existing is None: @@ -2735,6 +2878,51 @@ def specify_triage_task( sets.append("body = ?") params.append(body) changed_fields.append("body") + # Only fill assignee when it's currently empty — never overwrite + # an existing one. A pre-existing assignee may have come from the + # user (CLI ``--assignee``) or a routing engine; both take + # precedence over an LLM suggestion. + if ( + norm_assignee_suggestion is not None + and not (existing["assignee"] or "") + ): + sets.append("assignee = ?") + params.append(_canonical_assignee(norm_assignee_suggestion)) + changed_fields.append("assignee") + # Labels: prefer a dedicated ``labels`` column when present, then + # ``metadata`` (storing ``{"labels": [...]}``); if neither exists + # the payload of the ``specified`` event below carries the labels. + labels_payload_for_event: Optional[list[str]] = None + if norm_labels is not None: + if has_labels_col: + sets.append("labels = ?") + params.append(json.dumps(norm_labels, ensure_ascii=False)) + changed_fields.append("labels") + elif has_metadata_col: + # Merge into any existing metadata so we don't clobber + # unrelated fields the routing engine may also write. + existing_meta_row = conn.execute( + "SELECT metadata FROM tasks WHERE id = ?", + (task_id,), + ).fetchone() + meta: dict[str, Any] = {} + if existing_meta_row and existing_meta_row["metadata"]: + try: + loaded = json.loads(existing_meta_row["metadata"]) + if isinstance(loaded, dict): + meta = loaded + except (ValueError, TypeError): + meta = {} + meta["labels"] = norm_labels + sets.append("metadata = ?") + params.append(json.dumps(meta, ensure_ascii=False)) + changed_fields.append("labels") + else: + # No persistent home for labels yet — embed them in the + # 'specified' event payload below so they're recoverable. + labels_payload_for_event = norm_labels + if norm_labels: + changed_fields.append("labels") params.append(task_id) cur = conn.execute( f"UPDATE tasks SET {', '.join(sets)} " @@ -2761,11 +2949,22 @@ def specify_triage_task( int(time.time()), ), ) + event_payload: Optional[dict[str, Any]] = None + if changed_fields or labels_payload_for_event is not None: + event_payload = {} + if changed_fields: + event_payload["changed_fields"] = changed_fields + if labels_payload_for_event is not None: + # Fallback storage: keeps labels recoverable from the + # audit log when no schema column is available. Same key + # shape as the metadata fallback (``labels``) for + # downstream consistency. + event_payload["labels"] = labels_payload_for_event _append_event( conn, task_id, "specified", - {"changed_fields": changed_fields} if changed_fields else None, + event_payload, ) # Outside the write_txn above, so we don't nest BEGIN IMMEDIATE — the # ready-promotion pass opens its own IMMEDIATE txn. This runs the same diff --git a/hermes_cli/kanban_specify.py b/hermes_cli/kanban_specify.py index 0d57fbb2504a9..8a89271f63e3c 100644 --- a/hermes_cli/kanban_specify.py +++ b/hermes_cli/kanban_specify.py @@ -7,6 +7,8 @@ * A tightened title (optional — only replaces if the model proposes a materially different one) * A concrete body: goal, proposed approach, acceptance criteria + * An assignee suggestion drawn from the worker profile roster + * A list of short, lowercase, hyphenated routing labels and then flips the task ``triage -> todo`` via ``kanban_db.specify_triage_task``. The dispatcher promotes it to @@ -20,9 +22,20 @@ Keeps the surface area tiny and the failure modes predictable. * The prompt is a short system + user pair. We ask for JSON with - ``{title, body}``; if parsing fails, we fall back to treating the - whole response as the body and leave the title untouched. No - retry loop — one shot, keep cost bounded. + ``{title, body, assignee_suggestion, labels}``; if parsing fails, + we fall back to treating the whole response as the body and leave + the title untouched. No retry loop — one shot, keep cost bounded. + +* The ``assignee_suggestion`` is a *suggestion only*. A downstream + routing engine may override it. The DB write only fills the + assignee column when it was previously empty (see + ``specify_triage_task``). + +* ``labels`` are short routing/cost tags (e.g. ``long-running``, + ``review-only``, ``parallel-fan-out``). A parallel migration is + adding a dedicated ``labels`` column to ``tasks``; until that + lands, the DB layer falls back to recording them in the + ``specified`` event payload so nothing is lost. * Structured output / JSON mode is not requested explicitly so the specifier works on providers that don't implement it. The parse @@ -35,24 +48,39 @@ import logging import os import re -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import Optional from hermes_cli import kanban_db as kb +from hermes_cli import triage_routing logger = logging.getLogger(__name__) +_PROFILE_ROSTER = ( + "h2coder", + "h2architect", + "h2dispatch", + "h2librarian", + "h2reviewer", + "h2research", + "h2simple", + "ruflo-swarm", +) + + _SYSTEM_PROMPT = """You are the Kanban triage specifier for the Hermes Agent board. A user dropped a rough idea into the Triage column. Your job is to turn it into a concrete, actionable task spec that an autonomous worker can pick up and execute without further clarification. -Output a single JSON object with exactly two keys: +Output a single JSON object with exactly four keys: { "title": "", - "body": "" + "body": "", + "assignee_suggestion": "", + "labels": ["", "..."] } The body MUST include these sections, each prefixed with a bold markdown @@ -64,6 +92,28 @@ **Out of scope** — short list of things NOT to touch (omit if nothing obvious; never invent scope creep). +Worker profile roster (pick the best fit for ``assignee_suggestion``): + - h2coder — general code changes, bug fixes, refactors. + - h2architect — system design, multi-module redesigns, RFCs. + - h2dispatch — coordination, scheduling, dispatcher/board mechanics. + - h2librarian — docs, READMEs, comment cleanups, knowledge curation. + - h2reviewer — review-only or audit-only inspection passes. + - h2research — investigation, write-ups, no-code analysis tasks. + - h2simple — small, well-scoped chores a junior worker can finish. + - ruflo-swarm — long-running parallel fan-out across many subtasks. + +If no profile is a clear fit, set ``assignee_suggestion`` to ``null`` — +do NOT guess. A downstream routing engine may override your suggestion. + +Labels are short, lowercase, hyphenated routing/cost tags. Prefer these +when they apply, but free-form tags are fine when none match: + + long-running, autopilot, audit-only, review-only, large, + parallel-fan-out, est-hours-high + +Keep ``labels`` empty (``[]``) rather than inventing tags that don't add +information. 1–5 labels is a healthy range; never exceed 10. + Rules: - Keep the tightened title close in meaning to the original idea — do NOT invent a different project. @@ -90,6 +140,13 @@ class SpecifyOutcome: ok: bool reason: str = "" new_title: Optional[str] = None + # Best-guess profile name proposed by the LLM. ``None`` when the model + # declined to commit (the recommended behaviour when no profile is a + # clear fit). A downstream routing engine may override this. + assignee_suggestion: Optional[str] = None + # Short, lowercase, hyphenated routing/cost tags. Empty list when the + # model produced no usable labels. + labels: list[str] = field(default_factory=list) def _truncate(text: str, limit: int) -> str: @@ -122,6 +179,76 @@ def _extract_json_blob(raw: str) -> Optional[dict]: return val +_LABEL_RE = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$") +# Cap on labels per task — guards against runaway model output. Anything +# past this is silently dropped (we don't error: a noisy label list still +# lets the rest of the spec through). +_MAX_LABELS = 10 +_MAX_LABEL_LEN = 40 + + +def _normalize_label(raw: object) -> Optional[str]: + """Coerce a single label to the canonical ``lowercase-hyphenated`` form. + + Accepts mild deviations (uppercase, surrounding whitespace, snake_case, + spaces) and silently rejects strings that can't be cleaned up. Returning + ``None`` instead of raising keeps a single bad label from poisoning the + whole batch. + """ + if not isinstance(raw, str): + return None + cleaned = raw.strip().lower() + # Convert spaces/underscores to hyphens; collapse repeats. + cleaned = re.sub(r"[\s_]+", "-", cleaned) + cleaned = re.sub(r"-+", "-", cleaned).strip("-") + if not cleaned or len(cleaned) > _MAX_LABEL_LEN: + return None + if not _LABEL_RE.match(cleaned): + return None + return cleaned + + +def _normalize_labels(raw: object) -> list[str]: + """Coerce the model's ``labels`` field to a clean, de-duplicated list. + + Order is preserved (first occurrence wins). Caps at ``_MAX_LABELS``; + silently drops non-string entries and labels that fail the lowercase- + hyphenated regex. + """ + if not isinstance(raw, list): + return [] + seen: set[str] = set() + out: list[str] = [] + for item in raw: + norm = _normalize_label(item) + if norm is None or norm in seen: + continue + seen.add(norm) + out.append(norm) + if len(out) >= _MAX_LABELS: + break + return out + + +def _normalize_assignee_suggestion(raw: object) -> Optional[str]: + """Validate the model's ``assignee_suggestion`` against the roster. + + Returns the canonical profile name when the suggestion matches a known + profile (case-insensitive, whitespace-tolerant), or ``None`` otherwise. + ``None`` is the right default for an off-roster guess — we never want + to write a bogus profile name into the assignee column. + """ + if not isinstance(raw, str): + return None + cleaned = raw.strip().lower() + if not cleaned: + return None + for name in _PROFILE_ROSTER: + if cleaned == name.lower(): + return name + return None + + def _profile_author() -> str: """Mirror of ``hermes_cli.kanban._profile_author``. Kept local to avoid a circular import when kanban.py imports this module.""" @@ -207,10 +334,14 @@ def specify_task( new_title: Optional[str] new_body: Optional[str] + assignee_suggestion: Optional[str] = None + labels: list[str] = [] + final_assignee: Optional[str] = None if parsed is None: # Fall back: treat the whole reply as the body, leave title as-is. # Worst case the user edits afterward — still better than stranding - # the task in triage on a malformed LLM reply. + # the task in triage on a malformed LLM reply. No assignee/labels + # are inferable from prose, so leave them at their defaults. stripped_raw = raw.strip() if not stripped_raw: return SpecifyOutcome( @@ -233,6 +364,20 @@ def specify_task( return SpecifyOutcome( task_id, False, "LLM response missing title and body" ) + assignee_suggestion = _normalize_assignee_suggestion( + parsed.get("assignee_suggestion") + ) + labels = _normalize_labels(parsed.get("labels")) + try: + routing_rules = triage_routing.load_routing_rules() + final_assignee = triage_routing.route( + labels, assignee_suggestion, routing_rules + ) + except (ValueError, OSError) as exc: + logger.warning( + "routing failed (%s); falling back to LLM suggestion", exc + ) + final_assignee = assignee_suggestion with kb.connect() as conn: ok = kb.specify_triage_task( @@ -240,6 +385,8 @@ def specify_task( task_id, title=new_title, body=new_body, + assignee_suggestion=final_assignee, + labels=labels, author=author or _profile_author(), ) if not ok: @@ -248,7 +395,14 @@ def specify_task( return SpecifyOutcome( task_id, False, "task moved out of triage before promotion" ) - return SpecifyOutcome(task_id, True, "specified", new_title=new_title) + return SpecifyOutcome( + task_id, + True, + "specified", + new_title=new_title, + assignee_suggestion=assignee_suggestion, + labels=labels, + ) def list_triage_ids(*, tenant: Optional[str] = None) -> list[str]: diff --git a/hermes_cli/triage_routing.py b/hermes_cli/triage_routing.py new file mode 100644 index 0000000000000..cb62aea17715a --- /dev/null +++ b/hermes_cli/triage_routing.py @@ -0,0 +1,217 @@ +"""Kanban triage routing engine — map card labels to an assignee. + +Phase 1 of the Hermes Kanban triage layer. The LLM specifier emits a set +of labels (and optionally a suggested assignee) for each card; this +module resolves that into the actual assignee a card should be routed to +by walking user-configured routing rules from ``~/.hermes/config.yaml``. + +Design notes +------------ + +* Rules are evaluated in order. The first rule whose required labels + are a **subset** of the card's labels wins. This lets users encode + priority by ordering (e.g. put more-specific rules first). + +* The LLM's suggested assignee is the **fallback** — it's only used + when no user rule matches. Explicit user rules always override the + model's guess. + +* Validation is strict at load time so a typo in config.yaml fails + loudly rather than silently routing every card to ``None``. The + ``ValueError`` messages name the rule index so users can find the + bad entry. + +* Import-clean: no I/O, no global state, no side-effects at import + time. Config is only read when ``load_routing_rules()`` is called. + +Integration +----------- + +The specifier (``kanban_specify.py``, parallel agent) calls +``route(labels, llm_suggested_assignee, load_routing_rules())`` after +extracting labels from the LLM response, and writes the returned +assignee back to the kanban DB when promoting the card to ``todo``. +""" + +from __future__ import annotations + +import logging +from pathlib import Path +from typing import Any, Optional + +import yaml + +logger = logging.getLogger(__name__) + + +__all__ = ["route", "load_routing_rules"] + + +def _validate_rule(rule: Any, index: int) -> tuple[list[str], str]: + """Validate a single routing rule. Returns (labels, assignee). + + Raises ``ValueError`` with a message pointing at ``index`` if the + rule is malformed. + """ + if not isinstance(rule, dict): + raise ValueError( + f"routing rule at index {index} must be a dict, " + f"got {type(rule).__name__}" + ) + + if "labels" not in rule: + raise ValueError( + f"routing rule at index {index} is missing required key 'labels'" + ) + if "assignee" not in rule: + raise ValueError( + f"routing rule at index {index} is missing required key 'assignee'" + ) + + raw_labels = rule["labels"] + if not isinstance(raw_labels, list): + raise ValueError( + f"routing rule at index {index}: 'labels' must be a list, " + f"got {type(raw_labels).__name__}" + ) + for j, label in enumerate(raw_labels): + if not isinstance(label, str): + raise ValueError( + f"routing rule at index {index}: 'labels[{j}]' must be a " + f"string, got {type(label).__name__}" + ) + + assignee = rule["assignee"] + if not isinstance(assignee, str): + raise ValueError( + f"routing rule at index {index}: 'assignee' must be a string, " + f"got {type(assignee).__name__}" + ) + + return list(raw_labels), assignee + + +def route( + labels: list[str], + llm_suggested_assignee: Optional[str], + routing_rules: list[dict], +) -> Optional[str]: + """Resolve a card's assignee from its labels and user routing rules. + + Walk ``routing_rules`` in order. The first rule whose ``labels`` is + a (non-strict) subset of the card's ``labels`` wins and its + ``assignee`` is returned. If no rule matches, ``llm_suggested_assignee`` + is returned (which may itself be ``None``). + + Parameters + ---------- + labels: + The card's labels (as emitted by the LLM specifier). + llm_suggested_assignee: + The LLM's preferred assignee, used as fallback when no user rule + matches. ``None`` if the LLM had no preference. + routing_rules: + Ordered list of ``{"labels": [...], "assignee": "..."}`` dicts, + typically from :func:`load_routing_rules`. + + Returns + ------- + Optional[str] + The resolved assignee, or ``None`` if no rule matched and no + LLM suggestion was provided. + + Raises + ------ + ValueError + If ``routing_rules`` is malformed (not a list, contains a + non-dict, or any rule has the wrong shape). + """ + if not isinstance(routing_rules, list): + raise ValueError( + f"routing_rules must be a list, got {type(routing_rules).__name__}" + ) + + card_labels = set(labels or []) + + for index, rule in enumerate(routing_rules): + rule_labels, assignee = _validate_rule(rule, index) + if set(rule_labels).issubset(card_labels): + return assignee + + return llm_suggested_assignee + + +def _default_config_path() -> Path: + """Resolve the default ``~/.hermes/config.yaml`` path. + + Imported lazily so this module stays import-clean — the + ``hermes_constants`` module touches the filesystem at import time + (computing ``HERMES_HOME``), and we want callers that never invoke + ``load_routing_rules`` to pay nothing. + """ + from hermes_constants import get_hermes_home + + return get_hermes_home() / "config.yaml" + + +def load_routing_rules(config_path: Optional[str] = None) -> list[dict]: + """Load routing rules from ``triage.routing_rules`` in config.yaml. + + Returns an empty list if the config file doesn't exist, can't be + parsed, or has no ``triage.routing_rules`` key. Each rule is + validated as part of the lookup — a malformed rule raises + ``ValueError`` so the user sees the problem at routing time + instead of silently dropping rules. + + Parameters + ---------- + config_path: + Optional override for the config file path. When ``None`` + (the default), uses ``~/.hermes/config.yaml`` resolved via + the standard Hermes config-path helpers. + + Returns + ------- + list[dict] + The list of routing rule dicts, in the order they appear in + config.yaml. Empty list if no rules are configured. + + Raises + ------ + ValueError + If ``triage.routing_rules`` is present but not a list, or if + any individual rule is malformed. + """ + path = Path(config_path) if config_path is not None else _default_config_path() + + try: + with open(path, encoding="utf-8") as f: + raw = yaml.safe_load(f) or {} + except FileNotFoundError: + return [] + except (OSError, yaml.YAMLError) as exc: + logger.warning("triage_routing: failed to read %s: %s", path, exc) + return [] + + if not isinstance(raw, dict): + return [] + + triage_section = raw.get("triage") + if not isinstance(triage_section, dict): + return [] + + rules = triage_section.get("routing_rules") + if rules is None: + return [] + if not isinstance(rules, list): + raise ValueError( + "triage.routing_rules must be a list, " + f"got {type(rules).__name__}" + ) + + # Validate eagerly so callers see a bad config at load time, not + # halfway through a sweep when one specific card hits a bad rule. + for index, rule in enumerate(rules): + _validate_rule(rule, index) + + return list(rules) diff --git a/tests/hermes_cli/test_kanban_labels.py b/tests/hermes_cli/test_kanban_labels.py new file mode 100644 index 0000000000000..c922721d79d21 --- /dev/null +++ b/tests/hermes_cli/test_kanban_labels.py @@ -0,0 +1,239 @@ +"""Tests for the ``tasks.labels`` triage column (Phase 1 of the +Kanban triage layer). + +Covers: + * fresh DBs have the ``labels`` column with the JSON-array default + * legacy DBs missing the column are auto-migrated on ``connect()`` + * ``set_task_labels`` / ``get_task_labels`` roundtrip + * ``set_task_labels`` rejects malformed input +""" + +from __future__ import annotations + +import json +import sqlite3 +from pathlib import Path + +import pytest + +from hermes_cli import kanban_db as kb + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def kanban_home(tmp_path, monkeypatch): + """Isolated HERMES_HOME with an empty kanban DB. + + Mirrors ``tests/hermes_cli/test_kanban_db.py::kanban_home`` so the + label tests inherit the same path-resolution behaviour. + """ + home = tmp_path / ".hermes" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setattr(Path, "home", lambda: tmp_path) + # Wipe the per-process init cache so each test starts from a + # truly cold DB; otherwise a prior test in the same process can + # leave the migration "already run" flag set for tmp_path. + kb._INITIALIZED_PATHS.clear() + kb.init_db() + return home + + +# --------------------------------------------------------------------------- +# Schema: fresh DB +# --------------------------------------------------------------------------- + +def test_fresh_db_has_labels_column(kanban_home): + """A freshly-initialised DB must include the ``labels`` column. + + Guards against silent regressions in ``SCHEMA_SQL``. + """ + with kb.connect() as conn: + cols = {row["name"]: row for row in conn.execute("PRAGMA table_info(tasks)")} + assert "labels" in cols, "labels column missing from fresh DB" + info = cols["labels"] + assert info["type"].upper() == "TEXT" + assert int(info["notnull"]) == 1, "labels must be NOT NULL" + # SQLite quotes string defaults so the literal '[]' shows up as + # "'[]'" in dflt_value. Accept either form for forward compat. + assert info["dflt_value"] in ("'[]'", "[]") + + +def test_fresh_task_has_empty_labels(kanban_home): + """A task created on a fresh DB starts with an empty label list.""" + with kb.connect() as conn: + tid = kb.create_task(conn, title="needs triage") + assert kb.get_task_labels(conn, tid) == [] + + +# --------------------------------------------------------------------------- +# Migration: legacy DB +# --------------------------------------------------------------------------- + +def test_legacy_db_gets_labels_column_on_connect(tmp_path, monkeypatch): + """Manually build a DB without ``labels``, then connect. + + The connect() path must run ``_migrate_add_labels_column`` so the + column appears without the operator having to do anything. + """ + home = tmp_path / ".hermes" + home.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setattr(Path, "home", lambda: tmp_path) + kb._INITIALIZED_PATHS.clear() + + db_path = home / "kanban.db" + # Hand-craft a "legacy" tasks table that has every pre-labels + # column (i.e. the schema as it existed just before this change). + # Anything sparser breaks the index pass in + # ``conn.executescript(SCHEMA_SQL)`` because indexes reference + # columns ``_migrate_add_optional_columns`` has not yet added. + raw = sqlite3.connect(str(db_path), isolation_level=None) + raw.executescript( + """ + CREATE TABLE tasks ( + id TEXT PRIMARY KEY, + title TEXT NOT NULL, + body TEXT, + assignee TEXT, + status TEXT NOT NULL, + priority INTEGER DEFAULT 0, + created_by TEXT, + created_at INTEGER NOT NULL, + started_at INTEGER, + completed_at INTEGER, + workspace_kind TEXT NOT NULL DEFAULT 'scratch', + workspace_path TEXT, + claim_lock TEXT, + claim_expires INTEGER, + tenant TEXT, + result TEXT, + idempotency_key TEXT, + consecutive_failures INTEGER NOT NULL DEFAULT 0, + worker_pid INTEGER, + last_failure_error TEXT, + max_runtime_seconds INTEGER, + last_heartbeat_at INTEGER, + current_run_id INTEGER, + workflow_template_id TEXT, + current_step_key TEXT, + skills TEXT, + max_retries INTEGER + ); + INSERT INTO tasks (id, title, status, created_at) + VALUES ('t_legacy', 'old task', 'ready', 1); + """ + ) + raw.close() + + # Sanity: column really is absent before we touch it. + raw = sqlite3.connect(str(db_path)) + before = {r[1] for r in raw.execute("PRAGMA table_info(tasks)")} + raw.close() + assert "labels" not in before + + # connect() auto-runs migrations. + with kb.connect(db_path) as conn: + after = {row["name"] for row in conn.execute("PRAGMA table_info(tasks)")} + assert "labels" in after, "labels column was not added by migration" + # Pre-existing row should now read back with the default. + row = conn.execute( + "SELECT labels FROM tasks WHERE id = ?", ("t_legacy",) + ).fetchone() + assert row["labels"] == "[]" + assert kb.get_task_labels(conn, "t_legacy") == [] + + +def test_migration_is_idempotent(kanban_home): + """Running the migration twice must not error or duplicate columns.""" + with kb.connect() as conn: + # First run was implicit via init_db; calling it directly should + # short-circuit on the PRAGMA table_info check. + added_again = kb._migrate_add_labels_column(conn) + assert added_again is False + cols = [r["name"] for r in conn.execute("PRAGMA table_info(tasks)")] + # Column only appears once. + assert cols.count("labels") == 1 + + +# --------------------------------------------------------------------------- +# Set / get roundtrip +# --------------------------------------------------------------------------- + +def test_set_and_get_labels_roundtrip(kanban_home): + with kb.connect() as conn: + tid = kb.create_task(conn, title="triage me") + assert kb.set_task_labels(conn, tid, ["bug", "infra"]) is True + assert kb.get_task_labels(conn, tid) == ["bug", "infra"] + # Persists across statements. + stored = conn.execute( + "SELECT labels FROM tasks WHERE id = ?", (tid,) + ).fetchone()["labels"] + assert json.loads(stored) == ["bug", "infra"] + + +def test_set_labels_deduplicates_and_strips(kanban_home): + with kb.connect() as conn: + tid = kb.create_task(conn, title="dup tags") + kb.set_task_labels(conn, tid, ["bug", " bug ", "", "infra", "bug"]) + # Whitespace-only entries dropped, dupes collapsed, order preserved. + assert kb.get_task_labels(conn, tid) == ["bug", "infra"] + + +def test_set_labels_overwrites(kanban_home): + with kb.connect() as conn: + tid = kb.create_task(conn, title="replace me") + kb.set_task_labels(conn, tid, ["old"]) + kb.set_task_labels(conn, tid, ["new", "tags"]) + assert kb.get_task_labels(conn, tid) == ["new", "tags"] + + +def test_set_labels_empty_list_clears(kanban_home): + with kb.connect() as conn: + tid = kb.create_task(conn, title="clear me") + kb.set_task_labels(conn, tid, ["something"]) + kb.set_task_labels(conn, tid, []) + assert kb.get_task_labels(conn, tid) == [] + + +def test_set_labels_unknown_task_returns_false(kanban_home): + with kb.connect() as conn: + assert kb.set_task_labels(conn, "t_does_not_exist", ["x"]) is False + + +def test_get_labels_unknown_task_returns_empty(kanban_home): + with kb.connect() as conn: + assert kb.get_task_labels(conn, "t_ghost") == [] + + +# --------------------------------------------------------------------------- +# Malformed input rejection +# --------------------------------------------------------------------------- + +def test_set_labels_rejects_non_list(kanban_home): + with kb.connect() as conn: + tid = kb.create_task(conn, title="bad input") + with pytest.raises(TypeError, match="list of strings"): + kb.set_task_labels(conn, tid, "bug") # type: ignore[arg-type] + with pytest.raises(TypeError, match="list of strings"): + kb.set_task_labels(conn, tid, {"bug": True}) # type: ignore[arg-type] + with pytest.raises(TypeError, match="list of strings"): + kb.set_task_labels(conn, tid, ("bug",)) # type: ignore[arg-type] + with pytest.raises(TypeError, match="list of strings"): + kb.set_task_labels(conn, tid, None) # type: ignore[arg-type] + + +def test_set_labels_rejects_non_string_elements(kanban_home): + with kb.connect() as conn: + tid = kb.create_task(conn, title="bad items") + with pytest.raises(TypeError, match="only strings"): + kb.set_task_labels(conn, tid, ["ok", 7]) # type: ignore[list-item] + with pytest.raises(TypeError, match="only strings"): + kb.set_task_labels(conn, tid, ["ok", None]) # type: ignore[list-item] + with pytest.raises(TypeError, match="only strings"): + kb.set_task_labels(conn, tid, [["nested"]]) # type: ignore[list-item] + # Row stays untouched after a rejected write. + assert kb.get_task_labels(conn, tid) == [] diff --git a/tests/hermes_cli/test_kanban_specify.py b/tests/hermes_cli/test_kanban_specify.py index dd377001590a2..889aa4365f32f 100644 --- a/tests/hermes_cli/test_kanban_specify.py +++ b/tests/hermes_cli/test_kanban_specify.py @@ -335,3 +335,165 @@ def test_cli_specify_author_passed_through(kanban_home, capsys): with kb.connect() as conn: comments = kb.list_comments(conn, tid) assert comments and comments[0].author == "custom-agent" + + +# --------------------------------------------------------------------------- +# Phase-1 triage layer: assignee_suggestion + labels +# --------------------------------------------------------------------------- + +def test_normalize_label_canonical_form(): + # Lowercase hyphenated strings pass through unchanged. + assert spec._normalize_label("long-running") == "long-running" + # Uppercase / whitespace / underscores normalised to lowercase hyphens. + assert spec._normalize_label("Long_Running") == "long-running" + assert spec._normalize_label(" audit only ") == "audit-only" + assert spec._normalize_label("PARALLEL FAN OUT") == "parallel-fan-out" + # Rejects unsalvageable input. + assert spec._normalize_label("") is None + assert spec._normalize_label(" ") is None + assert spec._normalize_label("has/slashes") is None + assert spec._normalize_label(42) is None + assert spec._normalize_label(None) is None + # Caps overly long labels (over 40 chars). + assert spec._normalize_label("a" * 41) is None + + +def test_normalize_labels_dedupes_and_caps(): + raw = ["long-running", "Long_Running", "audit-only", "review-only"] + # ``Long_Running`` collapses to a duplicate of ``long-running`` — dropped. + assert spec._normalize_labels(raw) == [ + "long-running", "audit-only", "review-only", + ] + # Non-list returns empty list (defensive). + assert spec._normalize_labels("long-running") == [] + assert spec._normalize_labels(None) == [] + # Caps at _MAX_LABELS. + huge = [f"tag-{i}" for i in range(spec._MAX_LABELS + 5)] + assert len(spec._normalize_labels(huge)) == spec._MAX_LABELS + + +def test_normalize_assignee_suggestion_matches_roster(): + # Roster names match case-insensitively. + assert spec._normalize_assignee_suggestion("h2coder") == "h2coder" + assert spec._normalize_assignee_suggestion("H2Coder") == "h2coder" + assert spec._normalize_assignee_suggestion(" ruflo-swarm ") == "ruflo-swarm" + # Off-roster names are rejected (returns None). + assert spec._normalize_assignee_suggestion("random-agent") is None + assert spec._normalize_assignee_suggestion("") is None + assert spec._normalize_assignee_suggestion(None) is None + assert spec._normalize_assignee_suggestion(42) is None + + +def test_specify_task_parses_four_field_output(kanban_home): + """The LLM now emits {title, body, assignee_suggestion, labels} — + all four must be carried through to the outcome and persisted.""" + with kb.connect() as conn: + tid = kb.create_task(conn, title="rough", triage=True) + + content = jsonlib.dumps({ + "title": "Refined title", + "body": "**Goal**\nDo the thing.", + "assignee_suggestion": "h2coder", + "labels": ["long-running", "review-only"], + }) + p, _ = _patch_aux_client(content) + with p: + outcome = spec.specify_task(tid, author="ace") + + assert outcome.ok is True + assert outcome.new_title == "Refined title" + assert outcome.assignee_suggestion == "h2coder" + assert outcome.labels == ["long-running", "review-only"] + + +def test_specify_task_fills_empty_assignee_with_suggestion(kanban_home): + """When the task has no assignee, the LLM suggestion lands in + the assignee column.""" + with kb.connect() as conn: + tid = kb.create_task(conn, title="rough", triage=True) + # Sanity check — no pre-existing assignee. + assert kb.get_task(conn, tid).assignee in (None, "") + + content = jsonlib.dumps({ + "title": "Refined", + "body": "spec body", + "assignee_suggestion": "h2librarian", + "labels": [], + }) + p, _ = _patch_aux_client(content) + with p: + outcome = spec.specify_task(tid) + + assert outcome.ok is True + with kb.connect() as conn: + task = kb.get_task(conn, tid) + assert task.assignee == "h2librarian" + + +def test_specify_task_preserves_existing_assignee(kanban_home): + """A pre-existing assignee MUST NOT be overwritten by the LLM + suggestion — humans / routing engines win.""" + with kb.connect() as conn: + tid = kb.create_task( + conn, + title="rough", + triage=True, + assignee="h2architect", # already set by a human / earlier pass + ) + + content = jsonlib.dumps({ + "title": "Refined", + "body": "spec body", + "assignee_suggestion": "h2coder", # different suggestion — should be ignored + "labels": [], + }) + p, _ = _patch_aux_client(content) + with p: + outcome = spec.specify_task(tid) + + assert outcome.ok is True + # The outcome still reports what the LLM suggested (so callers / + # downstream routing can see it) — but the DB row keeps the original. + assert outcome.assignee_suggestion == "h2coder" + with kb.connect() as conn: + task = kb.get_task(conn, tid) + assert task.assignee == "h2architect" + + +def test_specify_task_off_roster_suggestion_is_dropped(kanban_home): + """If the model invents a profile name not on the roster, we + refuse to write it — better empty than wrong.""" + with kb.connect() as conn: + tid = kb.create_task(conn, title="rough", triage=True) + + content = jsonlib.dumps({ + "title": "Refined", + "body": "spec body", + "assignee_suggestion": "totally-made-up-profile", + "labels": ["long-running"], + }) + p, _ = _patch_aux_client(content) + with p: + outcome = spec.specify_task(tid) + + assert outcome.ok is True + assert outcome.assignee_suggestion is None + with kb.connect() as conn: + task = kb.get_task(conn, tid) + assert task.assignee in (None, "") + + +def test_specify_task_falls_back_body_only_clears_assignee_and_labels(kanban_home): + """When JSON parsing fails entirely, we still promote the task but + we don't invent an assignee suggestion or labels from prose.""" + with kb.connect() as conn: + tid = kb.create_task(conn, title="keep title", triage=True) + + content = "no JSON here — just prose." + p, _ = _patch_aux_client(content) + with p: + outcome = spec.specify_task(tid) + + assert outcome.ok is True + assert outcome.assignee_suggestion is None + assert outcome.labels == [] diff --git a/tests/hermes_cli/test_kanban_specify_db.py b/tests/hermes_cli/test_kanban_specify_db.py index 4128c8c522ac3..46a3145a7189c 100644 --- a/tests/hermes_cli/test_kanban_specify_db.py +++ b/tests/hermes_cli/test_kanban_specify_db.py @@ -182,3 +182,85 @@ def test_specify_second_call_noop_false(kanban_home): assert kb.specify_triage_task(conn, tid, body="spec") is True with kb.connect() as conn: assert kb.specify_triage_task(conn, tid, body="spec again") is False + + +# --------------------------------------------------------------------------- +# Phase-1 triage layer: assignee_suggestion + labels on the DB write +# --------------------------------------------------------------------------- + +def test_specify_writes_assignee_suggestion_when_empty(kanban_home): + with kb.connect() as conn: + tid = _create_triage(conn, title="rough") + assert kb.get_task(conn, tid).assignee in (None, "") + with kb.connect() as conn: + ok = kb.specify_triage_task( + conn, + tid, + body="spec body", + assignee_suggestion="h2coder", + ) + assert ok is True + with kb.connect() as conn: + task = kb.get_task(conn, tid) + assert task.assignee == "h2coder" + + +def test_specify_does_not_overwrite_existing_assignee(kanban_home): + with kb.connect() as conn: + tid = _create_triage(conn, title="rough", assignee="h2architect") + with kb.connect() as conn: + ok = kb.specify_triage_task( + conn, + tid, + body="spec body", + assignee_suggestion="h2coder", # ignored — already set + ) + assert ok is True + with kb.connect() as conn: + task = kb.get_task(conn, tid) + assert task.assignee == "h2architect" + + +def test_specify_labels_uses_labels_column(kanban_home): + """The schema includes a ``labels`` column (added by the migration in + kanban_db._migrate_add_labels_column). Labels go in the column; the + event payload reports the change in ``changed_fields`` but does not + duplicate the list.""" + with kb.connect() as conn: + tid = _create_triage(conn, title="rough") + with kb.connect() as conn: + ok = kb.specify_triage_task( + conn, + tid, + body="spec body", + labels=["large", "parallel-fan-out"], + author="ace", + ) + assert ok is True + # Labels written to the column. + with kb.connect() as conn: + row = conn.execute( + "SELECT labels FROM tasks WHERE id = ?", (tid,) + ).fetchone() + events = kb.list_events(conn, tid) + import json as _json + assert _json.loads(row["labels"]) == ["large", "parallel-fan-out"] + # Event payload reports labels in changed_fields but does NOT also + # carry the labels list (no need — they're in the column). + spec_ev = next(e for e in events if e.kind == "specified") + assert "labels" in (spec_ev.payload.get("changed_fields") or []) + assert "labels" not in (spec_ev.payload or {}) + + +def test_specify_assignee_suggestion_none_does_nothing(kanban_home): + """Passing ``assignee_suggestion=None`` is a no-op on assignee.""" + with kb.connect() as conn: + tid = _create_triage(conn, title="rough") + with kb.connect() as conn: + ok = kb.specify_triage_task( + conn, tid, body="spec body", assignee_suggestion=None + ) + assert ok is True + with kb.connect() as conn: + task = kb.get_task(conn, tid) + assert task.assignee in (None, "") diff --git a/tests/hermes_cli/test_triage_routing.py b/tests/hermes_cli/test_triage_routing.py new file mode 100644 index 0000000000000..636c0ff2f7984 --- /dev/null +++ b/tests/hermes_cli/test_triage_routing.py @@ -0,0 +1,283 @@ +"""Tests for the triage routing engine. + +Covers first-match-wins semantics, subset matching, LLM fallback, +malformed-rule validation, and the ``load_routing_rules`` config loader. +No network or DB — pure function tests plus YAML file I/O against a +``tmp_path``. +""" + +from __future__ import annotations + +import textwrap + +import pytest +import yaml + +from hermes_cli import triage_routing as tr + + +# --------------------------------------------------------------------------- +# route() — matching semantics +# --------------------------------------------------------------------------- + +class TestRouteMatching: + """Core matching behavior for ``route()``.""" + + def test_first_match_wins(self): + """Earlier rules must take precedence over later matching ones.""" + rules = [ + {"labels": ["bug"], "assignee": "alice"}, + {"labels": ["bug"], "assignee": "bob"}, + ] + assert tr.route(["bug"], None, rules) == "alice" + + def test_first_match_wins_with_different_label_sets(self): + """First rule whose labels are subset of card's labels wins, + even when a later, more-specific rule also matches.""" + rules = [ + {"labels": ["bug"], "assignee": "first"}, + {"labels": ["bug", "p0"], "assignee": "second"}, + ] + assert tr.route(["bug", "p0"], None, rules) == "first" + + def test_subset_match_single_label(self): + """A rule requiring one label matches a card that has that label + among others.""" + rules = [{"labels": ["security"], "assignee": "sec-team"}] + assert tr.route(["security", "bug", "p1"], None, rules) == "sec-team" + + def test_subset_match_multiple_required_labels(self): + """A rule with multiple labels requires ALL of them on the card.""" + rules = [ + {"labels": ["bug", "p0"], "assignee": "oncall"}, + ] + assert tr.route(["bug", "p0", "frontend"], None, rules) == "oncall" + + def test_no_match_when_rule_label_missing(self): + """A rule whose labels are NOT a subset of the card's labels + must not match.""" + rules = [{"labels": ["bug", "p0"], "assignee": "oncall"}] + # Card has 'bug' but not 'p0' — rule should not fire. + assert tr.route(["bug", "p1"], None, rules) is None + + def test_empty_rule_labels_matches_any_card(self): + """A rule with empty label set is vacuously a subset of any + card's labels — matches everything (acts as a catch-all).""" + rules = [{"labels": [], "assignee": "catch-all"}] + assert tr.route(["whatever"], None, rules) == "catch-all" + assert tr.route([], None, rules) == "catch-all" + + +# --------------------------------------------------------------------------- +# route() — fallback to LLM suggestion +# --------------------------------------------------------------------------- + +class TestRouteFallback: + """LLM suggestion is used only when no rule matches.""" + + def test_fallback_to_llm_when_no_rule_matches(self): + rules = [{"labels": ["security"], "assignee": "sec-team"}] + assert tr.route(["bug"], "llm-pick", rules) == "llm-pick" + + def test_fallback_to_llm_when_no_rules_configured(self): + assert tr.route(["anything"], "model-suggestion", []) == "model-suggestion" + + def test_returns_none_when_no_rule_and_no_suggestion(self): + """The None/None case — used when the LLM had no preference + either, and the card should fall through to manual triage.""" + assert tr.route(["bug"], None, []) is None + assert tr.route(["bug"], None, [{"labels": ["other"], "assignee": "x"}]) is None + + def test_rule_match_overrides_llm_suggestion(self): + """An explicit user rule must trump the LLM's guess.""" + rules = [{"labels": ["bug"], "assignee": "user-pick"}] + assert tr.route(["bug"], "llm-pick", rules) == "user-pick" + + def test_empty_labels_falls_through_to_llm(self): + """A card with no labels matches no rule with required labels + and should fall back to the LLM suggestion.""" + rules = [{"labels": ["bug"], "assignee": "buggy"}] + assert tr.route([], "llm-pick", rules) == "llm-pick" + + +# --------------------------------------------------------------------------- +# route() — validation +# --------------------------------------------------------------------------- + +class TestRouteValidation: + """``route()`` must reject malformed rule lists with a clear error.""" + + def test_malformed_rule_not_a_dict_raises(self): + rules = ["not-a-dict"] + with pytest.raises(ValueError, match="must be a dict"): + tr.route(["bug"], None, rules) + + def test_malformed_rule_missing_labels_raises(self): + rules = [{"assignee": "bob"}] + with pytest.raises(ValueError, match="missing required key 'labels'"): + tr.route(["bug"], None, rules) + + def test_malformed_rule_missing_assignee_raises(self): + rules = [{"labels": ["bug"]}] + with pytest.raises(ValueError, match="missing required key 'assignee'"): + tr.route(["bug"], None, rules) + + def test_malformed_rule_labels_not_list_raises(self): + rules = [{"labels": "bug", "assignee": "bob"}] + with pytest.raises(ValueError, match="'labels' must be a list"): + tr.route(["bug"], None, rules) + + def test_malformed_rule_label_element_not_string_raises(self): + rules = [{"labels": ["bug", 42], "assignee": "bob"}] + with pytest.raises(ValueError, match="must be a string"): + tr.route(["bug"], None, rules) + + def test_malformed_rule_assignee_not_string_raises(self): + rules = [{"labels": ["bug"], "assignee": 123}] + with pytest.raises(ValueError, match="'assignee' must be a string"): + tr.route(["bug"], None, rules) + + def test_routing_rules_not_a_list_raises(self): + with pytest.raises(ValueError, match="routing_rules must be a list"): + tr.route(["bug"], None, "not-a-list") # type: ignore[arg-type] + + def test_error_message_names_rule_index(self): + """The index of the offending rule is in the error message so + users can find it in config.yaml.""" + rules = [ + {"labels": ["a"], "assignee": "x"}, + {"labels": ["b"], "assignee": "y"}, + "broken", + ] + # The first rule matches "a" so we never reach the broken one. + # Use card labels that miss the first two rules so we walk to + # the malformed third. + with pytest.raises(ValueError, match="at index 2"): + tr.route(["c"], None, rules) + + +# --------------------------------------------------------------------------- +# load_routing_rules() +# --------------------------------------------------------------------------- + +class TestLoadRoutingRules: + """``load_routing_rules`` reads rules from config.yaml, tolerating + missing files and missing config keys.""" + + def test_missing_config_file_returns_empty(self, tmp_path): + missing = tmp_path / "does-not-exist.yaml" + assert tr.load_routing_rules(str(missing)) == [] + + def test_missing_triage_section_returns_empty(self, tmp_path): + cfg = tmp_path / "config.yaml" + cfg.write_text("model:\n default: gpt-4\n") + assert tr.load_routing_rules(str(cfg)) == [] + + def test_missing_routing_rules_key_returns_empty(self, tmp_path): + cfg = tmp_path / "config.yaml" + cfg.write_text(textwrap.dedent("""\ + triage: + other_setting: foo + """)) + assert tr.load_routing_rules(str(cfg)) == [] + + def test_empty_routing_rules_returns_empty(self, tmp_path): + cfg = tmp_path / "config.yaml" + cfg.write_text(textwrap.dedent("""\ + triage: + routing_rules: [] + """)) + assert tr.load_routing_rules(str(cfg)) == [] + + def test_loads_rules_in_order(self, tmp_path): + cfg = tmp_path / "config.yaml" + payload = { + "triage": { + "routing_rules": [ + {"labels": ["security"], "assignee": "sec-team"}, + {"labels": ["bug", "p0"], "assignee": "oncall"}, + {"labels": ["docs"], "assignee": "writer"}, + ] + } + } + cfg.write_text(yaml.safe_dump(payload)) + + rules = tr.load_routing_rules(str(cfg)) + assert len(rules) == 3 + assert rules[0] == {"labels": ["security"], "assignee": "sec-team"} + assert rules[1] == {"labels": ["bug", "p0"], "assignee": "oncall"} + assert rules[2] == {"labels": ["docs"], "assignee": "writer"} + + def test_loaded_rules_drive_route(self, tmp_path): + """End-to-end: rules loaded from disk feed straight into route().""" + cfg = tmp_path / "config.yaml" + payload = { + "triage": { + "routing_rules": [ + {"labels": ["security"], "assignee": "sec-team"}, + {"labels": ["bug"], "assignee": "bug-team"}, + ] + } + } + cfg.write_text(yaml.safe_dump(payload)) + + rules = tr.load_routing_rules(str(cfg)) + assert tr.route(["security", "bug"], None, rules) == "sec-team" + assert tr.route(["bug"], None, rules) == "bug-team" + assert tr.route(["other"], "fallback", rules) == "fallback" + + def test_malformed_routing_rules_not_list_raises(self, tmp_path): + cfg = tmp_path / "config.yaml" + cfg.write_text(textwrap.dedent("""\ + triage: + routing_rules: + labels: [bug] + assignee: bob + """)) + with pytest.raises(ValueError, match="routing_rules must be a list"): + tr.load_routing_rules(str(cfg)) + + def test_malformed_individual_rule_raises_at_load(self, tmp_path): + """A broken rule in the YAML should fail load immediately, with + the index in the message.""" + cfg = tmp_path / "config.yaml" + payload = { + "triage": { + "routing_rules": [ + {"labels": ["ok"], "assignee": "fine"}, + {"labels": "not-a-list", "assignee": "bob"}, + ] + } + } + cfg.write_text(yaml.safe_dump(payload)) + + with pytest.raises(ValueError, match="at index 1"): + tr.load_routing_rules(str(cfg)) + + def test_unparseable_yaml_returns_empty(self, tmp_path): + """Garbled YAML should NOT crash routing — we surface a warning + and act as if no rules were configured.""" + cfg = tmp_path / "config.yaml" + cfg.write_text(":\n invalid: : :\n\t- bad") + assert tr.load_routing_rules(str(cfg)) == [] + + def test_non_dict_root_returns_empty(self, tmp_path): + """A YAML file whose root is a list (instead of a mapping) + should be treated as 'no config'.""" + cfg = tmp_path / "config.yaml" + cfg.write_text("- just\n- a\n- list\n") + assert tr.load_routing_rules(str(cfg)) == [] + + def test_default_path_via_hermes_home(self, tmp_path, monkeypatch): + """When ``config_path`` is None, the module reads from + ``$HERMES_HOME/config.yaml``.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + (tmp_path / "config.yaml").write_text(yaml.safe_dump({ + "triage": { + "routing_rules": [ + {"labels": ["x"], "assignee": "y"} + ] + } + })) + rules = tr.load_routing_rules() + assert rules == [{"labels": ["x"], "assignee": "y"}]