feat(brainstormer): axis label namespace + axis-aware status/dispatch filter (#126)

docs/axis-labels.md

@@ -0,0 +1,89 @@
+# Axis labels (`axis:<slug>`)
+
+> Introduced in [#126](https://github.com/khalidx/forge-loop/issues/126).
+> Plumbing for focused-sprint mode: group / filter the loop's work by
+> topical concern without changing how issues are queued.
+
+## What it is
+
+A free-form label namespace, `axis:<slug>`, applied to GitHub issues
+by the brainstormer / PO / operators. The slug is any lowercase string
+(e.g. `dispatch`, `cli`, `observability`, `docs`). Multiple `axis:*`
+labels on the same issue are fine; they stack.
+
+There is **no enum, no registry, and no validation**. If you want to
+spin up a new axis tonight, just label an issue.
+
+## Why
+
+Operators staring at `forge-loop status` saw a flat list of
+`loop:ready` issues with no way to tell whether the backlog leaned
+toward dispatcher work, CLI work, or observability work. Likewise,
+`forge-loop run` greedily grabbed whatever was ready, so an operator
+who wanted to grind one area for a sprint had no knob.
+
+## Surface
+
+### Status grouping
+
+```
+$ forge-loop status
+... (existing rows)
+axes  dispatch (6)  #12, #14, #18, #19, #22, #27
+      cli      (5)  #11, #13, #21, #24, #26
+      unaligned(3)  #9, #17, #28
+warning  3 open issue(s) carry no axis:* label
+```
+
+JSON shape (`forge-loop status --json`):
+
+```json
+{
+  "queue_depth": 14,
+  "axes": {
+    "dispatch": [{"number": 12, "title": "..."}, ...],
+    "cli":      [{"number": 11, "title": "..."}, ...],
+    "unaligned": [{"number": 9,  "title": "..."}, ...]
+  },
+  "unaligned_count": 3,
+  "axis_filter": []
+}
+```
+
+Narrow the view with `--axis`:
+
+```
+$ forge-loop status --axis dispatch --json
+{ ..., "axes": { "dispatch": [...] }, "axis_filter": ["dispatch"] }
+```
+
+### Dispatch filter
+
+```
+$ forge-loop run --axis dispatch                # only axis:dispatch issues
+$ forge-loop run --axis dispatch --axis cli     # union
+$ forge-loop run                                # unchanged behaviour (no filter)
+```
+
+The runner logs the active filter at startup and emits
+`axis_filter_active` events per tick so a focused-sprint launch is
+auditable from `forge-loop events`. If no ready issues match, the tick
+exits cleanly (the loop just idles until a matching issue lands).
+
+## Edge cases
+
+| Case | Behaviour |
+| ---- | --------- |
+| Mixed case label (`Axis:Dispatch`) | Normalised to lowercase before matching. |
+| Empty slug label (`axis:`) | Treated as unaligned; logged, no crash. |
+| Issue carries multiple `axis:*` labels | Appears under each bucket; counted once in `unaligned_count`. |
+| `--axis foo` matches zero issues | Tick idles cleanly with an `axis_filter_empty` event. Exit code 0. |
+| `--axis` not passed | Behaviour is byte-identical to the pre-#126 loop. |
+
+## Out of scope
+
+- Automatic axis assignment from issue text (separate brainstormer
+  enhancement).
+- Closed registry / enum of allowed axes.
+- Persisting the filter into config or state — `--axis` is a
+  per-invocation knob.

src/forge_loop/axis.py

@@ -0,0 +1,140 @@
+"""Axis label namespace helpers (issue #126).
+
+The brainstormer/PO emits issues across many concerns: CLI, runner,
+dispatch, observability, etc. Operators want to focus a sprint on a
+single concern — but a flat ``loop:ready`` queue gives them no way to
+group or filter. Issue #126 introduces the ``axis:<slug>`` label
+namespace as plumbing:
+
+* Any label whose name (lowercased) starts with ``axis:`` is parsed
+  into a free-form axis slug. No registry, no validation, no enum.
+* ``forge-loop status`` groups open ready-issues by axis; issues with
+  no axis label go into the ``unaligned`` bucket and a soft warning
+  is surfaced.
+* ``forge-loop run --axis <name>`` (repeatable) narrows dispatch to
+  issues carrying any of those axis labels. Omitting the flag is a
+  no-op — behaviour is byte-identical to the pre-#126 loop.
+
+The helpers in this module are intentionally pure: they read labels
+off in-memory dicts, normalise case, and never touch GitHub. Callers
+fetch issues; we just bucket them.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any, Iterable
+
+
+AXIS_PREFIX = "axis:"
+UNALIGNED_BUCKET = "unaligned"
+AXIS_FILTER_ENV = "LOOP_AXIS_FILTER"
+
+
+def extract_axes(labels: Iterable[Any]) -> set[str]:
+    """Pull the ``axis:*`` slugs off a label list.
+
+    Accepts either raw strings (``"axis:dispatch"``) or dicts shaped
+    like the ``gh issue list --json labels`` payload (``{"name": ...}``).
+    Comparison is case-insensitive: ``Axis:Dispatch`` normalises to
+    ``dispatch``.
+
+    Empty slugs (the literal label ``axis:`` with nothing after the
+    colon) are dropped — the acceptance criteria call this out as
+    "treated as unaligned".
+    """
+    out: set[str] = set()
+    for raw in labels or []:
+        if isinstance(raw, dict):
+            name = str(raw.get("name") or "")
+        else:
+            name = str(raw or "")
+        low = name.strip().lower()
+        if not low.startswith(AXIS_PREFIX):
+            continue
+        slug = low[len(AXIS_PREFIX):].strip()
+        if not slug:
+            continue
+        out.add(slug)
+    return out
+
+
+def matches_axes(labels: Iterable[Any], wanted: Iterable[str]) -> bool:
+    """True iff the issue's axes intersect ``wanted``.
+
+    Used by the dispatcher. ``wanted`` empty means "no filter set" —
+    callers handle that branch separately (this function would return
+    False for empty intersection, which is the wrong answer).
+    """
+    want = {w.lower() for w in wanted if w}
+    if not want:
+        return True
+    return bool(extract_axes(labels) & want)
+
+
+def group_by_axis(
+    issues: list[dict[str, Any]],
+) -> tuple[dict[str, list[dict[str, Any]]], int]:
+    """Bucket issues by axis label.
+
+    Returns ``(buckets, unaligned_count)`` where ``buckets`` maps an
+    axis slug -> list of issues carrying that label, plus a special
+    ``"unaligned"`` key for issues with zero axis labels. An issue
+    carrying multiple axis labels appears under each bucket. The
+    ``unaligned_count`` reflects unique issues, not the sum of bucket
+    sizes (an issue can be in two axes but only counts once).
+    """
+    buckets: dict[str, list[dict[str, Any]]] = {}
+    unaligned = 0
+    for issue in issues:
+        axes = extract_axes(issue.get("labels") or [])
+        if not axes:
+            buckets.setdefault(UNALIGNED_BUCKET, []).append(issue)
+            unaligned += 1
+            continue
+        for ax in sorted(axes):
+            buckets.setdefault(ax, []).append(issue)
+    return buckets, unaligned
+
+
+def parse_filter_env(env: str | None = None) -> list[str]:
+    """Read ``LOOP_AXIS_FILTER`` (comma-separated) -> sorted unique slugs.
+
+    The CLI sets this env var when ``--axis`` is passed; the tick loop
+    consumes it. Empty / missing var -> empty list (= no filter).
+    """
+    raw = env if env is not None else os.environ.get(AXIS_FILTER_ENV, "")
+    if not raw:
+        return []
+    seen: list[str] = []
+    for chunk in raw.split(","):
+        c = chunk.strip().lower()
+        if c and c not in seen:
+            seen.append(c)
+    return seen
+
+
+def filter_issues_by_axes(
+    issues: list[dict[str, Any]],
+    wanted: list[str],
+) -> list[dict[str, Any]]:
+    """Return issues whose axis labels intersect ``wanted``.
+
+    Empty ``wanted`` -> returns input unchanged (preserves today's
+    behaviour exactly, per the regression-guard acceptance criterion).
+    """
+    if not wanted:
+        return issues
+    return [i for i in issues if matches_axes(i.get("labels") or [], wanted)]
+
+
+__all__ = [
+    "AXIS_FILTER_ENV",
+    "AXIS_PREFIX",
+    "UNALIGNED_BUCKET",
+    "extract_axes",
+    "filter_issues_by_axes",
+    "group_by_axis",
+    "matches_axes",
+    "parse_filter_env",
+]

src/forge_loop/cli.py

@@ -102,6 +102,20 @@ def _cmd_run(args: SimpleNamespace) -> int:
     if queue_url:
         os.environ["LOOP_QUEUE_URL"] = queue_url
 
+    # Issue #126 — axis-aware dispatch filter. The CLI accepts ``--axis``
+    # repeatedly; we serialise to a comma-separated env var so the tick
+    # loop (which is a separate function in another module) can read it
+    # without us having to thread a new kwarg through ``run_loop`` and
+    # every adjacent caller. Empty list -> env var stays unset, and the
+    # dispatcher takes the pre-#126 fast path verbatim.
+    from forge_loop.axis import AXIS_FILTER_ENV
+
+    axes = [a.strip().lower() for a in (getattr(args, "axis", None) or []) if a and a.strip()]
+    if axes:
+        os.environ[AXIS_FILTER_ENV] = ",".join(axes)
+    else:
+        os.environ.pop(AXIS_FILTER_ENV, None)
+
     orch = getattr(args, "orchestrator", "sync")
     if orch == "async":
         from forge_loop.runner import run_async as run_async_loop
@@ -324,7 +338,11 @@ def _cmd_status(args: SimpleNamespace) -> int:
             except json.JSONDecodeError:
                 pass
 
+    # Issue #126 — fetch labels + title alongside number so we can group
+    # the open ready-queue by ``axis:*`` label below. Cheap: same call,
+    # one additional JSON field.
     queue_depth = 0
+    ready_issues: list[dict[str, Any]] = []
     try:
         r = subprocess.run(
             [
@@ -337,18 +355,41 @@ def _cmd_status(args: SimpleNamespace) -> int:
                 cfg.labels.ready,
                 "--state",
                 "open",
+                "--limit",
+                "200",
                 "--json",
-                "number",
+                "number,title,labels",
             ],
             capture_output=True,
             text=True,
             timeout=15,
         )
         if r.returncode == 0:
-            queue_depth = len(json.loads(r.stdout or "[]"))
+            ready_issues = json.loads(r.stdout or "[]")
+            queue_depth = len(ready_issues)
     except (subprocess.TimeoutExpired, FileNotFoundError, json.JSONDecodeError):
         queue_depth = -1
 
+    # Group the open ready-queue by axis. Axis filter (--axis) narrows
+    # the bucketed view to just the requested slugs — this is the
+    # "sanity check before running" surface called out in the spec.
+    from forge_loop.axis import UNALIGNED_BUCKET, group_by_axis
+
+    axis_filter = [a.strip().lower() for a in (getattr(args, "axis", None) or []) if a and a.strip()]
+    grouped_all, unaligned_count = group_by_axis(ready_issues)
+    if axis_filter:
+        axes_view: dict[str, list[dict[str, Any]]] = {
+            k: v for k, v in grouped_all.items() if k in set(axis_filter)
+        }
+    else:
+        axes_view = grouped_all
+    # Render shape: {axis_slug: [{number, title}], ...} (drop the heavy
+    # ``labels`` blob from the per-issue payload).
+    axes_payload: dict[str, list[dict[str, Any]]] = {
+        k: [{"number": i.get("number"), "title": i.get("title", "")} for i in v]
+        for k, v in axes_view.items()
+    }
+
     payload: dict[str, Any] = {
         "pid": pid_text or None,
         "pid_alive": pid_alive,
@@ -361,6 +402,9 @@ def _cmd_status(args: SimpleNamespace) -> int:
         "last_failure": last_failure,
         "last_events": last_5_events,
         "events_file": str(cfg.events_file),
+        "axes": axes_payload,
+        "unaligned_count": unaligned_count,
+        "axis_filter": axis_filter,
     }
 
     if getattr(args, "json", False):
@@ -408,6 +452,28 @@ def _cmd_status(args: SimpleNamespace) -> int:
         table.add_row("last 5 events", joined)
     table.add_row("events", str(cfg.events_file))
 
+    # Issue #126 — axis breakdown. Render one row per axis (sorted for
+    # deterministic output) showing the issue count, plus a yellow
+    # "unaligned" warning iff any open ready-issue has no axis label.
+    if axis_filter:
+        table.add_row("axis filter", ", ".join(sorted(set(axis_filter))))
+    if axes_view:
+        axis_lines = Text()
+        for ax in sorted(k for k in axes_view if k != UNALIGNED_BUCKET):
+            nums = ", ".join(f"#{i.get('number')}" for i in axes_view[ax])
+            axis_lines.append(f"  {ax}", style="cyan")
+            axis_lines.append(f" ({len(axes_view[ax])})  {nums}\n")
+        if UNALIGNED_BUCKET in axes_view:
+            unaligned_nums = ", ".join(f"#{i.get('number')}" for i in axes_view[UNALIGNED_BUCKET])
+            axis_lines.append(f"  {UNALIGNED_BUCKET}", style="yellow")
+            axis_lines.append(f" ({len(axes_view[UNALIGNED_BUCKET])})  {unaligned_nums}\n")
+        table.add_row("axes", axis_lines)
+    if unaligned_count > 0 and not axis_filter:
+        table.add_row(
+            "[yellow]warning[/yellow]",
+            f"[yellow]{unaligned_count} open issue(s) carry no axis:* label[/yellow]",
+        )
+
     console.print(Panel(table, title="[bold]forge-loop status[/bold]", title_align="left"))
     return 0
 
@@ -1163,18 +1229,32 @@ def cmd_run(
         "--queue",
         help="Queue backend URL. Default in-memory; sqlite:///path for durable.",
     ),
+    axis: list[str] = typer.Option(
+        [],
+        "--axis",
+        help=(
+            "Narrow dispatch to issues carrying ``axis:<name>`` labels. "
+            "Repeatable; values are unioned. Omit to preserve pre-#126 "
+            "behaviour (no filter)."
+        ),
+    ),
 ) -> None:
     if orchestrator not in {"sync", "async"}:
         typer.echo(f"run: invalid --orchestrator {orchestrator!r}", err=True)
         raise typer.Exit(code=2)
-    _exit(_cmd_run(SimpleNamespace(orchestrator=orchestrator, queue=queue)))
+    _exit(_cmd_run(SimpleNamespace(orchestrator=orchestrator, queue=queue, axis=axis)))
 
 
 @app.command("status", help="Operator-facing health surface.")
 def cmd_status(
     json_: bool = typer.Option(False, "--json", help="Emit raw JSON for scripts."),
+    axis: list[str] = typer.Option(
+        [],
+        "--axis",
+        help="Narrow the axis-grouped view to these slugs (repeatable).",
+    ),
 ) -> None:
-    _exit(_cmd_status(SimpleNamespace(json=json_)))
+    _exit(_cmd_status(SimpleNamespace(json=json_, axis=axis)))
 
 
 @app.command("doctor", help="One-shot health check (config-independent checks still run).")